samples(pubsub): pubsub_optimistic_subscribe sample

cy-yun · cy-yun · commit 843376dc2c9a · 2025-08-05T14:14:35.000-07:00
diff --git a/google-cloud-pubsub/samples/acceptance/subscriptions_test.rb b/google-cloud-pubsub/samples/acceptance/subscriptions_test.rb
@@ -17,6 +17,7 @@
 require_relative "../pubsub_subscriber_exactly_once_delivery.rb"
 require_relative "../pubsub_create_subscription_with_exactly_once_delivery.rb"
 require_relative "../pubsub_create_bigquery_subscription.rb"
+require_relative "../pubsub_optimistic_subscribe.rb"
 require_relative "../pubsub_subscriber_async_pull_custom_attributes.rb"
 require_relative "../pubsub_subscriber_sync_pull.rb"
 require_relative "../pubsub_subscriber_flow_settings.rb"
@@ -281,4 +282,27 @@
       )
     end
   end 
+
+  it "supports pubsub_optimistic_subscribe" do
+    topic_id = @topic.name
+    subscription_id = random_subscription_id
+    publisher = pubsub.publisher @topic.name
+
+    out, _err = capture_io do
+      optimistic_subscribe topic_id: topic_id, subscription_id: subscription_id
+    end
+
+    assert_includes out, "Subscription #{subscription_id} does not exist."
+    assert_includes out, "Subscription #{subscription_id} created."
+
+    @created_subscriptions << subscription_id
+    publisher.publish "This is a test message."
+    sleep 5
+
+    out, _err = capture_io do
+      optimistic_subscribe topic_id: topic_id, subscription_id: subscription_id
+    end
+
+    assert_includes out, "Received message: This is a test message."
+  end
 end
diff --git a/google-cloud-pubsub/samples/pubsub_optimistic_subscribe.rb b/google-cloud-pubsub/samples/pubsub_optimistic_subscribe.rb
@@ -0,0 +1,49 @@
+# Copyright 2025 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
+#
+#     http://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 "google/cloud/pubsub"
+
+def optimistic_subscribe topic_id:, subscription_id:
+  # [START pubsub_optimistic_subscribe]
+  # topic_id = "your-topic-id"
+  # subscription_id = "your-subscription-id"
+
+  pubsub = Google::Cloud::Pubsub.new
+
+  # Propagate expection from child threads to the main thread as soon as it is
+  # raised. Exceptions happened in the callback thread are collected in the
+  # callback thread pool and do not propagate to the main thread
+  Thread.abort_on_exception = true
+
+  begin
+    subscriber = pubsub.subscriber subscription_id
+    listener   = subscriber.listen do |received_message|
+      puts "Received message: #{received_message.data}"
+      received_message.acknowledge!
+    end
+    listener.start
+    # Let the main thread sleep for 60 seconds so the thread for listening
+    # messages does not quit
+    sleep 60
+    listener.stop.wait!
+  rescue Google::Cloud::NotFoundError => e
+    puts "Subscription #{subscription_id} does not exist."
+    subscription_admin = pubsub.subscription_admin
+
+    subscription = subscription_admin.create_subscription name: pubsub.subscription_path(subscription_id),
+                                                          topic: pubsub.topic_path(topic_id)
+    puts "Subscription #{subscription_id} created."
+  end
+  # [END pubsub_optimistic_subscribe]
+end
