impl(bigtable): update DataConnectionImpl to create and call OperationContext hooks (#15320)

Author: scotthart

google/cloud/bigtable/CMakeLists.txt

@@ -456,6 +456,7 @@ if (BUILD_TESTING)
         internal/logging_data_client_test.cc
         internal/metrics_test.cc
         internal/mutate_rows_limiter_test.cc
+        internal/operation_context_factory_test.cc
         internal/operation_context_test.cc
         internal/prefix_range_end_test.cc
         internal/rate_limiter_test.cc

google/cloud/bigtable/bigtable_client_unit_tests.bzl

@@ -61,6 +61,7 @@ bigtable_client_unit_tests = [
     "internal/logging_data_client_test.cc",
     "internal/metrics_test.cc",
     "internal/mutate_rows_limiter_test.cc",
+    "internal/operation_context_factory_test.cc",
     "internal/operation_context_test.cc",
     "internal/prefix_range_end_test.cc",
     "internal/rate_limiter_test.cc",

google/cloud/bigtable/internal/data_connection_impl.cc

@@ -42,6 +42,13 @@ class DataConnectionImpl : public bigtable::DataConnection {
                      std::shared_ptr<MutateRowsLimiter> limiter,
                      Options options);
 
+  // This constructor is used for testing.
+  DataConnectionImpl(
+      std::unique_ptr<BackgroundThreads> background,
+      std::shared_ptr<BigtableStub> stub,
+      std::unique_ptr<OperationContextFactory> operation_context_factory,
+      std::shared_ptr<MutateRowsLimiter> limiter, Options options);
+
   Options options() override { return options_; }
 
   Status Apply(std::string const& table_name,
@@ -95,9 +102,17 @@ class DataConnectionImpl : public bigtable::DataConnection {
       bigtable::Filter filter) override;
 
  private:
-  std::unique_ptr<OperationContextFactory> operation_context_factory_;
+  void AsyncReadRowsHelper(std::string const& table_name,
+                           std::function<future<bool>(bigtable::Row)> on_row,
+                           std::function<void(Status)> on_finish,
+                           bigtable::RowSet row_set, std::int64_t rows_limit,
+                           bigtable::Filter filter,
+                           internal::ImmutableOptions const& current,
+                           std::shared_ptr<OperationContext> operation_context);
+
   std::unique_ptr<BackgroundThreads> background_;
   std::shared_ptr<BigtableStub> stub_;
+  std::unique_ptr<OperationContextFactory> operation_context_factory_;
   std::shared_ptr<MutateRowsLimiter> limiter_;
   Options options_;
 };