| 612 | } |
| 613 | |
| 614 | void ScannerImpl::ValidateCellValue(RowReader* value_reader) { |
| 615 | std::unique_ptr<NotificationContext> context((NotificationContext*)(value_reader->GetContext())); |
| 616 | std::shared_ptr<NotifyCell> notify_cell = context->notify_cell; |
| 617 | VLOG(12) << "[time] do read value finish. [row] " << notify_cell->row; |
| 618 | |
| 619 | std::unique_ptr<RowReader> cell_reader(value_reader); |
| 620 | |
| 621 | if (cell_reader->Done()) { |
| 622 | LOG(WARNING) << "No read value, row: " << notify_cell->row; |
| 623 | return; |
| 624 | } |
| 625 | |
| 626 | if (tera::ErrorCode::kOK == cell_reader->GetError().GetType()) { |
| 627 | notify_cell->value = cell_reader->Value(); |
| 628 | notify_cell->timestamp = cell_reader->Timestamp(); |
| 629 | |
| 630 | std::shared_ptr<std::map<std::string, TableObserveInfo>> table_observe_info_read_copy; |
| 631 | { |
| 632 | MutexLock locker(&table_mutex_); |
| 633 | table_observe_info_read_copy = table_observe_info_; |
| 634 | } |
| 635 | |
| 636 | auto it = table_observe_info_read_copy->find(notify_cell->observed_column.table_name); |
| 637 | if (it == table_observe_info_read_copy->end()) { |
| 638 | LOG(WARNING) << "table not found: " << notify_cell->observed_column.table_name; |
| 639 | return; |
| 640 | } |
| 641 | |
| 642 | if (it->second.observe_columns.find(notify_cell->observed_column) == |
| 643 | it->second.observe_columns.end()) { |
| 644 | LOG(WARNING) << "column not found. cf: " << notify_cell->observed_column.family |
| 645 | << " qu: " << notify_cell->observed_column.qualifier; |
| 646 | return; |
| 647 | } |
| 648 | |
| 649 | if (it->second.observe_columns[notify_cell->observed_column].size() == 0) { |
| 650 | LOG(WARNING) << "no match observers, table=" << notify_cell->observed_column.table_name |
| 651 | << " cf=" << notify_cell->observed_column.family |
| 652 | << " qu=" << notify_cell->observed_column.qualifier; |
| 653 | return; |
| 654 | } |
| 655 | |
| 656 | if (notify_cell->observer->GetTransactionType() != kGlobalTransaction) { |
| 657 | ObserveCell(notify_cell); |
| 658 | } else { |
| 659 | AsyncReadAck(notify_cell); |
| 660 | } |
| 661 | } else { |
| 662 | LOG(WARNING) << "[read failed] table=" << notify_cell->table->GetName() |
| 663 | << " cf=" << notify_cell->observed_column.family |
| 664 | << " qu=" << notify_cell->observed_column.qualifier << " row=" << notify_cell->row |
| 665 | << " err=" << cell_reader->GetError().GetType() |
| 666 | << cell_reader->GetError().GetReason(); |
| 667 | return; |
| 668 | } |
| 669 | } |
| 670 | |
| 671 | void ScannerImpl::ObserveCell(std::shared_ptr<NotifyCell> notify_cell) { |