diff --git a/google-cloud-pubsub/README.md b/google-cloud-pubsub/README.md index e7217eb456f7..2e6916306dd3 100644 --- a/google-cloud-pubsub/README.md +++ b/google-cloud-pubsub/README.md @@ -69,6 +69,26 @@ listener.start sleep ``` +### Subscriber Shutdown Options + +You can configure how the subscriber handles unprocessed messages when stopping via the `shutdown_behavior` and `shutdown_timeout` options: + +* `:wait_for_processing` (default) – Waits for all received messages to be processed and acknowledged by your callback before completing shutdown. +* `:nack_immediately` – Immediately nacks all unprocessed messages back to Pub/Sub so other subscribers can pick them up quickly. + +```ruby +# Immediately nack unprocessed messages on shutdown: +listener = subscriber.listen shutdown_behavior: :nack_immediately do |msg| + process_message msg.data + msg.ack! +end + +# Gracefully shut down with a 30-second timeout: +listener = subscriber.listen shutdown_behavior: :wait_for_processing, shutdown_timeout: 30 do |msg| + process_message msg.data + msg.ack! +end +``` ## Enabling Logging diff --git a/google-cloud-pubsub/lib/google/cloud/pubsub/subscriber.rb b/google-cloud-pubsub/lib/google/cloud/pubsub/subscriber.rb index 64b7d20caa26..e5ba6ed32852 100644 --- a/google-cloud-pubsub/lib/google/cloud/pubsub/subscriber.rb +++ b/google-cloud-pubsub/lib/google/cloud/pubsub/subscriber.rb @@ -418,6 +418,42 @@ def wait_for_messages max: 100 # # Shut down the subscriber when ready to stop receiving messages. # listener.stop! # + # @example Immediately release unprocessed messages on shutdown: + # require "google/cloud/pubsub" + # + # pubsub = Google::Cloud::PubSub.new + # + # subscriber = pubsub.subscriber "my-topic-sub" + # + # listener = subscriber.listen shutdown_behavior: :nack_immediately do |received_message| + # # process message + # puts "Data: #{received_message.message.data}" + # received_message.acknowledge! + # end + # + # listener.start + # + # # Shut down immediately, releasing (nacking) any unprocessed messages + # listener.stop! + # + # @example Gracefully shut down with a custom timeout: + # require "google/cloud/pubsub" + # + # pubsub = Google::Cloud::PubSub.new + # + # subscriber = pubsub.subscriber "my-topic-sub" + # + # listener = subscriber.listen shutdown_behavior: :wait_for_processing, shutdown_timeout: 30 do |received_message| + # # process message + # puts "Data: #{received_message.message.data}" + # received_message.acknowledge! + # end + # + # listener.start + # + # # Wait up to 30 seconds for in-flight callbacks before forcing remaining to nack + # listener.stop! + # def listen deadline: nil, message_ordering: nil, streams: nil, inventory: nil, threads: {}, shutdown_behavior: :wait_for_processing, shutdown_timeout: nil, &block ensure_service!