Document changes related to subscription management.

Author: fselmo

docs/internals.rst

@@ -430,10 +430,126 @@ request. The initial ``eth_subscribe`` request expects only one response, the
 subscription *id* value, but it also expects to receive many ``eth_subscription``
 messages if and when the request is successful. For this reason, the original request
 is considered a one-to-one request so that a subscription *id* can be returned to the
-user on the same line, but the ``process_subscriptions()`` method on the
+user on the same line. The many responses this call will produce can be handled in one
+of a few ways.
+
+The recommended way to handle one-to-many responses is to use the subscription manager
+API. The subscription manager API is a public API on the ``AsyncWeb3`` class, when
+connected to a ``PersistentConnectionProvider`` instance, that allows the user to
+subscribe to a subscription and handle the many responses asynchronously. The
+``subscription_manager`` instance is responsible for handling the many responses that
+come in over the socket connection, as long as handlers are passed to each subscription
+call. The subscription manager can also be used to unsubscribe from a subscription when
+the user is done with it.
+
+.. code-block:: python
+
+    >>> async def new_heads_handler(
+    ...     async_w3: AsyncWeb3,
+    ...     sx: EthSubscription,
+    ...     response: BlockData,
+    ... ) -> None:
+    ...     print(f"New block header: {response}\n")
+    ...     if response["number"] > 1234567:
+    ...         await sx.unsubscribe()
+
+    >>> async def ws_subscription_example():
+    ...     async with AsyncWeb3(WebSocketProvider(f"ws://127.0.0.1:8546")) as w3:
+    ...         # Subscribe to new block headers and receive the subscription_id.
+    ...         # A one-to-one call with a trigger for many responses
+    ...         subscription_id = await w3.eth.subscribe("newHeads", handler=new_heads_handler)
+    ...
+    ...         # Handle the subscription messages asynchronously using the subscription
+    ...         # manager. This will continue until no more subscriptions are present in
+    ...         # the subscription manager, or indefinitely if the `run_forever` flag
+    ...         # is set to `True`.
+    ...         await w3.subscription_manager.handle_subscriptions(run_forever=False)
+    >>> asyncio.run(ws_subscription_example())
+
+The manager can also subscribe to many subscriptions at one time. The
+``EthSubscription`` classes, available via ``web3.utils.subscriptions``, provide a
+friendly API for managing subscriptions. Since each connection and provider instance
+has its own message listener task and subscription manager instance, you can subscribe
+to many subscriptions at once and handle the many responses that come in over the socket
+connections.
+
+.. code-block:: python
+
+    >>> from web3 import (
+    ...     AsyncWeb3,
+    ...     WebSocketProvider,
+    ...     AsyncIPCProvider,
+    ... )
+    >>> from web3.utils.subscriptions import (
+    ...     EthSubscription,
+    ...     NewHeadsSubscription,
+    ...     PendingTxSubscription,
+    ...     LogsSubscription,
+    ... )
+
+    >>> async def new_heads_handler(
+    ...     async_w3: AsyncWeb3,
+    ...     sx: EthSubscription,
+    ...     response: BlockData,
+    ... ) -> None:
+    ...     print(f"New block header: {response}\n")
+    ...     if response["number"] > 1234567:
+    ...         await sx.unsubscribe()
+
+    >>> async def pending_txs_handler(
+    ...     async_w3: AsyncWeb3,
+    ...     sx: EthSubscription,
+    ...     response: TxData,
+    ... ) -> None:
+    ...     ...
+
+    >>> async def log_handler(
+    ...     async_w3: AsyncWeb3,
+    ...     sx: EthSubscription,
+    ...     response: LogData,
+    ... ) -> None:
+    ...     ...
+
+    >>> async def sx_manager():
+    ...     local_w3 = await AsyncWeb3(AsyncIPCProvider(LOCAL_IPC, label="mainnet-ipc"))
+    ...     # subscribe to many subscriptions via the subscription manager with handlers
+    ...     await local_w3.subscription_manager.subscribe(
+    ...         [
+    ...             NewHeadsSubscription(label="new-heads-mainnet", handler=new_heads_handler),
+    ...             PendingTxSubscription(
+    ...                 label="pending-tx-mainnet",  # optional label
+    ...                 full_transactions=True,
+    ...                 handler=pending_tx_handler,
+    ...             ),
+    ...             LogsSubscription(
+    ...                 label="WETH transfers",  # optional label
+    ...                 address=local_w3.to_checksum_address("0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2"),
+    ...                 topics=[HexStr("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef")],
+    ...                 handler=log_handler,
+    ...             ),
+    ...         ]
+    ...     )
+    ...
+    ...     public_w3 = await AsyncWeb3(WebSocketProvider(PUBLIC_PROVIDER_WS, label="public-ws"))
+    ...     # subscribe via eth_subscribe, with handler and label (optional)
+    ...     await public_w3.eth.subscribe("public_newHeads", handler=pending_tx_handler, label="new-heads-public-ws")
+
+    >>>     # This will handle all subscriptions until no more subscriptions are present
+    ...     # in either subscription manager instance. If the `run_forever` flag is set
+    ...     # to `True` on any manager instance, this will run indefinitely.
+    >>>     await asyncio.gather(
+    ...         public_w3.subscription_manager.handle_subscriptions(),
+    ...         local_w3.subscription_manager.handle_subscriptions(),
+    ...     )
+
+    >>> asyncio.run(sx_manager())
+
+
+The ``process_subscriptions()`` method on the
 :class:`~web3.providers.persistent.PersistentConnection` class, the public API for
-interacting with the active persistent socket connection, is set up to receive
-``eth_subscription`` responses over an asynchronous interator pattern.
+interacting with the active persistent socket connection, is also set up to receive
+``eth_subscription`` responses over an asynchronous interator pattern. You can use this
+method to listen for raw messages and process them as they come in.
 
 .. code-block:: python
 

web3/providers/persistent/subscription_manager.py

@@ -73,6 +73,13 @@ async def subscribe(
             )
 
     async def unsubscribe(self, subscription: EthSubscription) -> bool:
+        """
+        Used to unsubscribe from a subscription that is being managed by the
+        subscription manager.
+
+        :param subscription: The subscription to unsubscribe from.
+        :return: True if unsubscribing was successful, False otherwise.
+        """
         if subscription not in self.subscriptions:
             raise Web3ValueError(
                 f"Subscription not found or is not being managed by the subscription "
@@ -85,6 +92,12 @@ async def unsubscribe(self, subscription: EthSubscription) -> bool:
         return False
 
     async def unsubscribe_all(self) -> None:
+        """
+        Used to unsubscribe from all subscriptions that are being managed by the
+        subscription manager.
+
+        :return: None
+        """
         for sx in self.subscriptions:
             await self.unsubscribe(sx)
 
@@ -100,4 +113,12 @@ async def _handle_subscriptions(self, run_forever: bool = False) -> None:
         self._provider.logger.info("Subscription manager processing ended.")
 
     async def handle_subscriptions(self, run_forever: bool = False) -> None:
+        """
+        Used to process all subscriptions. It will run until all subscriptions are
+        unsubscribed from or, if `run_forever` is set to `True`, it will run
+        indefinitely.
+
+        :param run_forever: If `True`, the method will run indefinitely.
+        :return: None
+        """
         await self._handle_subscriptions(run_forever=run_forever)