Repository: abhinavsingh/proxy.py
Pull request #709 (2021-11-08 16:53:57)

`EventManager` is also a context manager

codecov[bot]: # [Codecov](https://codecov.io/gh/abhinavsingh/proxy.py/pull/709?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh) Report
> Merging [#709](https://codecov.io/gh/abhinavsingh/proxy.py/pull/709?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh) (7e8ddb5) into [develop](https://codecov.io/gh/abhinavsingh/proxy.py/commit/c6eaacefce2934eceac25f7a15d2710d476f1d4c?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh) (c6eaace) will **decrease** coverage by `0.18%`.
> The diff coverage is `80.00%`.

> :exclamation: Current head 7e8ddb5 differs from pull request most recent head 8aeb4e6. Consider uploading reports for the commit 8aeb4e6 to get more accurate results
[![Impacted file tree graph](https://codecov.io/gh/abhinavsingh/proxy.py/pull/709/graphs/tree.svg?width=650&height=150&src=pr&token=Zh9J7b4la2&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh)](https://codecov.io/gh/abhinavsingh/proxy.py/pull/709?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh)

```diff
@@             Coverage Diff             @@
##           develop     #709      +/-   ##
===========================================
- Coverage    86.16%   85.97%   -0.19%     
===========================================
  Files          110      110              
  Lines         4915     4921       +6     
  Branches       404      404              
===========================================
- Hits          4235     4231       -4     
- Misses         604      609       +5     
- Partials        76       81       +5     
```

| Flag | Coverage Δ | |
|---|---|---|
| pytest | `85.93% <80.00%> (-0.09%)` | :arrow_down: |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Impacted Files](https://codecov.io/gh/abhinavsingh/proxy.py/pull/709?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh) | Coverage Δ | |
|---|---|---|
| [proxy/core/event/manager.py](https://codecov.io/gh/abhinavsingh/proxy.py/pull/709/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh#diff-cHJveHkvY29yZS9ldmVudC9tYW5hZ2VyLnB5) | `45.00% <66.66%> (+0.88%)` | :arrow_up: |
| [proxy/proxy.py](https://codecov.io/gh/abhinavsingh/proxy.py/pull/709/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh#diff-cHJveHkvcHJveHkucHk=) | `100.00% <100.00%> (ø)` | |
| [tests/test\_main.py](https://codecov.io/gh/abhinavsingh/proxy.py/pull/709/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh#diff-dGVzdHMvdGVzdF9tYWluLnB5) | `100.00% <100.00%> (ø)` | |
| [proxy/common/types.py](https://codecov.io/gh/abhinavsingh/proxy.py/pull/709/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh#diff-cHJveHkvY29tbW9uL3R5cGVzLnB5) | `86.66% <0.00%> (-13.34%)` | :arrow_down: |
| [tests/test\_set\_open\_file\_limit.py](https://codecov.io/gh/abhinavsingh/proxy.py/pull/709/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh#diff-dGVzdHMvdGVzdF9zZXRfb3Blbl9maWxlX2xpbWl0LnB5) | `96.29% <0.00%> (-3.71%)` | :arrow_down: |
| [proxy/common/flag.py](https://codecov.io/gh/abhinavsingh/proxy.py/pull/709/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh#diff-cHJveHkvY29tbW9uL2ZsYWcucHk=) | `93.16% <0.00%> (-1.71%)` | :arrow_down: |
| [proxy/common/utils.py](https://codecov.io/gh/abhinavsingh/proxy.py/pull/709/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh#diff-cHJveHkvY29tbW9uL3V0aWxzLnB5) | `91.73% <0.00%> (-1.66%)` | :arrow_down: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/abhinavsingh/proxy.py/pull/709?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/abhinavsingh/proxy.py/pull/709?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh). Last update [c6eaace...8aeb4e6](https://codecov.io/gh/abhinavsingh/proxy.py/pull/709?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh).


Patch:
diff --git a/examples/pubsub_eventing.py b/examples/pubsub_eventing.py
index 01dff1e..16436ee 100644
--- a/examples/pubsub_eventing.py
+++ b/examples/pubsub_eventing.py
@@ -44,6 +44,7 @@ def publisher_process(
     logger.info('publisher shutdown')
 
 
+# Execute within a separate thread context
 def on_event(payload: Dict[str, Any]) -> None:
     '''Subscriber callback.'''
     global num_events_received
@@ -55,49 +56,47 @@ def on_event(payload: Dict[str, Any]) -> None:
 
 if __name__ == '__main__':
     start_time = time.time()
-
-    # Start dispatcher thread using EventManager
-    event_manager = EventManager()
-    event_manager.start_event_dispatcher()
-    assert event_manager.event_queue
-
-    # Create a subscriber
-    subscriber = EventSubscriber(event_manager.event_queue)
-    # Internally, subscribe will start a separate thread
-    # to receive incoming published messages
-    subscriber.subscribe(on_event)
-
-    # Start a publisher process to demonstrate safe exchange
-    # of messages between processes.
-    publisher_shutdown_event = multiprocessing.Event()
-    publisher = multiprocessing.Process(
-        target=publisher_process, args=(
-            publisher_shutdown_event, event_manager.event_queue, ),
-    )
-    publisher.start()
-
-    try:
-        while True:
-            # Dispatch event from main process
-            event_manager.event_queue.publish(
-                request_id=main_publisher_request_id,
-                event_name=eventNames.WORK_STARTED,
-                event_payload={'time': time.time()},
-                publisher_id='eventing_pubsub_main',
-            )
-    except KeyboardInterrupt:
-        logger.info('bye!!!')
-    finally:
-        # Stop publisher
-        publisher_shutdown_event.set()
-        publisher.join()
-        # Stop subscriber thread
-        subscriber.unsubscribe()
-        # Signal dispatcher to shutdown
-        event_manager.stop_event_dispatcher()
-    logger.info(
-        'Received {0} events from main thread, {1} events from another process, in {2} seconds'.format(
-            num_events_received[0], num_events_received[1], time.time(
-            ) - start_time,
-        ),
-    )
+    # Start eventing core
+    with EventManager() as event_manager:
+        assert event_manager.queue
+
+        # Create a subscriber.
+        # Internally, subscribe will start a separate thread
+        # to receive incoming published messages.
+        subscriber = EventSubscriber(event_manager.queue)
+        subscriber.subscribe(on_event)
+
+        # Start a publisher process to demonstrate safe exchange
+        # of messages between processes.
+        publisher_shutdown_event = multiprocessing.Event()
+        publisher = multiprocessing.Process(
+            target=publisher_process, args=(
+                publisher_shutdown_event, event_manager.queue, ),
+        )
+        publisher.start()
+
+        # Dispatch event from main process too
+        # to demonstrate safe exchange of messages
+        # between threads.
+        try:
+            while True:
+                event_manager.queue.publish(
+                    request_id=main_publisher_request_id,
+                    event_name=eventNames.WORK_STARTED,
+                    event_payload={'time': time.time()},
+                    publisher_id='eventing_pubsub_main',
+                )
+        except KeyboardInterrupt:
+            logger.info('bye!!!')
+        finally:
+            # Stop publisher
+            publisher_shutdown_event.set()
+            publisher.join()
+            # Stop subscriber thread
+            subscriber.unsubscribe()
+        logger.info(
+            'Received {0} events from main thread, {1} events from another process, in {2} seconds'.format(
+                num_events_received[0], num_events_received[1], time.time(
+                ) - start_time,
+            ),
+        )
diff --git a/proxy/core/event/manager.py b/proxy/core/event/manager.py
index 0f22c52..6d83125 100644
--- a/proxy/core/event/manager.py
+++ b/proxy/core/event/manager.py
@@ -12,7 +12,8 @@ import logging
 import threading
 import multiprocessing
 
-from typing import Optional
+from typing import Optional, Type
+from types import TracebackType
 
 from .queue import EventQueue
 from .dispatcher import EventDispatcher
@@ -33,39 +34,52 @@ flags.add_argument(
 
 
 class EventManager:
-    """Event manager is an encapsulation around various initialization, dispatcher
-    start / stop API required for end-to-end eventing.
+    """Event manager is a context manager which provides
+    encapsulation around various setup and shutdown steps
+    to start the eventing core.
     """
 
     def __init__(self) -> None:
-        self.event_queue: Optional[EventQueue] = None
-        self.event_dispatcher: Optional[EventDispatcher] = None
-        self.event_dispatcher_thread: Optional[threading.Thread] = None
-        self.event_dispatcher_shutdown: Optional[threading.Event] = None
+        self.queue: Optional[EventQueue] = None
+        self.dispatcher: Optional[EventDispatcher] = None
+        self.dispatcher_thread: Optional[threading.Thread] = None
+        self.dispatcher_shutdown: Optional[threading.Event] = None
         self.manager: Optional[multiprocessing.managers.SyncManager] = None
 
-    def start_event_dispatcher(self) -> None:
+    def __enter__(self) -> 'EventManager':
+        self.setup()
+        return self
+
+    def __exit__(
+            self,
+            exc_type: Optional[Type[BaseException]],
+            exc_val: Optional[BaseException],
+            exc_tb: Optional[TracebackType],
+    ) -> None:
+        self.shutdown()
+
+    def setup(self) -> None:
         self.manager = multiprocessing.Manager()
-        self.event_queue = EventQueue(self.manager.Queue())
-        self.event_dispatcher_shutdown = threading.Event()
-        assert self.event_dispatcher_shutdown
-        assert self.event_queue
-        self.event_dispatcher = EventDispatcher(
-            shutdown=self.event_dispatcher_shutdown,
-            event_queue=self.event_queue,
+        self.queue = EventQueue(self.manager.Queue())
+        self.dispatcher_shutdown = threading.Event()
+        assert self.dispatcher_shutdown
+        assert self.queue
+        self.dispatcher = EventDispatcher(
+            shutdown=self.dispatcher_shutdown,
+            event_queue=self.queue,
         )
-        self.event_dispatcher_thread = threading.Thread(
-            target=self.event_dispatcher.run,
+        self.dispatcher_thread = threading.Thread(
+            target=self.dispatcher.run,
         )
-        self.event_dispatcher_thread.start()
-        logger.debug('Thread ID: %d', self.event_dispatcher_thread.ident)
+        self.dispatcher_thread.start()
+        logger.debug('Thread ID: %d', self.dispatcher_thread.ident)
 
-    def stop_event_dispatcher(self) -> None:
-        assert self.event_dispatcher_shutdown
-        assert self.event_dispatcher_thread
-        self.event_dispatcher_shutdown.set()
-        self.event_dispatcher_thread.join()
+    def shutdown(self) -> None:
+        assert self.dispatcher_shutdown
+        assert self.dispatcher_thread
+        self.dispatcher_shutdown.set()
+        self.dispatcher_thread.join()
         logger.debug(
             'Shutdown of global event dispatcher thread %d successful',
-            self.event_dispatcher_thread.ident,
+            self.dispatcher_thread.ident,
         )
diff --git a/proxy/proxy.py b/proxy/proxy.py
index ced5c53..4c65178 100644
--- a/proxy/proxy.py
+++ b/proxy/proxy.py
@@ -114,11 +114,11 @@ class Proxy:
         if self.flags.enable_events:
             logger.info('Core Event enabled')
             self.event_manager = EventManager()
-            self.event_manager.start_event_dispatcher()
+            self.event_manager.setup()
         self.pool = AcceptorPool(
             flags=self.flags,
             work_klass=self.work_klass,
-            event_queue=self.event_manager.event_queue if self.event_manager is not None else None,
+            event_queue=self.event_manager.queue if self.event_manager is not None else None,
         )
         self.pool.setup()
         return self
@@ -133,7 +133,7 @@ class Proxy:
         self.pool.shutdown()
         if self.flags.enable_events:
             assert self.event_manager is not None
-            self.event_manager.stop_event_dispatcher()
+            self.event_manager.shutdown()
 
 
 def main(


Test patch:
diff --git a/tests/test_main.py b/tests/test_main.py
index 4948b86..2075791 100644
--- a/tests/test_main.py
+++ b/tests/test_main.py
@@ -127,12 +127,12 @@ class TestMain(unittest.TestCase):
         mock_initialize.return_value.enable_events = True
         main([])
         mock_event_manager.assert_called_once()
-        mock_event_manager.return_value.start_event_dispatcher.assert_called_once()
-        mock_event_manager.return_value.stop_event_dispatcher.assert_called_once()
+        mock_event_manager.return_value.setup.assert_called_once()
+        mock_event_manager.return_value.shutdown.assert_called_once()
         mock_acceptor_pool.assert_called_with(
             flags=mock_initialize.return_value,
             work_klass=HttpProtocolHandler,
-            event_queue=mock_event_manager.return_value.event_queue,
+            event_queue=mock_event_manager.return_value.queue,
         )
         mock_acceptor_pool.return_value.setup.assert_called()
         mock_acceptor_pool.return_value.shutdown.assert_called()
@@ -171,8 +171,8 @@ class TestMain(unittest.TestCase):
         mock_acceptor_pool.return_value.setup.assert_called()
         # dashboard will also enable eventing
         mock_event_manager.assert_called_once()
-        mock_event_manager.return_value.start_event_dispatcher.assert_called_once()
-        mock_event_manager.return_value.stop_event_dispatcher.assert_called_once()
+        mock_event_manager.return_value.setup.assert_called_once()
+        mock_event_manager.return_value.shutdown.assert_called_once()
 
     @mock.patch('time.sleep')
     @mock.patch('proxy.common.plugins.Plugins.load')