| 292 | return true; |
| 293 | } |
| 294 | void ScannerImpl::PrepareNotifyCell(tera::Table* table, const std::string& rowkey, |
| 295 | const std::set<Column>& observe_columns, |
| 296 | const std::vector<Column>& notify_columns, |
| 297 | std::shared_ptr<AutoRowUnlocker> unlocker, |
| 298 | std::vector<std::shared_ptr<NotifyCell>>* notify_cells) { |
| 299 | std::shared_ptr<std::map<std::string, TableObserveInfo>> table_observe_info_read_copy; |
| 300 | { |
| 301 | MutexLock locker(&table_mutex_); |
| 302 | // shared_ptr ref +1 |
| 303 | table_observe_info_read_copy = table_observe_info_; |
| 304 | } |
| 305 | |
| 306 | for (auto notify_column = notify_columns.begin(); notify_column != notify_columns.end(); |
| 307 | ++notify_column) { |
| 308 | if (observe_columns.find(*notify_column) == observe_columns.end()) { |
| 309 | LOG(WARNING) << "miss observed column, table_name" << table->GetName() |
| 310 | << " cf=" << notify_column->family << " qu=" << notify_column->qualifier; |
| 311 | continue; |
| 312 | } |
| 313 | std::map<Column, std::set<Observer*>>& observe_columns = |
| 314 | (*table_observe_info_read_copy)[table->GetName()].observe_columns; |
| 315 | |
| 316 | TransactionType type = (*table_observe_info_read_copy)[table->GetName()].type; |
| 317 | |
| 318 | for (auto observer = observe_columns[*notify_column].begin(); |
| 319 | observer != observe_columns[*notify_column].end(); ++observer) { |
| 320 | semaphore_.Acquire(); |
| 321 | std::shared_ptr<NotifyCell> notify_cell(new NotifyCell(semaphore_)); |
| 322 | switch (type) { |
| 323 | case kGlobalTransaction: |
| 324 | notify_cell->notify_transaction.reset(tera_client_->NewGlobalTransaction()); |
| 325 | if (!notify_cell->notify_transaction) { |
| 326 | LOG(ERROR) << "NewGlobalTransaction failed. Notify cell ignored. table: " |
| 327 | << table->GetName() << " row: " << rowkey |
| 328 | << " family: " << notify_column->family |
| 329 | << " qualifier: " << notify_column->qualifier; |
| 330 | continue; |
| 331 | } |
| 332 | break; |
| 333 | case kSingleRowTransaction: |
| 334 | notify_cell->notify_transaction.reset(table->StartRowTransaction(rowkey)); |
| 335 | if (!notify_cell->notify_transaction) { |
| 336 | LOG(ERROR) << "StartRowTransaction failed. Notify cell ignored. table: " |
| 337 | << table->GetName() << " row: " << rowkey |
| 338 | << " family: " << notify_column->family |
| 339 | << " qualifier: " << notify_column->qualifier; |
| 340 | continue; |
| 341 | } |
| 342 | break; |
| 343 | default: |
| 344 | break; |
| 345 | } |
| 346 | |
| 347 | notify_cell->table = table; |
| 348 | notify_cell->row = rowkey; |
| 349 | notify_cell->observed_column = *notify_column; |
| 350 | notify_cell->unlocker = unlocker; |
| 351 | notify_cell->observer = *observer; |