diff --git a/google-cloud-spanner/lib/google/cloud/spanner/commit.rb b/google-cloud-spanner/lib/google/cloud/spanner/commit.rb index 29295180..319c391d 100644 --- a/google-cloud-spanner/lib/google/cloud/spanner/commit.rb +++ b/google-cloud-spanner/lib/google/cloud/spanner/commit.rb @@ -280,6 +280,124 @@ def delete table, keys = [] keys end + ## + # Sends a message to a queue. + # + # All changes are accumulated in memory until the block passed to + # {Client#commit} 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}), 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`. + # * `Hash` - Serialized to a JSON string. + # * `Google::Protobuf::MessageExts` - Serialized and Base64-encoded. + # @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. + # + # @example + # require "google/cloud/spanner" + # + # spanner = Google::Cloud::Spanner.new + # + # db = spanner.client "my-instance", "my-database" + # + # # OrderQueue has a STRING(MAX) payload column. + # db.commit do |c| + # c.send_message "OrderQueue", 101, "Order payload" + # end + # + # @example Send binary data to a queue with a BYTES payload column + # require "google/cloud/spanner" + # + # spanner = Google::Cloud::Spanner.new + # + # db = spanner.client "my-instance", "my-database" + # + # db.commit do |c| + # c.send_message "LabelQueue", 101, StringIO.new(File.binread("label.png")) + # end + # + # @example Deliver the message in five minutes + # require "google/cloud/spanner" + # + # spanner = Google::Cloud::Spanner.new + # + # db = spanner.client "my-instance", "my-database" + # + # db.commit do |c| + # c.send_message "OrderQueue", 101, "Order payload", + # deliver_time: Time.now + 300 + # end + # + def send_message queue, key, payload, deliver_time: nil + send_mutation = V1::Mutation::Send.new( + queue: queue, + key: key_list_value(key), + payload: Convert.object_to_grpc_value(payload), + deliver_time: Convert.time_to_timestamp(deliver_time) + ) + @mutations += [V1::Mutation.new(send: send_mutation)] + key + 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#commit} 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 [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. + # + # @example + # require "google/cloud/spanner" + # + # spanner = Google::Cloud::Spanner.new + # + # db = spanner.client "my-instance", "my-database" + # + # db.commit do |c| + # c.ack_message "OrderQueue", 101 + # end + # + # @example Ack a message whether or not it exists + # require "google/cloud/spanner" + # + # spanner = Google::Cloud::Spanner.new + # + # db = spanner.client "my-instance", "my-database" + # + # db.commit do |c| + # c.ack_message "OrderQueue", 101, ignore_not_found: true + # end + # + def ack_message queue, key, ignore_not_found: false + ack_mutation = V1::Mutation::Ack.new( + queue: queue, + key: key_list_value(key), + ignore_not_found: ignore_not_found + ) + @mutations += [V1::Mutation.new(ack: ack_mutation)] + key + end + # Return the current mutations added to this `Spanner::Commit` object. # @private # @return [Array] @@ -325,6 +443,21 @@ def mutations_from_rows table, rows, type rows end + ## + # @private + # Generates a `ListValue` protobuf containing gRPC values for the given primary key. + # + # @param [Object, Array] key A scalar key, or an array of a composite key's parts. + # @return [Google::Protobuf::ListValue] + # + def key_list_value key + key = [key] unless key.is_a? Array + key_list = key.map do |k| + Convert.object_to_grpc_value k + end + Google::Protobuf::ListValue.new values: key_list + end + def key_set keys return V1::KeySet.new all: true if keys.nil? keys = [keys] unless keys.is_a? Array diff --git a/google-cloud-spanner/test/google/cloud/spanner/commit_queues_test.rb b/google-cloud-spanner/test/google/cloud/spanner/commit_queues_test.rb new file mode 100644 index 00000000..fae81e2a --- /dev/null +++ b/google-cloud-spanner/test/google/cloud/spanner/commit_queues_test.rb @@ -0,0 +1,123 @@ +# 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" + +describe Google::Cloud::Spanner::Commit, :queues, :mock_spanner do + let(:commit) { Google::Cloud::Spanner::Commit.new } + let(:deliver_time) { Time.now + 300 } + + it "builds a send mutation with scalar key" do + commit.send_message "TestQueue", 2, "Hello, Queues!" + + assert_equal 1, commit.mutations.count + mutation = commit.mutations.first + + assert mutation["send"] + assert_equal "TestQueue", mutation["send"].queue + assert_equal Google::Cloud::Spanner::Convert.object_to_grpc_value([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 "builds a send mutation with composite key" do + commit.send_message "TestQueue", ["tenant1", 2], "Hello, Queues!" + + assert_equal 1, commit.mutations.count + mutation = commit.mutations.first + + assert mutation["send"] + 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 "encodes StringIO payloads as Base64" do + raw_bytes = "\x00\x01\x02\xFF".b + commit.send_message "TestQueue", 1, StringIO.new(raw_bytes) + + assert_equal 1, commit.mutations.count + assert_equal Base64.strict_encode64(raw_bytes), commit.mutations.first["send"].payload.string_value + end + + it "encodes Hash payload as JSON string" do + hash_payload = { "event" => "user_created", "status" => "active" } + + commit.send_message "TestQueue", 1, hash_payload + + assert_equal 1, commit.mutations.count + mutation = commit.mutations.first + + assert_equal hash_payload.to_json, mutation["send"].payload.string_value + end + + it "raises ArgumentError for unsupported payload types" do + assert_raises ArgumentError do + commit.send_message "TestQueue", 1, Object.new + end + end + + it "sets deliver_time" do + commit.send_message "TestQueue", 1, "payload", deliver_time: deliver_time + + expected = Google::Cloud::Spanner::Convert.time_to_timestamp deliver_time + assert_equal expected, commit.mutations.first["send"].deliver_time + end + + it "raises ArgumentError for an unknown send_message keyword" do + assert_raises(ArgumentError) { commit.send_message "TestQueue", 1, "payload", deliver_at: deliver_time } + assert_empty commit.mutations + end + + it "returns the key from send_message" do + assert_equal 2, commit.send_message("TestQueue", 2, "payload") + assert_equal ["tenant1", 2], commit.send_message("TestQueue", ["tenant1", 2], "payload") + end + + it "builds an ack mutation with scalar and composite keys" do + commit.ack_message "TestQueue", 2 + commit.ack_message "TestQueue", ["tenant1", 2] + + assert_equal 2, commit.mutations.count + m1 = commit.mutations[0] + m2 = commit.mutations[1] + + assert m1.ack + assert_equal "TestQueue", m1.ack.queue + assert_equal Google::Cloud::Spanner::Convert.object_to_grpc_value([2]).list_value, m1.ack.key + assert_equal false, m1.ack.ignore_not_found + + assert m2.ack + assert_equal "TestQueue", m2.ack.queue + assert_equal Google::Cloud::Spanner::Convert.object_to_grpc_value(["tenant1", 2]).list_value, m2.ack.key + assert_equal false, m2.ack.ignore_not_found + end + + it "sets ignore_not_found" do + commit.ack_message "TestQueue", 1, ignore_not_found: true + + assert_equal true, commit.mutations.first.ack.ignore_not_found + end + + it "raises ArgumentError for an unknown ack_message keyword" do + assert_raises(ArgumentError) { commit.ack_message "TestQueue", 1, ignore_missing: true } + assert_empty commit.mutations + end + + it "returns the key from ack_message" do + assert_equal 2, commit.ack_message("TestQueue", 2) + assert_equal ["tenant1", 2], commit.ack_message("TestQueue", ["tenant1", 2]) + end +end