Fix sphinx problem of eventhub

sdk/eventhub/azure-eventhubs/azure/eventhub/aio/client_async.py

@@ -251,7 +251,7 @@ def create_producer(self, *, partition_id=None, operation=None, send_timeout=Non
          queued. Default value is 60 seconds. If set to 0, there will be no timeout.
         :type send_timeout: float
         :param loop: An event loop. If not specified the default event loop will be used.
-        :rtype ~azure.eventhub.aio.producer_async.EventHubProducer
+        :rtype: ~azure.eventhub.aio.producer_async.EventHubProducer
 
         Example:
             .. literalinclude:: ../examples/async_examples/test_examples_eventhub_async.py

sdk/eventhub/azure-eventhubs/azure/eventhub/aio/producer_async.py

@@ -156,6 +156,7 @@ async def create_batch(self, max_size=None, partition_key=None):
         """
         Create an EventDataBatch object with max size being max_size.
         The max_size should be no greater than the max allowed message size defined by the service side.
+
         :param max_size: The maximum size of bytes data that an EventDataBatch object can hold.
         :type max_size: int
         :param partition_key: With the given partition_key, event data will land to
@@ -197,7 +198,8 @@ async def send(self, event_data, *, partition_key=None, timeout=None):
         :type partition_key: str
         :param timeout: The maximum wait time to send the event data.
          If not specified, the default wait time specified when the producer was created will be used.
-        :type timeout:float
+        :type timeout: float
+
         :raises: ~azure.eventhub.AuthenticationError, ~azure.eventhub.ConnectError, ~azure.eventhub.ConnectionLostError,
                 ~azure.eventhub.EventDataError, ~azure.eventhub.EventDataSendError, ~azure.eventhub.EventHubError
         :return: None

sdk/eventhub/azure-eventhubs/azure/eventhub/consumer.py

@@ -197,6 +197,7 @@ def receive(self, max_batch_size=None, timeout=None):
         :rtype: list[~azure.eventhub.common.EventData]
         :raises: ~azure.eventhub.AuthenticationError, ~azure.eventhub.ConnectError, ~azure.eventhub.ConnectionLostError,
                 ~azure.eventhub.EventHubError
+
         Example:
             .. literalinclude:: ../examples/test_examples_eventhub.py
                 :start-after: [START eventhub_client_sync_receive]

sdk/eventhub/azure-eventhubs/azure/eventhub/eventprocessor/event_processor.py

@@ -21,16 +21,42 @@
 
 
 class EventProcessor(object):
+    """
+    An EventProcessor constantly receives events from all partitions of the Event Hub in the context of a given
+    consumer group. The received data will be sent to PartitionProcessor to be processed.
+
+    It provides the user a convenient way to receive events from multiple partitions and save checkpoints.
+    If multiple EventProcessors are running for an event hub, they will automatically balance load.
+    This load balancing won't be available until preview 3.
+
+    Example:
+        .. code-block:: python
+
+            class MyPartitionProcessor(PartitionProcessor):
+                async def process_events(self, events):
+                    if events:
+                        # do something sync or async to process the events
+                        await self._checkpoint_manager.update_checkpoint(events[-1].offset, events[-1].sequence_number)
+
+            import asyncio
+            from azure.eventhub.aio import EventHubClient
+            from azure.eventhub.eventprocessor import EventProcessor, PartitionProcessor, Sqlite3PartitionManager
+            client = EventHubClient.from_connection_string("<your connection string>", receive_timeout=5, retry_total=3)
+            partition_manager = Sqlite3PartitionManager()
+            try:
+                event_processor = EventProcessor(client, "$default", MyPartitionProcessor, partition_manager)
+                asyncio.ensure_future(event_processor.start())
+                await asyncio.sleep(100)  # allow it to run 100 seconds
+                await event_processor.stop()
+            finally:
+                await partition_manager.close()
+
+    """
     def __init__(self, eventhub_client: EventHubClient, consumer_group_name: str,
                  partition_processor_factory: Callable[[CheckpointManager], PartitionProcessor],
                  partition_manager: PartitionManager, **kwargs):
         """
-        An EventProcessor constantly receives events from all partitions of the Event Hub in the context of a given
-        consumer group. The received data will be sent to PartitionProcessor to be processed.
-
-        It provides the user a convenient way to receive events from multiple partitions and save checkpoints.
-        If multiple EventProcessors are running for an event hub, they will automatically balance load.
-        This load balancing won't be available until preview 3.
+        Instantiate an EventProcessor.
 
         :param eventhub_client: An instance of ~azure.eventhub.aio.EventClient object
         :type eventhub_client: ~azure.eventhub.aio.EventClient
@@ -46,29 +72,6 @@ def __init__(self, eventhub_client: EventHubClient, consumer_group_name: str,
         :param initial_event_position: The offset to start a partition consumer if the partition has no checkpoint yet.
         :type initial_event_position: int or str
 
-        Example:
-            ```python
-                    class MyPartitionProcessor(PartitionProcessor):
-                        async def process_events(self, events):
-                            if events:
-                                # do something sync or async to process the events
-                                await self._checkpoint_manager.update_checkpoint(events[-1].offset, events[-1].sequence_number)
-
-
-                    import asyncio
-                    from azure.eventhub.aio import EventHubClient
-                    from azure.eventhub.eventprocessor import EventProcessor, PartitionProcessor, Sqlite3PartitionManager
-                    client = EventHubClient.from_connection_string("<your connection string>", receive_timeout=5, retry_total=3)
-                    partition_manager = Sqlite3PartitionManager()
-                    try:
-                        event_processor = EventProcessor(client, "$default", MyPartitionProcessor, partition_manager)
-                        asyncio.ensure_future(event_processor.start())
-                        await asyncio.sleep(100)  # allow it to run 100 seconds
-                        await event_processor.stop()
-                    finally:
-                        await partition_manager.close()
-            ```
-
         """
         self._consumer_group_name = consumer_group_name
         self._eventhub_client = eventhub_client
@@ -87,12 +90,12 @@ def __repr__(self):
     async def start(self):
         """Start the EventProcessor.
 
-            1. retrieve the partition ids from eventhubs
-            2. claim partition ownership of these partitions.
-            3. repeatedly call EvenHubConsumer.receive() to retrieve events and
-            call user defined PartitionProcessor.process_events()
+        1. retrieve the partition ids from eventhubs.
+        2. claim partition ownership of these partitions.
+        3. repeatedly call EvenHubConsumer.receive() to retrieve events and call user defined PartitionProcessor.process_events().
+
+        :return: None
 
-            :return None
         """
         log.info("EventProcessor %r is being started", self._id)
         partition_ids = await self._eventhub_client.get_partition_ids()
@@ -104,7 +107,8 @@ async def stop(self):
 
         This method cancels tasks that are running EventHubConsumer.receive() for the partitions owned by this EventProcessor.
 
-        :return None
+        :return: None
+
         """
         for i in range(len(self._tasks)):
             task = self._tasks.pop()

sdk/eventhub/azure-eventhubs/azure/eventhub/eventprocessor/partition_processor.py

@@ -30,10 +30,10 @@ async def close(self, reason):
         """Called when EventProcessor stops processing this PartitionProcessor.
 
         There are different reasons to trigger the PartitionProcessor to close.
-        Refer to enum class CloseReason
+        Refer to enum class ~azure.eventhub.eventprocessor.CloseReason
 
         :param reason: Reason for closing the PartitionProcessor.
-        :type reason: CloseReason
+        :type reason: ~azure.eventhub.eventprocessor.CloseReason
 
         """
         pass

sdk/eventhub/azure-eventhubs/azure/eventhub/producer.py

@@ -163,6 +163,7 @@ def create_batch(self, max_size=None, partition_key=None):
         """
         Create an EventDataBatch object with max size being max_size.
         The max_size should be no greater than the max allowed message size defined by the service side.
+
         :param max_size: The maximum size of bytes data that an EventDataBatch object can hold.
         :type max_size: int
         :param partition_key: With the given partition_key, event data will land to
@@ -204,7 +205,8 @@ def send(self, event_data, partition_key=None, timeout=None):
         :type partition_key: str
         :param timeout: The maximum wait time to send the event data.
          If not specified, the default wait time specified when the producer was created will be used.
-        :type timeout:float
+        :type timeout: float
+
         :raises: ~azure.eventhub.AuthenticationError, ~azure.eventhub.ConnectError, ~azure.eventhub.ConnectionLostError,
                 ~azure.eventhub.EventDataError, ~azure.eventhub.EventDataSendError, ~azure.eventhub.EventHubError
         :return: None

sdk/eventhub/azure-eventhubs/examples/test_examples_eventhub.py

@@ -57,7 +57,7 @@ def test_example_eventhub_sync_send_and_receive(live_eventhub_config):
 
     from azure.eventhub import EventData, EventPosition
 
-    # [START create_eventhub_client_producer]
+    # [START create_eventhub_client_sender]
     client = EventHubClient.from_connection_string(connection_str)
     # Create a producer.
     producer = client.create_producer(partition_id="0")