leverage CommitOptions to propagate flag

Author: scotthart

google/cloud/spanner/commit_options.cc

@@ -33,6 +33,10 @@ CommitOptions::CommitOptions(Options const& opts)
   if (opts.has<MaxCommitDelayOption>()) {
     max_commit_delay_ = opts.get<MaxCommitDelayOption>();
   }
+  if (opts.has<ExcludeTransactionFromChangeStreamsOption>()) {
+    exclude_txn_from_change_streams_ =
+        opts.get<ExcludeTransactionFromChangeStreamsOption>();
+  }
 }
 
 CommitOptions::operator Options() const {
@@ -41,6 +45,10 @@ CommitOptions::operator Options() const {
   if (request_priority_) opts.set<RequestPriorityOption>(*request_priority_);
   if (transaction_tag_) opts.set<TransactionTagOption>(*transaction_tag_);
   if (max_commit_delay_) opts.set<MaxCommitDelayOption>(*max_commit_delay_);
+  if (exclude_txn_from_change_streams_) {
+    opts.set<ExcludeTransactionFromChangeStreamsOption>(
+        *exclude_txn_from_change_streams_);
+  }
   return opts;
 }
 

google/cloud/spanner/commit_options.h

@@ -107,13 +107,23 @@ class CommitOptions {
     return max_commit_delay_;
   }
 
+  CommitOptions& set_exclude_txn_from_change_streams(bool exclude) {
+    exclude_txn_from_change_streams_ = exclude;
+    return *this;
+  }
+
+  absl::optional<bool> const& exclude_txn_from_change_streams() const {
+    return exclude_txn_from_change_streams_;
+  }
+
  private:
   // Note that CommitRequest.request_options.request_tag is ignored,
   // so we do not even provide a mechanism to specify one.
   bool return_stats_ = false;
   absl::optional<RequestPriority> request_priority_;
   absl::optional<std::string> transaction_tag_;
   absl::optional<std::chrono::milliseconds> max_commit_delay_;
+  absl::optional<bool> exclude_txn_from_change_streams_;
 };
 
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END

google/cloud/spanner/internal/connection_impl.cc

@@ -580,12 +580,6 @@ StatusOr<google::spanner::v1::Transaction> ConnectionImpl::BeginTransaction(
 
   auto stub = GetStubBasedOnSessionMode(*session, ctx);
   auto const& current = internal::CurrentOptions();
-
-  if (current.has<spanner::ExcludeTransactionFromChangeStreamsOption>() &&
-      current.get<spanner::ExcludeTransactionFromChangeStreamsOption>()) {
-    begin.mutable_options()->set_exclude_txn_from_change_streams(true);
-  }
-
   auto response = RetryLoop(
       RetryPolicyPrototype(current)->clone(),
       BackoffPolicyPrototype(current)->clone(), Idempotency::kIdempotent,