feat(pubsub): Add skip_lookup to Project#publisher to enable lazy loading (#32263)

docs(pubsub): Fix overview documentation URL

---------

Co-authored-by: Dmitry Lebed <@d-lebed>

Author: aandreassa

google-cloud-pubsub/OVERVIEW.md

@@ -36,7 +36,7 @@ into Admin Operations and Data Plane Operations.
 * **Data Plane Operations**: For the core functionality of publishing and receiving messages.
 
 To learn more about Pub/Sub, read the [Google Cloud Pub/Sub Overview
-](https://cloud.google.com/pubsub/overview).
+](https://docs.cloud.google.com/pubsub/docs/overview).
 
 ## Admin Operations
 

google-cloud-pubsub/lib/google/cloud/pubsub/project.rb

@@ -133,12 +133,17 @@ def iam
         # @param [Hash] async A hash of values to configure the topic's
         #   {AsyncPublisher} that is created when {Publisher#publish_async}
         #   is called. Optional.
+        # @param [Boolean] skip_lookup Optionally create a {Publisher} object
+        #   without verifying the topic resource exists on the Pub/Sub
+        #   service. Calls made on this object will raise errors if the service
+        #   resource does not exist. Default is `false`.
         #
         # @return [Google::Cloud::PubSub::Publisher]
         #
-        def publisher topic_name, project: nil, async: nil
+        def publisher topic_name, project: nil, async: nil, skip_lookup: nil
           ensure_service!
           options = { project: project, async: async }
+          return Publisher.from_name topic_name, service, options if skip_lookup
           grpc = topic_admin.get_topic topic: service.topic_path(topic_name, options)
           Publisher.from_grpc grpc, service, async: async
         end

google-cloud-pubsub/lib/google/cloud/pubsub/publisher.rb

@@ -87,9 +87,56 @@ def async_publisher
         #   `projects/{project_id}/topics/{topic_id}`.
         #
         def name
+          return @resource_name if reference?
           @grpc.name
         end
 
+        ##
+        # Determines whether the publisher object was created with a resource
+        # representation from the Pub/Sub service.
+        #
+        # @return [Boolean] `true` when the publisher was created with a resource
+        #   representation, `false` otherwise.
+        #
+        def reference?
+          @grpc.nil?
+        end
+
+        ##
+        # Determines whether the publisher object was created with a resource
+        # representation from the Pub/Sub service.
+        #
+        # @return [Boolean] `true` when the publisher was created with a resource
+        #   representation, `false` otherwise.
+        #
+        def resource?
+          !@grpc.nil?
+        end
+
+
+        ##
+        # Reloads the publisher with current data from the Pub/Sub service.
+        #
+        # @return [Google::Cloud::PubSub::Publisher] Returns the reloaded
+        #   publisher
+        #
+        # @example
+        #   require "google/cloud/pubsub"
+        #
+        #   pubsub = Google::Cloud::PubSub.new
+        #
+        #   publisher = pubsub.publisher "my-topic", skip_lookup: true
+        #
+        #   publisher.reload!
+        #
+        def reload!
+          ensure_service!
+          topic_path = service.topic_path name
+          @grpc = service.topic_admin.get_topic topic: topic_path
+          @resource_name = nil
+          self
+        end
+
         ##
         # Publishes one or more messages to the publisher.
         #
@@ -166,7 +213,7 @@ def name
         #
         def publish data = nil, attributes = nil, ordering_key: nil, compress: nil, compression_bytes_threshold: nil,
                     **extra_attrs, &block
-          ensure_service!
+          ensure_grpc!
           batch = BatchPublisher.new data,
                                      attributes,
                                      ordering_key,
@@ -290,8 +337,7 @@ def publish data = nil, attributes = nil, ordering_key: nil, compress: nil, comp
         #   publisher.async_publisher.stop!
         #
         def publish_async data = nil, attributes = nil, ordering_key: nil, **extra_attrs, &callback
-          ensure_service!
-
+          ensure_grpc!
           @async_publisher ||= AsyncPublisher.new name, service, **@async_opts
           @async_publisher.publish data, attributes, ordering_key: ordering_key, **extra_attrs, &callback
         end
@@ -351,6 +397,16 @@ def self.from_grpc grpc, service, async: nil
           end
         end
 
+        ##
+        # @private New reference {Publisher} from a topic name without making
+        # a HTTP request.
+        def self.from_name name, service, options = {}
+          name = service.topic_path name, options
+          from_grpc(nil, service, async: options[:async]).tap do |t|
+            t.instance_variable_set :@resource_name, name
+          end
+        end
+
         protected
 
         ##
@@ -364,6 +420,7 @@ def ensure_service!
         # Ensures a Google::Cloud::PubSub::V1::Topic object exists.
         def ensure_grpc!
           ensure_service!
+          reload! if reference?
         end
       end
     end

google-cloud-pubsub/test/google/cloud/pubsub/publisher/lazy_test.rb

@@ -0,0 +1,39 @@
+# Copyright 2015 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::PubSub::Publisher, :name, :mock_pubsub do
+  let(:topic_name) { "topic-name-goes-here" }
+  let(:full_topic_name) { topic_path topic_name }
+  let(:topic_grpc_hash) { topic_hash topic_name }
+  let(:topic_grpc) { Google::Cloud::PubSub::V1::Topic.new topic_grpc_hash }
+  let(:publisher) { Google::Cloud::PubSub::Publisher.from_grpc topic_grpc, pubsub.service }
+
+  it "is not reference when created with an HTTP method" do
+    _(publisher).wont_be :reference?
+    _(publisher).must_be :resource?
+  end
+
+  describe "reference publisher" do
+    let :publisher do
+      Google::Cloud::PubSub::Publisher.from_name topic_name, pubsub.service
+    end
+
+    it "is reference" do
+      _(publisher).must_be :reference?
+      _(publisher).wont_be :resource?
+    end
+  end
+end

google-cloud-pubsub/test/google/cloud/pubsub/publisher_test.rb

@@ -141,4 +141,49 @@
     _(msgs.last.message_id).must_equal "msg2"
     _(msgs.last.attributes["format"]).must_equal "none"
   end
-end
+
+  it "can be created with skip_lookup: true" do
+    mock = Minitest::Mock.new
+    pubsub.service.mocked_topic_admin = mock
+
+    publisher = pubsub.publisher topic_name, skip_lookup: true
+
+    _(publisher.name).must_equal topic_path(topic_name)
+    _(publisher).must_be :reference?
+
+    mock.verify
+  end
+
+  it "reloads the resource when publishing if created with skip_lookup: true" do
+    mock = Minitest::Mock.new
+    pubsub.service.mocked_topic_admin = mock
+
+    publisher = pubsub.publisher topic_name, skip_lookup: true
+
+    _(publisher).must_be :reference?
+
+    mock.expect :get_topic, Google::Cloud::PubSub::V1::Topic.new(topic_hash(topic_name)) do |actual_topic|
+      actual_topic == {topic: topic_path(topic_name)}
+    end
+
+    # Expect publish to be called
+    message = "new-message-here"
+    encoded_msg = message.encode(Encoding::ASCII_8BIT)
+    messages = [
+      Google::Cloud::PubSub::V1::PubsubMessage.new(data: encoded_msg)
+    ]
+    publish_res = Google::Cloud::PubSub::V1::PublishResponse.new({ message_ids: ["msg1"] })
+    expected_request = {topic: topic_path(topic_name), messages: messages}
+    
+    mock.expect :publish_internal, publish_res do |actual_request, actual_option|
+      actual_request == expected_request && actual_option.nil?
+    end
+
+    msg = publisher.publish message
+
+    _(publisher).wont_be :reference?
+    _(msg.message_id).must_equal "msg1"
+
+    mock.verify
+  end
+end