From 370d1b9f822c65a22a40b5f0b926f579707755e8 Mon Sep 17 00:00:00 2001 From: Torrey Payne <11740989+torreypayne@users.noreply.github.com> Date: Mon, 12 Oct 2026 19:52:51 +0000 Subject: [PATCH 1/3] feat(spanner): add send_message and ack_message to Transaction --- .../lib/google/cloud/spanner/transaction.rb | 84 ++++++++++++++++++ .../cloud/spanner/client/transaction_test.rb | 47 ++++++++++ .../cloud/spanner/transaction/queues_test.rb | 86 +++++++++++++++++++ 3 files changed, 217 insertions(+) create mode 100644 google-cloud-spanner/test/google/cloud/spanner/transaction/queues_test.rb diff --git a/google-cloud-spanner/lib/google/cloud/spanner/transaction.rb b/google-cloud-spanner/lib/google/cloud/spanner/transaction.rb index 7880959e..0856f6be 100644 --- a/google-cloud-spanner/lib/google/cloud/spanner/transaction.rb +++ b/google-cloud-spanner/lib/google/cloud/spanner/transaction.rb @@ -1075,6 +1075,90 @@ def delete table, keys = [] @commit.delete table, keys end + ## + # Sends a message to a queue. + # + # All changes are accumulated in memory until the block passed to + # {Client#transaction} completes. + # + # @param [String] queue The name of the queue. + # @param [Object, Array] key A single key or array of composite key parts for the message. + # @param [String, IO, StringIO, Hash, Google::Protobuf::MessageExts] payload + # The message payload. It is converted the same way as column values + # (see {#upsert}): + # * `String` - Sent as text. + # * `IO`, `StringIO`, `File` - Sent as bytes (Base64-encoded). Ruby has + # no bytes type, so wrap binary data in a `StringIO`. + # * `Hash` - Serialized to a JSON string. + # * `Google::Protobuf::MessageExts` - Serialized and Base64-encoded. + # @param [SendOptions, Hash] send_options A {SendOptions} or a hash of + # send options. Send options are optional. The following options can + # be provided: + # + # * `:deliver_time` (Time) The time at which Spanner begins trying to + # deliver the message. If not set, the message is delivered + # immediately. A time in the past is replaced with a time closer to + # the current time. + # + # @return [Object, Array] The key of the message. + # + # @raise [ArgumentError] If the payload type is unsupported, or if + # `send_options` is not a {SendOptions}, a hash of its options, or + # `nil`. + # + # @example Send a message in the same transaction as the row it describes + # require "google/cloud/spanner" + # + # spanner = Google::Cloud::Spanner.new + # db = spanner.client "my-instance", "my-database" + # + # db.transaction do |tx| + # tx.insert "orders", [{ id: 101, status: "NEW" }] + # tx.send_message "OrderQueue", 101, StringIO.new("Order payload") + # end + # + def send_message queue, key, payload, send_options: nil + ensure_session! + @commit.send_message queue, key, payload, send_options: send_options + end + + ## + # Acknowledges a message in a queue, which removes it from the queue. + # + # All changes are accumulated in memory until the block passed to + # {Client#transaction} completes. + # + # @param [String] queue The name of the queue. + # @param [Object, Array] key A single key or array of composite key parts to match. + # @param [AckOptions, Hash] ack_options An {AckOptions} or a hash of ack + # options. Ack options are optional. The following options can be + # provided: + # + # * `:ignore_not_found` (Boolean) If `true`, the ack succeeds even if + # the message doesn't exist. Default value is `false`, which fails + # the commit with `NOT_FOUND` when the message doesn't exist. + # + # @return [Object, Array] The key of the message. + # + # @raise [ArgumentError] If `ack_options` is not an {AckOptions}, a + # hash of its options, or `nil`. + # + # @example Ack a message in the same transaction as the write it caused + # require "google/cloud/spanner" + # + # spanner = Google::Cloud::Spanner.new + # db = spanner.client "my-instance", "my-database" + # + # db.transaction do |tx| + # tx.update "orders", [{ id: 101, status: "PROCESSED" }] + # tx.ack_message "OrderQueue", 101 + # end + # + def ack_message queue, key, ack_options: nil + ensure_session! + @commit.ack_message queue, key, ack_options: ack_options + end + ## # @private # Returns the field names and types for a table. diff --git a/google-cloud-spanner/test/google/cloud/spanner/client/transaction_test.rb b/google-cloud-spanner/test/google/cloud/spanner/client/transaction_test.rb index 002e43a0..3ca56e3a 100644 --- a/google-cloud-spanner/test/google/cloud/spanner/client/transaction_test.rb +++ b/google-cloud-spanner/test/google/cloud/spanner/client/transaction_test.rb @@ -454,6 +454,53 @@ mock.verify end + it "sends and acks queue messages" do + mutations = [ + Google::Cloud::Spanner::V1::Mutation.new( + send: Google::Cloud::Spanner::V1::Mutation::Send.new( + queue: "TestQueue", + key: Google::Cloud::Spanner::Convert.object_to_grpc_value([101]).list_value, + payload: Google::Cloud::Spanner::Convert.object_to_grpc_value("Hello, Queues!") + ) + ), + Google::Cloud::Spanner::V1::Mutation.new( + ack: Google::Cloud::Spanner::V1::Mutation::Ack.new( + queue: "TestQueue", + key: Google::Cloud::Spanner::Convert.object_to_grpc_value([102]).list_value, + ignore_not_found: false + ) + ) + ] + + mock = Minitest::Mock.new + mock.expect :create_session, session_grpc, [{ database: database_path(instance_id, database_id), session: default_session_request }, default_options] + mock.expect :begin_transaction, transaction_grpc, [{ + session: session_grpc.name, + options: tx_opts, + request_options: nil, + mutation_key: mutations[0] + }, default_options] + + mock.expect :commit, commit_resp, [{ + session: session_grpc.name, + mutations: mutations, + transaction_id: transaction_id, + single_use_transaction: nil, precommit_token: nil, + request_options: nil + }, default_options] + spanner.service.mocked_service = mock + + timestamp = client.transaction do |tx| + tx.send_message "TestQueue", 101, "Hello, Queues!" + tx.ack_message "TestQueue", 102 + end + _(timestamp).must_equal commit_time + + shutdown_client! client + + mock.verify + end + it "deletes multiple rows of key ranges" do mutations = [ Google::Cloud::Spanner::V1::Mutation.new( diff --git a/google-cloud-spanner/test/google/cloud/spanner/transaction/queues_test.rb b/google-cloud-spanner/test/google/cloud/spanner/transaction/queues_test.rb new file mode 100644 index 00000000..8eb88fe1 --- /dev/null +++ b/google-cloud-spanner/test/google/cloud/spanner/transaction/queues_test.rb @@ -0,0 +1,86 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +require "helper" + +# Transaction#send_message and #ack_message delegate to the transaction's +# Commit, so payload and key conversion is covered in commit_queues_test.rb. +# These tests check that every argument reaches the transaction's mutations. +describe Google::Cloud::Spanner::Transaction, :queues, :mock_spanner do + let(:instance_id) { "my-instance-id" } + let(:database_id) { "my-database-id" } + let(:session_id) { "session123" } + let(:session_grpc) { Google::Cloud::Spanner::V1::Session.new name: session_path(instance_id, database_id, session_id) } + let(:session) { Google::Cloud::Spanner::Session.from_grpc session_grpc, spanner.service } + let(:transaction_id) { "tx789" } + let(:transaction_grpc) { Google::Cloud::Spanner::V1::Transaction.new id: transaction_id } + let(:transaction) { Google::Cloud::Spanner::Transaction.from_grpc transaction_grpc, session } + let(:deliver_time) { Time.now + 300 } + + it "adds a send mutation to the transaction" do + key = transaction.send_message "TestQueue", ["tenant1", 2], "Hello, Queues!" + + assert_equal ["tenant1", 2], key + assert_equal 1, transaction.mutations.count + mutation = transaction.mutations.first + + assert_equal "TestQueue", mutation["send"].queue + assert_equal Google::Cloud::Spanner::Convert.object_to_grpc_value(["tenant1", 2]).list_value, mutation["send"].key + assert_equal Google::Cloud::Spanner::Convert.object_to_grpc_value("Hello, Queues!"), mutation["send"].payload + assert_nil mutation["send"].deliver_time + end + + it "passes send_options to the send mutation" do + send_options = Google::Cloud::Spanner::SendOptions.new deliver_time: deliver_time + + transaction.send_message "TestQueue", 1, "payload", send_options: send_options + transaction.send_message "TestQueue", 2, "payload", send_options: { deliver_time: deliver_time } + + expected = Google::Cloud::Spanner::Convert.time_to_timestamp deliver_time + assert_equal [expected, expected], transaction.mutations.map { |mutation| mutation["send"].deliver_time } + end + + it "adds an ack mutation to the transaction" do + key = transaction.ack_message "TestQueue", ["tenant1", 2] + + assert_equal ["tenant1", 2], key + assert_equal 1, transaction.mutations.count + mutation = transaction.mutations.first + + assert_equal "TestQueue", mutation.ack.queue + assert_equal Google::Cloud::Spanner::Convert.object_to_grpc_value(["tenant1", 2]).list_value, mutation.ack.key + assert_equal false, mutation.ack.ignore_not_found + end + + it "passes ack_options to the ack mutation" do + ack_options = Google::Cloud::Spanner::AckOptions.new ignore_not_found: true + + transaction.ack_message "TestQueue", 1, ack_options: ack_options + transaction.ack_message "TestQueue", 2, ack_options: { ignore_not_found: true } + + assert_equal [true, true], transaction.mutations.map { |mutation| mutation.ack.ignore_not_found } + end + + it "raises when session is missing" do + transaction.session = nil + + assert_raises RuntimeError do + transaction.send_message "TestQueue", 1, "test" + end + + assert_raises RuntimeError do + transaction.ack_message "TestQueue", 1 + end + end +end From 4da6dbcfdbe599c8d3a90db0238c2db5ded0f7d3 Mon Sep 17 00:00:00 2001 From: Torrey Payne <11740989+torreypayne@users.noreply.github.com> Date: Tue, 20 Oct 2026 18:04:56 +0000 Subject: [PATCH 2/3] refactor(spanner): pass the queue keyword arguments through Transaction Transaction#send_message and #ack_message now take deliver_time: and ignore_not_found: and pass them to the transaction's Commit, matching the keyword arguments on Commit. --- .../lib/google/cloud/spanner/transaction.rb | 37 ++++++------------- .../cloud/spanner/transaction/queues_test.rb | 18 +++------ 2 files changed, 18 insertions(+), 37 deletions(-) diff --git a/google-cloud-spanner/lib/google/cloud/spanner/transaction.rb b/google-cloud-spanner/lib/google/cloud/spanner/transaction.rb index 0856f6be..3bad6976 100644 --- a/google-cloud-spanner/lib/google/cloud/spanner/transaction.rb +++ b/google-cloud-spanner/lib/google/cloud/spanner/transaction.rb @@ -1091,20 +1091,14 @@ def delete table, keys = [] # no bytes type, so wrap binary data in a `StringIO`. # * `Hash` - Serialized to a JSON string. # * `Google::Protobuf::MessageExts` - Serialized and Base64-encoded. - # @param [SendOptions, Hash] send_options A {SendOptions} or a hash of - # send options. Send options are optional. The following options can - # be provided: - # - # * `:deliver_time` (Time) The time at which Spanner begins trying to - # deliver the message. If not set, the message is delivered - # immediately. A time in the past is replaced with a time closer to - # the current time. + # @param [Time, nil] deliver_time The time at which Spanner begins + # trying to deliver the message. Optional. If not set, the message is + # delivered immediately. A time in the past is replaced with a time + # closer to the current time. # # @return [Object, Array] The key of the message. # - # @raise [ArgumentError] If the payload type is unsupported, or if - # `send_options` is not a {SendOptions}, a hash of its options, or - # `nil`. + # @raise [ArgumentError] If the payload type is unsupported. # # @example Send a message in the same transaction as the row it describes # require "google/cloud/spanner" @@ -1117,9 +1111,9 @@ def delete table, keys = [] # tx.send_message "OrderQueue", 101, StringIO.new("Order payload") # end # - def send_message queue, key, payload, send_options: nil + def send_message queue, key, payload, deliver_time: nil ensure_session! - @commit.send_message queue, key, payload, send_options: send_options + @commit.send_message queue, key, payload, deliver_time: deliver_time end ## @@ -1130,19 +1124,12 @@ def send_message queue, key, payload, send_options: nil # # @param [String] queue The name of the queue. # @param [Object, Array] key A single key or array of composite key parts to match. - # @param [AckOptions, Hash] ack_options An {AckOptions} or a hash of ack - # options. Ack options are optional. The following options can be - # provided: - # - # * `:ignore_not_found` (Boolean) If `true`, the ack succeeds even if - # the message doesn't exist. Default value is `false`, which fails - # the commit with `NOT_FOUND` when the message doesn't exist. + # @param [Boolean] ignore_not_found If `true`, the ack succeeds even if + # the message doesn't exist. Optional. The default is `false`, which + # fails the commit with `NOT_FOUND` when the message doesn't exist. # # @return [Object, Array] The key of the message. # - # @raise [ArgumentError] If `ack_options` is not an {AckOptions}, a - # hash of its options, or `nil`. - # # @example Ack a message in the same transaction as the write it caused # require "google/cloud/spanner" # @@ -1154,9 +1141,9 @@ def send_message queue, key, payload, send_options: nil # tx.ack_message "OrderQueue", 101 # end # - def ack_message queue, key, ack_options: nil + def ack_message queue, key, ignore_not_found: false ensure_session! - @commit.ack_message queue, key, ack_options: ack_options + @commit.ack_message queue, key, ignore_not_found: ignore_not_found end ## diff --git a/google-cloud-spanner/test/google/cloud/spanner/transaction/queues_test.rb b/google-cloud-spanner/test/google/cloud/spanner/transaction/queues_test.rb index 8eb88fe1..8414c615 100644 --- a/google-cloud-spanner/test/google/cloud/spanner/transaction/queues_test.rb +++ b/google-cloud-spanner/test/google/cloud/spanner/transaction/queues_test.rb @@ -41,14 +41,11 @@ assert_nil mutation["send"].deliver_time end - it "passes send_options to the send mutation" do - send_options = Google::Cloud::Spanner::SendOptions.new deliver_time: deliver_time - - transaction.send_message "TestQueue", 1, "payload", send_options: send_options - transaction.send_message "TestQueue", 2, "payload", send_options: { deliver_time: deliver_time } + it "passes deliver_time to the send mutation" do + transaction.send_message "TestQueue", 1, "payload", deliver_time: deliver_time expected = Google::Cloud::Spanner::Convert.time_to_timestamp deliver_time - assert_equal [expected, expected], transaction.mutations.map { |mutation| mutation["send"].deliver_time } + assert_equal expected, transaction.mutations.first["send"].deliver_time end it "adds an ack mutation to the transaction" do @@ -63,13 +60,10 @@ assert_equal false, mutation.ack.ignore_not_found end - it "passes ack_options to the ack mutation" do - ack_options = Google::Cloud::Spanner::AckOptions.new ignore_not_found: true - - transaction.ack_message "TestQueue", 1, ack_options: ack_options - transaction.ack_message "TestQueue", 2, ack_options: { ignore_not_found: true } + it "passes ignore_not_found to the ack mutation" do + transaction.ack_message "TestQueue", 1, ignore_not_found: true - assert_equal [true, true], transaction.mutations.map { |mutation| mutation.ack.ignore_not_found } + assert_equal true, transaction.mutations.first.ack.ignore_not_found end it "raises when session is missing" do From 95a1477e2ef32ca4b4c38a23ab35f541c30381c6 Mon Sep 17 00:00:00 2001 From: Torrey Payne <11740989+torreypayne@users.noreply.github.com> Date: Tue, 20 Oct 2026 23:03:19 +0000 Subject: [PATCH 3/3] docs(spanner): use a String payload in the Transaction#send_message example Match the Commit#send_message examples: pass a String and say the queue's payload column is STRING(MAX), instead of wrapping text in a StringIO, which is only right for a BYTES payload column. --- .../lib/google/cloud/spanner/transaction.rb | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/google-cloud-spanner/lib/google/cloud/spanner/transaction.rb b/google-cloud-spanner/lib/google/cloud/spanner/transaction.rb index 3bad6976..e6e0bb9f 100644 --- a/google-cloud-spanner/lib/google/cloud/spanner/transaction.rb +++ b/google-cloud-spanner/lib/google/cloud/spanner/transaction.rb @@ -1085,7 +1085,8 @@ def delete table, keys = [] # @param [Object, Array] key A single key or array of composite key parts for the message. # @param [String, IO, StringIO, Hash, Google::Protobuf::MessageExts] payload # The message payload. It is converted the same way as column values - # (see {#upsert}): + # (see {#upsert}), so its type must match the queue's payload column, + # such as a `String` for `STRING` or a `StringIO` for `BYTES`: # * `String` - Sent as text. # * `IO`, `StringIO`, `File` - Sent as bytes (Base64-encoded). Ruby has # no bytes type, so wrap binary data in a `StringIO`. @@ -1106,9 +1107,10 @@ def delete table, keys = [] # spanner = Google::Cloud::Spanner.new # db = spanner.client "my-instance", "my-database" # + # # OrderQueue has a STRING(MAX) payload column. # db.transaction do |tx| # tx.insert "orders", [{ id: 101, status: "NEW" }] - # tx.send_message "OrderQueue", 101, StringIO.new("Order payload") + # tx.send_message "OrderQueue", 101, "Order payload" # end # def send_message queue, key, payload, deliver_time: nil