cleanup(bigtable): AsyncReadRows accepts move-only types again (#9317)

Author: dbolduc

google/cloud/bigtable/table.h

@@ -889,19 +889,32 @@ class Table {
             future<bool>>::value,
         "RowFunctor should return a future<bool>.");
 
+    auto on_row_ptr = std::make_shared<RowFunctor>(std::move(on_row));
+    // NOLINTNEXTLINE(performance-unnecessary-value-param)
+    auto on_row_fn = [on_row_ptr](Row row) {
+      return (*on_row_ptr)(std::move(row));
+    };
+
+    auto on_finish_ptr = std::make_shared<FinishFunctor>(std::move(on_finish));
+    // NOLINTNEXTLINE(performance-unnecessary-value-param)
+    auto on_finish_fn = [on_finish_ptr](Status status) {
+      return (*on_finish_ptr)(std::move(status));
+    };
+
     if (connection_) {
       google::cloud::internal::OptionsSpan span(options_);
-      connection_->AsyncReadRows(
-          app_profile_id_, table_name_, std::move(on_row), std::move(on_finish),
-          std::move(row_set), rows_limit, std::move(filter));
+      connection_->AsyncReadRows(app_profile_id_, table_name_,
+                                 std::move(on_row_fn), std::move(on_finish_fn),
+                                 std::move(row_set), rows_limit,
+                                 std::move(filter));
       return;
     }
 
     bigtable_internal::LegacyAsyncRowReader::Create(
         background_threads_->cq(), client_, app_profile_id_, table_name_,
-        std::move(on_row), std::move(on_finish), std::move(row_set), rows_limit,
-        std::move(filter), clone_rpc_retry_policy(), clone_rpc_backoff_policy(),
-        metadata_update_policy_,
+        std::move(on_row_fn), std::move(on_finish_fn), std::move(row_set),
+        rows_limit, std::move(filter), clone_rpc_retry_policy(),
+        clone_rpc_backoff_policy(), metadata_update_policy_,
         absl::make_unique<bigtable::internal::ReadRowsParserFactory>());
   }
 

google/cloud/bigtable/table_test.cc

@@ -35,6 +35,7 @@ using ::testing::Matcher;
 using ::testing::MockFunction;
 using ::testing::Property;
 using ::testing::Return;
+using ::testing::Unused;
 
 auto const* const kProjectId = "test-project";
 auto const* const kInstanceId = "test-instance";
@@ -498,6 +499,40 @@ TEST(TableTest, AsyncReadRowsWithRowLimit) {
                       TestRowSet(), 42, TestFilter());
 }
 
+TEST(TableTest, AsyncReadRowsAcceptsMoveOnlyTypes) {
+  auto mock = std::make_shared<MockDataConnection>();
+  EXPECT_CALL(*mock, AsyncReadRows)
+      .WillOnce([](Unused, Unused,
+                   std::function<future<bool>(bigtable::Row)> const& on_row,
+                   std::function<void(Status)> const& on_finish, Unused, Unused,
+                   Unused) {
+        // Invoke the callbacks.
+        EXPECT_TRUE(on_row(bigtable::Row("row", {})).get());
+        on_finish(PermanentError());
+      });
+
+  class MoveOnly {
+   public:
+    MoveOnly() = default;
+    MoveOnly(MoveOnly&&) = default;
+    MoveOnly& operator=(MoveOnly&&) = default;
+    MoveOnly(const MoveOnly&) = delete;
+    MoveOnly& operator=(const MoveOnly&) = delete;
+
+    future<bool> operator()(Row const& row) {
+      EXPECT_EQ("row", row.row_key());
+      return make_ready_future(true);
+    }
+
+    void operator()(Status const& status) {
+      EXPECT_THAT(status, StatusIs(StatusCode::kPermissionDenied));
+    }
+  };
+
+  auto table = TestTable(std::move(mock));
+  table.AsyncReadRows(MoveOnly{}, MoveOnly{}, TestRowSet(), TestFilter());
+}
+
 TEST(TableTest, AsyncReadRow) {
   auto mock = std::make_shared<MockDataConnection>();
   EXPECT_CALL(*mock, AsyncReadRow)