diff --git a/acceptance/data/fixtures.rb b/acceptance/data/fixtures.rb index 0cdf4ae5..df35979d 100644 --- a/acceptance/data/fixtures.rb +++ b/acceptance/data/fixtures.rb @@ -196,6 +196,17 @@ def items_ddl_statement ITEMS end + # Not part of schema_ddl_statements: queues need an Enterprise edition + # instance, and the emulator doesn't support them yet. + def queue_ddl_statement + <<~QUEUE + CREATE QUEUE IF NOT EXISTS TestQueue ( + Id INT64 NOT NULL, + Payload BYTES(MAX) NOT NULL + ) PRIMARY KEY (Id) + QUEUE + end + def schema_pg_ddl_statements [ stuff_pg_ddl_statement, diff --git a/google-cloud-spanner/OVERVIEW.md b/google-cloud-spanner/OVERVIEW.md index 25c6195e..5b79d017 100644 --- a/google-cloud-spanner/OVERVIEW.md +++ b/google-cloud-spanner/OVERVIEW.md @@ -298,6 +298,60 @@ db.snapshot do |snp| end ``` +## Using queues + +A queue holds messages for consumers to receive. You create queues with the +`CREATE QUEUE` statement (see the [DDL +reference](https://cloud.google.com/spanner/docs/reference/standard-sql/data-definition-language)). + +Use {Google::Cloud::Spanner::Commit#send_message Commit#send_message} to send a +message. The payload is converted the same way as column values, so wrap binary +data in a `StringIO`, as you would for a `BYTES` column: + +```ruby +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, StringIO.new("Order payload") +end +``` + +Receive messages with the `RECEIVE_` table-valued function, and use +{Google::Cloud::Spanner::Transaction#ack_message Transaction#ack_message} to +acknowledge each one in the same transaction as the write it caused: + +```ruby +require "google/cloud/spanner" + +spanner = Google::Cloud::Spanner.new + +db = spanner.client "my-instance", "my-database" + +results = db.execute_query "SELECT Id, Payload FROM RECEIVE_OrderQueue()" + +results.rows.each do |row| + db.transaction do |tx| + tx.update "orders", [{ id: row[:Id], status: "PROCESSED" }] + tx.ack_message "OrderQueue", row[:Id] + end +end +``` + +Optional settings go in a {Google::Cloud::Spanner::SendOptions SendOptions} or +{Google::Cloud::Spanner::AckOptions AckOptions}, or in a hash with the same +keys: + +```ruby +db.commit do |c| + c.send_message "OrderQueue", 102, StringIO.new("Order payload"), + send_options: { deliver_time: Time.now + 300 } +end +``` + ## Deleting databases Use {Google::Cloud::Spanner::Admin::Database#database_admin diff --git a/google-cloud-spanner/acceptance/spanner/client/queues_test.rb b/google-cloud-spanner/acceptance/spanner/client/queues_test.rb new file mode 100644 index 00000000..f72f3faa --- /dev/null +++ b/google-cloud-spanner/acceptance/spanner/client/queues_test.rb @@ -0,0 +1,59 @@ +# 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 "spanner_helper" + +# An ack without ignore_not_found fails with NOT_FOUND unless the message +# exists, so a successful ack also shows that the send landed. +describe "Spanner Client Queues", :queues, :spanner do + let(:db) { spanner_client } + + before do + skip "Cloud Spanner Emulator does not yet support Queues (Send & Ack)" if emulator_enabled? + + # Created once per run, in the per-run test database. + $spanner_test_queue_created ||= begin + db_job = db.database.update statements: [queue_ddl_statement] + db_job.wait_until_done! + raise GRPC::BadStatus.new(db_job.error.code, db_job.error.message) if db_job.error? + true + end + end + + it "sends and acks messages with Client#commit" do + commit_time = db.commit do |c| + c.send_message "TestQueue", 1, StringIO.new("payload1") + c.send_message "TestQueue", 2, StringIO.new("payload2") + end + _(commit_time).must_be_kind_of Time + + commit_time = db.commit do |c| + c.ack_message "TestQueue", 1 + c.ack_message "TestQueue", 2 + end + _(commit_time).must_be_kind_of Time + end + + it "sends and acks a message with Client#transaction" do + commit_time = db.transaction do |tx| + tx.send_message "TestQueue", 3, StringIO.new("payload3") + end + _(commit_time).must_be_kind_of Time + + commit_time = db.transaction do |tx| + tx.ack_message "TestQueue", 3 + end + _(commit_time).must_be_kind_of Time + end +end diff --git a/google-cloud-spanner/google-cloud-spanner.gemspec b/google-cloud-spanner/google-cloud-spanner.gemspec index 219adb61..9d2bbdc1 100644 --- a/google-cloud-spanner/google-cloud-spanner.gemspec +++ b/google-cloud-spanner/google-cloud-spanner.gemspec @@ -24,5 +24,5 @@ Gem::Specification.new do |gem| gem.add_dependency "google-cloud-core", "~> 1.7" gem.add_dependency "google-cloud-spanner-admin-database-v1", "~> 1.4" gem.add_dependency "google-cloud-spanner-admin-instance-v1", "~> 1.6" - gem.add_dependency "google-cloud-spanner-v1", "~> 1.6" + gem.add_dependency "google-cloud-spanner-v1", "~> 1.12" end diff --git a/google-cloud-spanner/lib/google/cloud/spanner/ack_options.rb b/google-cloud-spanner/lib/google/cloud/spanner/ack_options.rb new file mode 100644 index 00000000..7206b48c --- /dev/null +++ b/google-cloud-spanner/lib/google/cloud/spanner/ack_options.rb @@ -0,0 +1,60 @@ +# 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. + + +module Google + module Cloud + module Spanner + ## + # # AckOptions + # + # Optional settings for acknowledging a queue message with + # {Commit#ack_message} or {Transaction#ack_message}. A hash with the + # same keys can be passed instead. + # + # @example + # require "google/cloud/spanner" + # + # spanner = Google::Cloud::Spanner.new + # + # db = spanner.client "my-instance", "my-database" + # + # ack_options = Google::Cloud::Spanner::AckOptions.new ignore_not_found: true + # + # db.commit do |c| + # c.ack_message "OrderQueue", 101, ack_options: ack_options + # end + # + class AckOptions + ## + # Whether the ack succeeds when the message doesn't exist. + # + # @return [Boolean] + attr_reader :ignore_not_found + + ## + # Creates a new AckOptions. + # + # @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. + # + def initialize ignore_not_found: false + @ignore_not_found = ignore_not_found + end + end + end + end +end diff --git a/google-cloud-spanner/lib/google/cloud/spanner/client.rb b/google-cloud-spanner/lib/google/cloud/spanner/client.rb index dacb4536..33d7d717 100644 --- a/google-cloud-spanner/lib/google/cloud/spanner/client.rb +++ b/google-cloud-spanner/lib/google/cloud/spanner/client.rb @@ -35,6 +35,12 @@ module Spanner # # A client is used to read and/or modify data in a Cloud Spanner database. # + # A client also works with queues. Send and acknowledge messages with + # {Commit#send_message} and {Commit#ack_message} inside {#commit}, or with + # {Transaction#send_message} and {Transaction#ack_message} inside + # {#transaction}. Receive messages by querying the queue's `RECEIVE_` + # function with {#execute_query}. + # # See {Google::Cloud::Spanner::Project#client}. # # @example @@ -506,6 +512,20 @@ def directed_read_options # puts "User #{row[:id]} is #{row[:name]}" # end # + # @example Receive messages from a queue with the RECEIVE_ function: + # require "google/cloud/spanner" + # + # spanner = Google::Cloud::Spanner.new + # + # db = spanner.client "my-instance", "my-database" + # + # # RECEIVE_ needs a strong read, which is the default here. + # results = db.execute_query "SELECT Id, Payload FROM RECEIVE_OrderQueue()" + # + # results.rows.each do |row| + # puts "Received message #{row[:Id]}" + # end + # def execute_query sql, params: nil, types: nil, single_use: nil, query_options: nil, request_options: nil, call_options: nil, directed_read_options: nil diff --git a/google-cloud-spanner/lib/google/cloud/spanner/commit.rb b/google-cloud-spanner/lib/google/cloud/spanner/commit.rb index 29295180..6b71cde8 100644 --- a/google-cloud-spanner/lib/google/cloud/spanner/commit.rb +++ b/google-cloud-spanner/lib/google/cloud/spanner/commit.rb @@ -15,6 +15,8 @@ require "google/cloud/spanner/data" require "google/cloud/spanner/convert" +require "google/cloud/spanner/send_options" +require "google/cloud/spanner/ack_options" module Google module Cloud @@ -280,6 +282,130 @@ 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}): + # * `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 + # 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, StringIO.new("Order payload") + # end + # + # @example Deliver the message in five minutes + # require "google/cloud/spanner" + # + # spanner = Google::Cloud::Spanner.new + # + # db = spanner.client "my-instance", "my-database" + # + # send_options = { deliver_time: Time.now + 300 } + # + # db.commit do |c| + # c.send_message "OrderQueue", 101, StringIO.new("Order payload"), + # send_options: send_options + # end + # + def send_message queue, key, payload, send_options: nil + send_options = build_options send_options, SendOptions + 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(send_options.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 [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 + # 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" + # + # ack_options = { ignore_not_found: true } + # + # db.commit do |c| + # c.ack_message "OrderQueue", 101, ack_options: ack_options + # end + # + def ack_message queue, key, ack_options: nil + ack_options = build_options ack_options, AckOptions + ack_mutation = V1::Mutation::Ack.new( + queue: queue, + key: key_list_value(key), + ignore_not_found: ack_options.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 +451,42 @@ 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 single, or array of keys to convert. + # @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 + + ## + # @private + # Returns `options` as an instance of `options_class`, so callers can + # pass the options object, a hash of its keyword arguments, or `nil`. + # + # @param [Object, Hash, nil] options The options given by the caller. + # @param [Class] options_class The options class, e.g. {SendOptions}. + # @return [Object] An instance of `options_class`. + # @raise [ArgumentError] If `options` is any other type, or the hash + # has a key that `options_class` doesn't accept. + # + def build_options options, options_class + case options + when options_class then options + when nil then options_class.new + when Hash then options_class.new(**options) + else + raise ArgumentError, "expected #{options_class}, Hash, or nil, got #{options.class}" + end + 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/lib/google/cloud/spanner/send_options.rb b/google-cloud-spanner/lib/google/cloud/spanner/send_options.rb new file mode 100644 index 00000000..d0e380eb --- /dev/null +++ b/google-cloud-spanner/lib/google/cloud/spanner/send_options.rb @@ -0,0 +1,61 @@ +# 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. + + +module Google + module Cloud + module Spanner + ## + # # SendOptions + # + # Optional settings for sending a message to a queue with + # {Commit#send_message} or {Transaction#send_message}. A hash with the + # same keys can be passed instead. + # + # @example + # require "google/cloud/spanner" + # + # spanner = Google::Cloud::Spanner.new + # + # db = spanner.client "my-instance", "my-database" + # + # send_options = Google::Cloud::Spanner::SendOptions.new deliver_time: Time.now + 300 + # + # db.commit do |c| + # c.send_message "OrderQueue", 101, StringIO.new("Order payload"), + # send_options: send_options + # end + # + class SendOptions + ## + # The time at which Spanner begins trying to deliver the message. + # + # @return [Time, nil] + attr_reader :deliver_time + + ## + # Creates a new SendOptions. + # + # @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. + # + def initialize deliver_time: nil + @deliver_time = deliver_time + end + end + end + end +end 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/support/doctest_helper.rb b/google-cloud-spanner/support/doctest_helper.rb index 0e4a55f7..5afc2d2f 100644 --- a/google-cloud-spanner/support/doctest_helper.rb +++ b/google-cloud-spanner/support/doctest_helper.rb @@ -762,6 +762,19 @@ def verify_mock_params actual, expected end end + # SendOptions and AckOptions + + ["Google::Cloud::Spanner::SendOptions", "Google::Cloud::Spanner::AckOptions"].each do |options_class| + doctest.before options_class do + mock_spanner do |mock, mock_instances, mock_databases| + mock.expect :batch_create_sessions, OpenStruct.new(session: Array.new(10) { session_grpc }), [{database: "projects/my-project/instances/my-instance/databases/my-database", session_count: 10, session_template: nil }, Hash] + mock.expect :commit, commit_resp do |req| + verify_mock_params(req, session: "session-name", mutations: Array) + end + end + end + end + # Data doctest.before "Google::Cloud::Spanner::Data" do 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/commit_queues_test.rb b/google-cloud-spanner/test/google/cloud/spanner/commit_queues_test.rb new file mode 100644 index 00000000..0c58d42b --- /dev/null +++ b/google-cloud-spanner/test/google/cloud/spanner/commit_queues_test.rb @@ -0,0 +1,137 @@ +# 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 from SendOptions or a Hash" do + send_options = Google::Cloud::Spanner::SendOptions.new deliver_time: deliver_time + + commit.send_message "TestQueue", 1, "payload", send_options: send_options + commit.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], commit.mutations.map { |mutation| mutation["send"].deliver_time } + end + + it "raises ArgumentError for invalid send_options" do + assert_raises(ArgumentError) { commit.send_message "TestQueue", 1, "payload", send_options: { deliver_at: deliver_time } } + assert_raises(ArgumentError) { commit.send_message "TestQueue", 1, "payload", send_options: deliver_time } + assert_raises ArgumentError do + commit.send_message "TestQueue", 1, "payload", send_options: Google::Cloud::Spanner::AckOptions.new + end + 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 from AckOptions or a Hash" do + ack_options = Google::Cloud::Spanner::AckOptions.new ignore_not_found: true + + commit.ack_message "TestQueue", 1, ack_options: ack_options + commit.ack_message "TestQueue", 2, ack_options: { ignore_not_found: true } + + assert_equal [true, true], commit.mutations.map { |mutation| mutation.ack.ignore_not_found } + end + + it "raises ArgumentError for invalid ack_options" do + assert_raises(ArgumentError) { commit.ack_message "TestQueue", 1, ack_options: { ignore_missing: true } } + assert_raises(ArgumentError) { commit.ack_message "TestQueue", 1, ack_options: true } + assert_raises ArgumentError do + commit.ack_message "TestQueue", 1, ack_options: Google::Cloud::Spanner::SendOptions.new + end + 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 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