From 6ff5b2f6ba6c77f8226feaadb584a44420e73f5c Mon Sep 17 00:00:00 2001 From: Dan Lee Date: Tue, 2 Aug 2022 18:31:20 +0000 Subject: [PATCH 1/2] docs: reorganize sphinx structure --- docs/index.rst | 8 +- docs/pubsub/publisher/api/client.rst | 6 + docs/pubsub/publisher/api/futures.rst | 6 + docs/pubsub/publisher/api/pagers.rst | 6 + docs/pubsub/publisher/index.rst | 187 ++++++++++++++++++ docs/pubsub/subscriber/api/client.rst | 6 + docs/pubsub/subscriber/api/futures.rst | 6 + docs/pubsub/subscriber/api/message.rst | 6 + docs/pubsub/subscriber/api/pagers.rst | 6 + docs/pubsub/subscriber/api/scheduler.rst | 6 + docs/pubsub/subscriber/index.rst | 236 +++++++++++++++++++++++ docs/pubsub/types.rst | 6 + 12 files changed, 481 insertions(+), 4 deletions(-) create mode 100644 docs/pubsub/publisher/api/client.rst create mode 100644 docs/pubsub/publisher/api/futures.rst create mode 100644 docs/pubsub/publisher/api/pagers.rst create mode 100644 docs/pubsub/publisher/index.rst create mode 100644 docs/pubsub/subscriber/api/client.rst create mode 100644 docs/pubsub/subscriber/api/futures.rst create mode 100644 docs/pubsub/subscriber/api/message.rst create mode 100644 docs/pubsub/subscriber/api/pagers.rst create mode 100644 docs/pubsub/subscriber/api/scheduler.rst create mode 100644 docs/pubsub/subscriber/index.rst create mode 100644 docs/pubsub/types.rst diff --git a/docs/index.rst b/docs/index.rst index 06b09605f..6b3e1583b 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -13,11 +13,11 @@ API Documentation across the documentation. .. toctree:: - :maxdepth: 3 + :maxdepth: 4 - Publisher Client - Subscriber Client - Types + Publisher Client + Subscriber Client + Types Migration Guide diff --git a/docs/pubsub/publisher/api/client.rst b/docs/pubsub/publisher/api/client.rst new file mode 100644 index 000000000..d1a54ff5e --- /dev/null +++ b/docs/pubsub/publisher/api/client.rst @@ -0,0 +1,6 @@ +Publisher Client API (v1) +========================= + +.. automodule:: google.cloud.pubsub_v1.publisher.client + :members: + :inherited-members: diff --git a/docs/pubsub/publisher/api/futures.rst b/docs/pubsub/publisher/api/futures.rst new file mode 100644 index 000000000..b02b9bf90 --- /dev/null +++ b/docs/pubsub/publisher/api/futures.rst @@ -0,0 +1,6 @@ +Futures +======= + +.. automodule:: google.cloud.pubsub_v1.publisher.futures + :members: + :inherited-members: diff --git a/docs/pubsub/publisher/api/pagers.rst b/docs/pubsub/publisher/api/pagers.rst new file mode 100644 index 000000000..3bbfff33c --- /dev/null +++ b/docs/pubsub/publisher/api/pagers.rst @@ -0,0 +1,6 @@ +Pagers +====== + +.. automodule:: google.pubsub_v1.services.publisher.pagers + :members: + :inherited-members: diff --git a/docs/pubsub/publisher/index.rst b/docs/pubsub/publisher/index.rst new file mode 100644 index 000000000..2a0ad320e --- /dev/null +++ b/docs/pubsub/publisher/index.rst @@ -0,0 +1,187 @@ +Publishing Messages +=================== + +Publishing messages is handled through the +:class:`~.pubsub_v1.publisher.client.Client` class (aliased as +``google.cloud.pubsub.PublisherClient``). This class provides methods to +create topics, and (most importantly) a +:meth:`~.pubsub_v1.publisher.client.Client.publish` method that publishes +messages to Pub/Sub. + +Instantiating a publishing client is straightforward: + +.. code-block:: python + + from google.cloud import pubsub + publish_client = pubsub.PublisherClient() + + +Publish a Message +----------------- + +To publish a message, use the +:meth:`~.pubsub_v1.publisher.client.Client.publish` method. This method accepts +two positional arguments: the topic to publish to, and the body of the message. +It also accepts arbitrary keyword arguments, which are passed along as +attributes of the message. + +The topic is passed along as a string; all topics have the canonical form of +``projects/{project_name}/topics/{topic_name}``. + +Therefore, a very basic publishing call looks like: + +.. code-block:: python + + topic = 'projects/{project}/topics/{topic}' + future = publish_client.publish(topic, b'This is my message.') + +.. note:: + + The message data in Pub/Sub is an opaque blob of bytes, and as such, you + *must* send a ``bytes`` object in Python 3 (``str`` object in Python 2). + If you send a text string (``str`` in Python 3, ``unicode`` in Python 2), + the method will raise :exc:`TypeError`. + + The reason it works this way is because there is no reasonable guarantee + that the same language or environment is being used by the subscriber, + and so it is the responsibility of the publisher to properly encode + the payload. + +If you want to include attributes, simply add keyword arguments: + +.. code-block:: python + + topic = 'projects/{project}/topics/{topic}' + future = publish_client.publish(topic, b'This is my message.', foo='bar') + + +Batching +-------- + +Whenever you publish a message, the publisher will automatically batch the +messages over a small time window to avoid making too many separate requests to +the service. This helps increase throughput. + +.. note:: + + By default, this uses ``threading``, and you will need to be in an + environment with threading enabled. It is possible to provide an + alternative batch class that uses another concurrency strategy. + +The way that this works is that on the first message that you send, a new batch +is created automatically. For every subsequent message, if there is already a +valid batch that is still accepting messages, then that batch is used. When the +batch is created, it begins a countdown that publishes the batch once +sufficient time has elapsed (by default, this is 0.01 seconds). + +If you need different batching settings, simply provide a +:class:`~.pubsub_v1.types.BatchSettings` object when you instantiate the +:class:`~.pubsub_v1.publisher.client.Client`: + +.. code-block:: python + + from google.cloud import pubsub + from google.cloud.pubsub import types + + client = pubsub.PublisherClient( + batch_settings=types.BatchSettings( + max_messages=500, # default 100 + max_bytes=1024, # default 1 MB + max_latency=1 # default .01 seconds + ), + ) + +The `max_bytes` argument is the maximum total size of the messages to collect +before automatically publishing the batch, (in bytes) including any byte size +overhead of the publish request itself. The maximum value is bound by the +server-side limit of 10_000_000 bytes. The default value is 1 MB. + +The `max_messages` argument is the maximum number of messages to collect +before automatically publishing the batch, the default value is 100 messages. + +The `max_latency` is the maximum number of seconds to wait for additional +messages before automatically publishing the batch, the default is .01 seconds. + + +Futures +------- + +Every call to :meth:`~.pubsub_v1.publisher.client.Client.publish` returns +an instance of :class:`~.pubsub_v1.publisher.futures.Future`. + +.. note:: + + The returned future conforms for the most part to the interface of + the standard library's :class:`~concurrent.futures.Future`, but might not + be usable in all cases which expect that exact implementaton. + +You can use this to ensure that the publish succeeded: + +.. code-block:: python + + # The .result() method will block until the future is complete. + # If there is an error, it will raise an exception. + future = client.publish(topic, b'My awesome message.') + message_id = future.result() + +You can also attach a callback to the future: + +.. code-block:: python + + # Callbacks receive the future as their only argument, as defined in + # the Future interface. + def callback(future): + message_id = future.result() + do_something_with(message_id) + + # The callback is added once you get the future. If you add a callback + # and the future is already done, it will simply be executed immediately. + future = client.publish(topic, b'My awesome message.') + future.add_done_callback(callback) + + +Publish Flow Control +-------------------- + +If publishing large amounts of messages or very large messages in quick +succession, some of the publish requests might time out, especially if the +bandwidth available is limited. To mitigate this the client can be +configured with custom :class:`~.pubsub_v1.types.PublishFlowControl` settings. + +You can configure the maximum desired number of messages and their maximum total +size, as well as the action that should be taken when the threshold is reached. + +.. code-block:: python + + from google.cloud import pubsub_v1 + + client = pubsub_v1.PublisherClient( + publisher_options=pubsub_v1.types.PublisherOptions( + flow_control=pubsub_v1.types.PublishFlowControl( + message_limit=500, + byte_limit=2 * 1024 * 1024, + limit_exceeded_behavior=pubsub_v1.types.LimitExceededBehavior.BLOCK, + ), + ), + ) + +The action to be taken on overflow can be one of the following: + +* :attr:`~.pubsub_v1.types.LimitExceededBehavior.IGNORE` (default): Ignore the + overflow and continue publishing the messages as normal. +* :attr:`~.pubsub_v1.types.LimitExceededBehavior.ERROR`: Raise + :exc:`~.pubsub_v1.publisher.exceptions.FlowControlLimitError` and reject the message. +* :attr:`~.pubsub_v1.types.LimitExceededBehavior.BLOCK`: Temporarily block in the + :meth:`~.pubsub_v1.publisher.client.Client.publish` method until there is + enough capacity available. + + +API Reference +------------- + +.. toctree:: + :maxdepth: 2 + + api/client + api/futures + api/pagers diff --git a/docs/pubsub/subscriber/api/client.rst b/docs/pubsub/subscriber/api/client.rst new file mode 100644 index 000000000..d26243eba --- /dev/null +++ b/docs/pubsub/subscriber/api/client.rst @@ -0,0 +1,6 @@ +Subscriber Client API (v1) +========================== + +.. automodule:: google.cloud.pubsub_v1.subscriber.client + :members: + :inherited-members: diff --git a/docs/pubsub/subscriber/api/futures.rst b/docs/pubsub/subscriber/api/futures.rst new file mode 100644 index 000000000..fb0264279 --- /dev/null +++ b/docs/pubsub/subscriber/api/futures.rst @@ -0,0 +1,6 @@ +Futures +======= + +.. automodule:: google.cloud.pubsub_v1.subscriber.futures + :members: + :inherited-members: diff --git a/docs/pubsub/subscriber/api/message.rst b/docs/pubsub/subscriber/api/message.rst new file mode 100644 index 000000000..6e7a55ded --- /dev/null +++ b/docs/pubsub/subscriber/api/message.rst @@ -0,0 +1,6 @@ +Messages +======== + +.. autoclass:: google.cloud.pubsub_v1.subscriber.message.Message + :members: + :noindex: diff --git a/docs/pubsub/subscriber/api/pagers.rst b/docs/pubsub/subscriber/api/pagers.rst new file mode 100644 index 000000000..367c65ca7 --- /dev/null +++ b/docs/pubsub/subscriber/api/pagers.rst @@ -0,0 +1,6 @@ +Pagers +====== + +.. automodule:: google.pubsub_v1.services.subscriber.pagers + :members: + :inherited-members: diff --git a/docs/pubsub/subscriber/api/scheduler.rst b/docs/pubsub/subscriber/api/scheduler.rst new file mode 100644 index 000000000..06e839f21 --- /dev/null +++ b/docs/pubsub/subscriber/api/scheduler.rst @@ -0,0 +1,6 @@ +Scheduler +========= + +.. automodule:: google.cloud.pubsub_v1.subscriber.scheduler + :members: + :inherited-members: diff --git a/docs/pubsub/subscriber/index.rst b/docs/pubsub/subscriber/index.rst new file mode 100644 index 000000000..aa21cd37b --- /dev/null +++ b/docs/pubsub/subscriber/index.rst @@ -0,0 +1,236 @@ +Subscribing to Messages +======================= + +Subscribing to messages is handled through the +:class:`~.pubsub_v1.subscriber.client.Client` class (aliased as +``google.cloud.pubsub.SubscriberClient``). This class provides a +:meth:`~.pubsub_v1.subscriber.client.Client.subscribe` method to +attach to subscriptions on existing topics. + +Instantiating a subscriber client is straightforward: + +.. code-block:: python + + from google.cloud import pubsub + + with pubsub.SubscriberClient() as subscriber: + # ... + +Creating a Subscription +----------------------- + +In Pub/Sub, a **subscription** is a discrete pull of messages from a topic. +If multiple clients pull the same subscription, then messages are split +between them. If multiple clients create a subscription each, then each client +will get every message. + +.. note:: + + Remember that Pub/Sub operates under the principle of "everything at least + once". Even in the case where multiple clients pull the same subscription, + *some* redundancy is likely. + +Creating a subscription requires that you already know what topic you want +to subscribe to, and it must already exist. Once you have that, it is easy: + +.. code-block:: python + + # Substitute PROJECT, SUBSCRIPTION, and TOPIC with appropriate values for + # your application. + + # from google.cloud import pubsub + # publisher = pubsub.PublisherClient() + + topic_path = publisher.topic_path(PROJECT, TOPIC) + + with pubsub.SubscriberClient() as subscriber: + sub_path = subscriber.subscription_path(PROJECT, SUBSCRIPTION) + subscriber.create_subscription(request={"name": sub_path, "topic": topic_path}) + +Once you have created a subscription (or if you already had one), the next +step is to pull data from it. + + +Pulling a Subscription Synchronously +------------------------------------ + +To pull the messages synchronously, use the client's +:meth:`~.pubsub_v1.subscriber.client.Client.pull` method. + +.. code-block:: python + + # Wrap the following code in `with pubsub.SubscriberClient() as subscriber:` + + # Substitute PROJECT and SUBSCRIPTION with appropriate values for your + # application. + subscription_path = subscriber.subscription_path(PROJECT, SUBSCRIPTION) + response = subscriber.pull( + request={ + "subscription": subscription_path, + "max_messages": 5, + } + ) + + for msg in response.received_messages: + print("Received message:", msg.message.data) + + ack_ids = [msg.ack_id for msg in response.received_messages] + subscriber.acknowledge( + request={ + "subscription": subscription_path, + "ack_ids": ack_ids, + } + ) + +The method returns a :class:`~.pubsub_v1.types.PullResponse` instance that +contains a list of received :class:`~.pubsub_v1.types.ReceivedMessage` +instances. + +If you want to **nack** some of the received messages (see :ref:`explaining-ack` below), +you can use the :meth:`~.pubsub_v1.subscriber.client.Client.modify_ack_deadline` +method and set their acknowledge deadlines to zero. This will cause them to +be dropped by this client and the backend will try to re-deliver them. + +.. code-block:: python + + # Wrap the following code in `with pubsub.SubscriberClient() as subscriber:` + + ack_ids = [] # TODO: populate with `ack_ids` of the messages to NACK + ack_deadline_seconds = 0 + subscriber.modify_ack_deadline( + request={ + "subscription": subscription_path, + "ack_ids": ack_ids, + "ack_deadline_seconds": ack_deadline_seconds, + } + ) + + +Pulling a Subscription Asynchronously +------------------------------------- + +The subscriber client uses the +:meth:`~.pubsub_v1.subscriber.client.Client.subscribe` method to start a +background thread to receive messages from Pub/Sub and calls a callback with +each message received. + +.. code-block:: python + + # Wrap the following code in `with pubsub.SubscriberClient() as subscriber:` + + # Substitute PROJECT and SUBSCRIPTION with appropriate values for your + # application. + subscription_path = subscriber.subscription_path(PROJECT, SUBSCRIPTION) + future = subscriber.subscribe(subscription_path, callback) + +This will return a +:class:`~.pubsub_v1.subscriber.futures.StreamingPullFuture`. This future allows +you to control the background thread that is managing the subscription. + + +Subscription Callbacks +---------------------- + +Messages received from a subscription are processed asynchronously through +**callbacks**. + +The basic idea: Define a function that takes one argument; this argument +will be a :class:`~.pubsub_v1.subscriber.message.Message` instance, which is +a convenience wrapper around the :class:`~.pubsub_v1.types.PubsubMessage` +instance received from the server (and stored under the ``message`` attribute). + +This function should do whatever processing is necessary. At the end, the +function should either :meth:`~.pubsub_v1.subscriber.message.Message.ack` +or :meth:`~.pubsub_v1.subscriber.message.Message.nack` the message. + +When you call :meth:`~.pubsub_v1.subscriber.client.Client.subscribe`, you +must pass the callback that will be used. + +Here is an example: + +.. code-block:: python + + # Define the callback. + # Note that the callback is defined *before* the subscription is opened. + def callback(message): + do_something_with(message) # Replace this with your actual logic. + message.ack() # Asynchronously acknowledge the message. + + # Wrap the following code in `with pubsub.SubscriberClient() as subscriber:` + + # Substitute PROJECT and SUBSCRIPTION with appropriate values for your + # application. + subscription_path = subscriber.subscription_path(PROJECT, SUBSCRIPTION) + + # Open the subscription, passing the callback. + future = subscriber.subscribe(subscription_path, callback) + +The :meth:`~.pubsub_v1.subscriber.client.Client.subscribe` method returns +a :class:`~.pubsub_v1.subscriber.futures.StreamingPullFuture`, which is both +the interface to wait on messages (e.g. block the primary thread) and to +address exceptions. + +To block the thread you are in while messages are coming in the stream, +use the :meth:`~.pubsub_v1.subscriber.futures.Future.result` method: + +.. code-block:: python + + future.result() + +.. note: This will block forever assuming no errors or that ``cancel`` is never + called. + +You can also use this for error handling; any exceptions that crop up on a +thread will be set on the future. + +.. code-block:: python + + try: + future.result() + except Exception as ex: + # Close the subscriber if not using a context manager. + subscriber.close() + raise + +Finally, you can use +:meth:`~.pubsub_v1.subscriber.futures.StreamingPullFuture.cancel` to stop +receiving messages. + + +.. code-block:: python + + future.cancel() + + +.. _explaining-ack: + +Explaining Ack +-------------- + +In Pub/Sub, the term **ack** stands for "acknowledge". You should ack a +message when your processing of that message *has completed*. When you ack +a message, you are telling Pub/Sub that you do not need to see it again. + +It might be tempting to ack messages immediately on receipt. While there +are valid use cases for this, in general it is unwise. The reason why: If +there is some error or edge case in your processing logic, and processing +of the message fails, you will have already told Pub/Sub that you successfully +processed the message. By contrast, if you ack only upon completion, then +Pub/Sub will eventually re-deliver the unacknowledged message. + +It is also possible to **nack** a message, which is the opposite. When you +nack, it tells Pub/Sub that you are unable or unwilling to deal with the +message, and that the service should redeliver it. + + +API Reference +------------- + +.. toctree:: + :maxdepth: 2 + + api/client + api/message + api/futures + api/pagers + api/scheduler diff --git a/docs/pubsub/types.rst b/docs/pubsub/types.rst new file mode 100644 index 000000000..308f66971 --- /dev/null +++ b/docs/pubsub/types.rst @@ -0,0 +1,6 @@ +Pub/Sub Client Types +==================== + +.. automodule:: google.cloud.pubsub_v1.types + :members: + :noindex: From ffdd9f446642a23c6b830509197b5735045ff40b Mon Sep 17 00:00:00 2001 From: Dan Lee Date: Tue, 2 Aug 2022 19:13:42 +0000 Subject: [PATCH 2/2] docs: update directory --- docs/publisher/api/client.rst | 6 - docs/publisher/api/futures.rst | 6 - docs/publisher/api/pagers.rst | 6 - docs/publisher/index.rst | 187 ----------------------- docs/subscriber/api/client.rst | 6 - docs/subscriber/api/futures.rst | 6 - docs/subscriber/api/message.rst | 6 - docs/subscriber/api/pagers.rst | 6 - docs/subscriber/api/scheduler.rst | 6 - docs/subscriber/index.rst | 236 ------------------------------ docs/types.rst | 6 - 11 files changed, 477 deletions(-) delete mode 100644 docs/publisher/api/client.rst delete mode 100644 docs/publisher/api/futures.rst delete mode 100644 docs/publisher/api/pagers.rst delete mode 100644 docs/publisher/index.rst delete mode 100644 docs/subscriber/api/client.rst delete mode 100644 docs/subscriber/api/futures.rst delete mode 100644 docs/subscriber/api/message.rst delete mode 100644 docs/subscriber/api/pagers.rst delete mode 100644 docs/subscriber/api/scheduler.rst delete mode 100644 docs/subscriber/index.rst delete mode 100644 docs/types.rst diff --git a/docs/publisher/api/client.rst b/docs/publisher/api/client.rst deleted file mode 100644 index d1a54ff5e..000000000 --- a/docs/publisher/api/client.rst +++ /dev/null @@ -1,6 +0,0 @@ -Publisher Client API (v1) -========================= - -.. automodule:: google.cloud.pubsub_v1.publisher.client - :members: - :inherited-members: diff --git a/docs/publisher/api/futures.rst b/docs/publisher/api/futures.rst deleted file mode 100644 index b02b9bf90..000000000 --- a/docs/publisher/api/futures.rst +++ /dev/null @@ -1,6 +0,0 @@ -Futures -======= - -.. automodule:: google.cloud.pubsub_v1.publisher.futures - :members: - :inherited-members: diff --git a/docs/publisher/api/pagers.rst b/docs/publisher/api/pagers.rst deleted file mode 100644 index 3bbfff33c..000000000 --- a/docs/publisher/api/pagers.rst +++ /dev/null @@ -1,6 +0,0 @@ -Pagers -====== - -.. automodule:: google.pubsub_v1.services.publisher.pagers - :members: - :inherited-members: diff --git a/docs/publisher/index.rst b/docs/publisher/index.rst deleted file mode 100644 index 2a0ad320e..000000000 --- a/docs/publisher/index.rst +++ /dev/null @@ -1,187 +0,0 @@ -Publishing Messages -=================== - -Publishing messages is handled through the -:class:`~.pubsub_v1.publisher.client.Client` class (aliased as -``google.cloud.pubsub.PublisherClient``). This class provides methods to -create topics, and (most importantly) a -:meth:`~.pubsub_v1.publisher.client.Client.publish` method that publishes -messages to Pub/Sub. - -Instantiating a publishing client is straightforward: - -.. code-block:: python - - from google.cloud import pubsub - publish_client = pubsub.PublisherClient() - - -Publish a Message ------------------ - -To publish a message, use the -:meth:`~.pubsub_v1.publisher.client.Client.publish` method. This method accepts -two positional arguments: the topic to publish to, and the body of the message. -It also accepts arbitrary keyword arguments, which are passed along as -attributes of the message. - -The topic is passed along as a string; all topics have the canonical form of -``projects/{project_name}/topics/{topic_name}``. - -Therefore, a very basic publishing call looks like: - -.. code-block:: python - - topic = 'projects/{project}/topics/{topic}' - future = publish_client.publish(topic, b'This is my message.') - -.. note:: - - The message data in Pub/Sub is an opaque blob of bytes, and as such, you - *must* send a ``bytes`` object in Python 3 (``str`` object in Python 2). - If you send a text string (``str`` in Python 3, ``unicode`` in Python 2), - the method will raise :exc:`TypeError`. - - The reason it works this way is because there is no reasonable guarantee - that the same language or environment is being used by the subscriber, - and so it is the responsibility of the publisher to properly encode - the payload. - -If you want to include attributes, simply add keyword arguments: - -.. code-block:: python - - topic = 'projects/{project}/topics/{topic}' - future = publish_client.publish(topic, b'This is my message.', foo='bar') - - -Batching --------- - -Whenever you publish a message, the publisher will automatically batch the -messages over a small time window to avoid making too many separate requests to -the service. This helps increase throughput. - -.. note:: - - By default, this uses ``threading``, and you will need to be in an - environment with threading enabled. It is possible to provide an - alternative batch class that uses another concurrency strategy. - -The way that this works is that on the first message that you send, a new batch -is created automatically. For every subsequent message, if there is already a -valid batch that is still accepting messages, then that batch is used. When the -batch is created, it begins a countdown that publishes the batch once -sufficient time has elapsed (by default, this is 0.01 seconds). - -If you need different batching settings, simply provide a -:class:`~.pubsub_v1.types.BatchSettings` object when you instantiate the -:class:`~.pubsub_v1.publisher.client.Client`: - -.. code-block:: python - - from google.cloud import pubsub - from google.cloud.pubsub import types - - client = pubsub.PublisherClient( - batch_settings=types.BatchSettings( - max_messages=500, # default 100 - max_bytes=1024, # default 1 MB - max_latency=1 # default .01 seconds - ), - ) - -The `max_bytes` argument is the maximum total size of the messages to collect -before automatically publishing the batch, (in bytes) including any byte size -overhead of the publish request itself. The maximum value is bound by the -server-side limit of 10_000_000 bytes. The default value is 1 MB. - -The `max_messages` argument is the maximum number of messages to collect -before automatically publishing the batch, the default value is 100 messages. - -The `max_latency` is the maximum number of seconds to wait for additional -messages before automatically publishing the batch, the default is .01 seconds. - - -Futures -------- - -Every call to :meth:`~.pubsub_v1.publisher.client.Client.publish` returns -an instance of :class:`~.pubsub_v1.publisher.futures.Future`. - -.. note:: - - The returned future conforms for the most part to the interface of - the standard library's :class:`~concurrent.futures.Future`, but might not - be usable in all cases which expect that exact implementaton. - -You can use this to ensure that the publish succeeded: - -.. code-block:: python - - # The .result() method will block until the future is complete. - # If there is an error, it will raise an exception. - future = client.publish(topic, b'My awesome message.') - message_id = future.result() - -You can also attach a callback to the future: - -.. code-block:: python - - # Callbacks receive the future as their only argument, as defined in - # the Future interface. - def callback(future): - message_id = future.result() - do_something_with(message_id) - - # The callback is added once you get the future. If you add a callback - # and the future is already done, it will simply be executed immediately. - future = client.publish(topic, b'My awesome message.') - future.add_done_callback(callback) - - -Publish Flow Control --------------------- - -If publishing large amounts of messages or very large messages in quick -succession, some of the publish requests might time out, especially if the -bandwidth available is limited. To mitigate this the client can be -configured with custom :class:`~.pubsub_v1.types.PublishFlowControl` settings. - -You can configure the maximum desired number of messages and their maximum total -size, as well as the action that should be taken when the threshold is reached. - -.. code-block:: python - - from google.cloud import pubsub_v1 - - client = pubsub_v1.PublisherClient( - publisher_options=pubsub_v1.types.PublisherOptions( - flow_control=pubsub_v1.types.PublishFlowControl( - message_limit=500, - byte_limit=2 * 1024 * 1024, - limit_exceeded_behavior=pubsub_v1.types.LimitExceededBehavior.BLOCK, - ), - ), - ) - -The action to be taken on overflow can be one of the following: - -* :attr:`~.pubsub_v1.types.LimitExceededBehavior.IGNORE` (default): Ignore the - overflow and continue publishing the messages as normal. -* :attr:`~.pubsub_v1.types.LimitExceededBehavior.ERROR`: Raise - :exc:`~.pubsub_v1.publisher.exceptions.FlowControlLimitError` and reject the message. -* :attr:`~.pubsub_v1.types.LimitExceededBehavior.BLOCK`: Temporarily block in the - :meth:`~.pubsub_v1.publisher.client.Client.publish` method until there is - enough capacity available. - - -API Reference -------------- - -.. toctree:: - :maxdepth: 2 - - api/client - api/futures - api/pagers diff --git a/docs/subscriber/api/client.rst b/docs/subscriber/api/client.rst deleted file mode 100644 index d26243eba..000000000 --- a/docs/subscriber/api/client.rst +++ /dev/null @@ -1,6 +0,0 @@ -Subscriber Client API (v1) -========================== - -.. automodule:: google.cloud.pubsub_v1.subscriber.client - :members: - :inherited-members: diff --git a/docs/subscriber/api/futures.rst b/docs/subscriber/api/futures.rst deleted file mode 100644 index fb0264279..000000000 --- a/docs/subscriber/api/futures.rst +++ /dev/null @@ -1,6 +0,0 @@ -Futures -======= - -.. automodule:: google.cloud.pubsub_v1.subscriber.futures - :members: - :inherited-members: diff --git a/docs/subscriber/api/message.rst b/docs/subscriber/api/message.rst deleted file mode 100644 index 6e7a55ded..000000000 --- a/docs/subscriber/api/message.rst +++ /dev/null @@ -1,6 +0,0 @@ -Messages -======== - -.. autoclass:: google.cloud.pubsub_v1.subscriber.message.Message - :members: - :noindex: diff --git a/docs/subscriber/api/pagers.rst b/docs/subscriber/api/pagers.rst deleted file mode 100644 index 367c65ca7..000000000 --- a/docs/subscriber/api/pagers.rst +++ /dev/null @@ -1,6 +0,0 @@ -Pagers -====== - -.. automodule:: google.pubsub_v1.services.subscriber.pagers - :members: - :inherited-members: diff --git a/docs/subscriber/api/scheduler.rst b/docs/subscriber/api/scheduler.rst deleted file mode 100644 index 06e839f21..000000000 --- a/docs/subscriber/api/scheduler.rst +++ /dev/null @@ -1,6 +0,0 @@ -Scheduler -========= - -.. automodule:: google.cloud.pubsub_v1.subscriber.scheduler - :members: - :inherited-members: diff --git a/docs/subscriber/index.rst b/docs/subscriber/index.rst deleted file mode 100644 index aa21cd37b..000000000 --- a/docs/subscriber/index.rst +++ /dev/null @@ -1,236 +0,0 @@ -Subscribing to Messages -======================= - -Subscribing to messages is handled through the -:class:`~.pubsub_v1.subscriber.client.Client` class (aliased as -``google.cloud.pubsub.SubscriberClient``). This class provides a -:meth:`~.pubsub_v1.subscriber.client.Client.subscribe` method to -attach to subscriptions on existing topics. - -Instantiating a subscriber client is straightforward: - -.. code-block:: python - - from google.cloud import pubsub - - with pubsub.SubscriberClient() as subscriber: - # ... - -Creating a Subscription ------------------------ - -In Pub/Sub, a **subscription** is a discrete pull of messages from a topic. -If multiple clients pull the same subscription, then messages are split -between them. If multiple clients create a subscription each, then each client -will get every message. - -.. note:: - - Remember that Pub/Sub operates under the principle of "everything at least - once". Even in the case where multiple clients pull the same subscription, - *some* redundancy is likely. - -Creating a subscription requires that you already know what topic you want -to subscribe to, and it must already exist. Once you have that, it is easy: - -.. code-block:: python - - # Substitute PROJECT, SUBSCRIPTION, and TOPIC with appropriate values for - # your application. - - # from google.cloud import pubsub - # publisher = pubsub.PublisherClient() - - topic_path = publisher.topic_path(PROJECT, TOPIC) - - with pubsub.SubscriberClient() as subscriber: - sub_path = subscriber.subscription_path(PROJECT, SUBSCRIPTION) - subscriber.create_subscription(request={"name": sub_path, "topic": topic_path}) - -Once you have created a subscription (or if you already had one), the next -step is to pull data from it. - - -Pulling a Subscription Synchronously ------------------------------------- - -To pull the messages synchronously, use the client's -:meth:`~.pubsub_v1.subscriber.client.Client.pull` method. - -.. code-block:: python - - # Wrap the following code in `with pubsub.SubscriberClient() as subscriber:` - - # Substitute PROJECT and SUBSCRIPTION with appropriate values for your - # application. - subscription_path = subscriber.subscription_path(PROJECT, SUBSCRIPTION) - response = subscriber.pull( - request={ - "subscription": subscription_path, - "max_messages": 5, - } - ) - - for msg in response.received_messages: - print("Received message:", msg.message.data) - - ack_ids = [msg.ack_id for msg in response.received_messages] - subscriber.acknowledge( - request={ - "subscription": subscription_path, - "ack_ids": ack_ids, - } - ) - -The method returns a :class:`~.pubsub_v1.types.PullResponse` instance that -contains a list of received :class:`~.pubsub_v1.types.ReceivedMessage` -instances. - -If you want to **nack** some of the received messages (see :ref:`explaining-ack` below), -you can use the :meth:`~.pubsub_v1.subscriber.client.Client.modify_ack_deadline` -method and set their acknowledge deadlines to zero. This will cause them to -be dropped by this client and the backend will try to re-deliver them. - -.. code-block:: python - - # Wrap the following code in `with pubsub.SubscriberClient() as subscriber:` - - ack_ids = [] # TODO: populate with `ack_ids` of the messages to NACK - ack_deadline_seconds = 0 - subscriber.modify_ack_deadline( - request={ - "subscription": subscription_path, - "ack_ids": ack_ids, - "ack_deadline_seconds": ack_deadline_seconds, - } - ) - - -Pulling a Subscription Asynchronously -------------------------------------- - -The subscriber client uses the -:meth:`~.pubsub_v1.subscriber.client.Client.subscribe` method to start a -background thread to receive messages from Pub/Sub and calls a callback with -each message received. - -.. code-block:: python - - # Wrap the following code in `with pubsub.SubscriberClient() as subscriber:` - - # Substitute PROJECT and SUBSCRIPTION with appropriate values for your - # application. - subscription_path = subscriber.subscription_path(PROJECT, SUBSCRIPTION) - future = subscriber.subscribe(subscription_path, callback) - -This will return a -:class:`~.pubsub_v1.subscriber.futures.StreamingPullFuture`. This future allows -you to control the background thread that is managing the subscription. - - -Subscription Callbacks ----------------------- - -Messages received from a subscription are processed asynchronously through -**callbacks**. - -The basic idea: Define a function that takes one argument; this argument -will be a :class:`~.pubsub_v1.subscriber.message.Message` instance, which is -a convenience wrapper around the :class:`~.pubsub_v1.types.PubsubMessage` -instance received from the server (and stored under the ``message`` attribute). - -This function should do whatever processing is necessary. At the end, the -function should either :meth:`~.pubsub_v1.subscriber.message.Message.ack` -or :meth:`~.pubsub_v1.subscriber.message.Message.nack` the message. - -When you call :meth:`~.pubsub_v1.subscriber.client.Client.subscribe`, you -must pass the callback that will be used. - -Here is an example: - -.. code-block:: python - - # Define the callback. - # Note that the callback is defined *before* the subscription is opened. - def callback(message): - do_something_with(message) # Replace this with your actual logic. - message.ack() # Asynchronously acknowledge the message. - - # Wrap the following code in `with pubsub.SubscriberClient() as subscriber:` - - # Substitute PROJECT and SUBSCRIPTION with appropriate values for your - # application. - subscription_path = subscriber.subscription_path(PROJECT, SUBSCRIPTION) - - # Open the subscription, passing the callback. - future = subscriber.subscribe(subscription_path, callback) - -The :meth:`~.pubsub_v1.subscriber.client.Client.subscribe` method returns -a :class:`~.pubsub_v1.subscriber.futures.StreamingPullFuture`, which is both -the interface to wait on messages (e.g. block the primary thread) and to -address exceptions. - -To block the thread you are in while messages are coming in the stream, -use the :meth:`~.pubsub_v1.subscriber.futures.Future.result` method: - -.. code-block:: python - - future.result() - -.. note: This will block forever assuming no errors or that ``cancel`` is never - called. - -You can also use this for error handling; any exceptions that crop up on a -thread will be set on the future. - -.. code-block:: python - - try: - future.result() - except Exception as ex: - # Close the subscriber if not using a context manager. - subscriber.close() - raise - -Finally, you can use -:meth:`~.pubsub_v1.subscriber.futures.StreamingPullFuture.cancel` to stop -receiving messages. - - -.. code-block:: python - - future.cancel() - - -.. _explaining-ack: - -Explaining Ack --------------- - -In Pub/Sub, the term **ack** stands for "acknowledge". You should ack a -message when your processing of that message *has completed*. When you ack -a message, you are telling Pub/Sub that you do not need to see it again. - -It might be tempting to ack messages immediately on receipt. While there -are valid use cases for this, in general it is unwise. The reason why: If -there is some error or edge case in your processing logic, and processing -of the message fails, you will have already told Pub/Sub that you successfully -processed the message. By contrast, if you ack only upon completion, then -Pub/Sub will eventually re-deliver the unacknowledged message. - -It is also possible to **nack** a message, which is the opposite. When you -nack, it tells Pub/Sub that you are unable or unwilling to deal with the -message, and that the service should redeliver it. - - -API Reference -------------- - -.. toctree:: - :maxdepth: 2 - - api/client - api/message - api/futures - api/pagers - api/scheduler diff --git a/docs/types.rst b/docs/types.rst deleted file mode 100644 index 308f66971..000000000 --- a/docs/types.rst +++ /dev/null @@ -1,6 +0,0 @@ -Pub/Sub Client Types -==================== - -.. automodule:: google.cloud.pubsub_v1.types - :members: - :noindex: