Method ObserveTest

src/observer/test/observer_test.cc:462–532

  }

  void ObserveTest() {
    tera::ErrorCode err;
    tera::Client* client = tera::Client::NewClient(FLAGS_flagfile, &err);
    // for ut test
    EXPECT_EQ(tera::ErrorCode::kOK, err.GetType());
    // for no core
    if (tera::ErrorCode::kOK != err.GetType()) {
      LOG(ERROR) << "new client failed";
      return;
    }

    // create table
    tera::TableDescriptor table_desc("observer_table");
    table_desc.EnableTxn();
    table_desc.AddLocalityGroup("notify");
    tera::ColumnFamilyDescriptor* cf_t = table_desc.AddColumnFamily(kNotifyColumnFamily, "notify");
    cf_t->EnableGlobalTransaction();

    table_desc.AddLocalityGroup("lg1");
    tera::ColumnFamilyDescriptor* cf1 = table_desc.AddColumnFamily("cf", "lg1");
    cf1->EnableGlobalTransaction();
    cf1->EnableNotify();
    tera::ColumnFamilyDescriptor* cf2 = table_desc.AddColumnFamily("cf_1", "lg1");
    cf2->EnableGlobalTransaction();
    cf2->EnableNotify();

    ExtendNotifyLgToDescriptor(&table_desc);

    client->CreateTable(table_desc, &err);
    if (err.GetType() != tera::ErrorCode::kOK) {
      LOG(ERROR) << "Create table fail";
    }

    FLAGS_tera_sdk_client_for_gtxn = true;
    FLAGS_tera_coord_type = "ins";
    common::ThreadPool thread_pool(5);
    ScannerImpl* scanner = new ScannerImpl();
    Observer* observer = new DemoObserver();
    scanner->key_selector_.reset(new RandomKeySelector());

    // single thread

    err = scanner->Observe("observer_table", "cf", "qualifier", observer);
    EXPECT_TRUE(err.GetType() != tera::ErrorCode::kOK);

    scanner->tera_client_.reset(tera::Client::NewClient(FLAGS_flagfile, &err));
    EXPECT_EQ(scanner->table_observe_info_->size(), 0);

    err = scanner->Observe("observer_table", "cf", "qualifier", observer);
    EXPECT_TRUE(err.GetType() == tera::ErrorCode::kOK);

    err = scanner->Observe("observer_table", "cf", "qualifier", observer);
    EXPECT_FALSE(err.GetType() == tera::ErrorCode::kOK);

    err = scanner->Observe("observer_table", "cf_1", "qualifier", observer);
    EXPECT_TRUE(err.GetType() == tera::ErrorCode::kOK);

    // multi thread