Repository: abhinavsingh/proxy.py
Pull request #720 (2021-11-10 14:43:18)

Add `Listener`, Web server close on header, use `Pipe` instead of `Manager` in eventing core
- Also added acknowledgements for sub & un-sub actions
- Dispatcher now sends a shutdown notice to active subscribers

codecov[bot]: # [Codecov](https://codecov.io/gh/abhinavsingh/proxy.py/pull/720?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh) Report
> Merging [#720](https://codecov.io/gh/abhinavsingh/proxy.py/pull/720?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh) (8a0668f) into [develop](https://codecov.io/gh/abhinavsingh/proxy.py/commit/df5a66807cf8ba2f20340887130e087a33f8d12a?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh) (df5a668) will **decrease** coverage by `0.04%`.
> The diff coverage is `87.63%`.

[![Impacted file tree graph](https://codecov.io/gh/abhinavsingh/proxy.py/pull/720/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/720?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     #720      +/-   ##
===========================================
- Coverage    86.14%   86.10%   -0.05%     
===========================================
  Files          114      113       -1     
  Lines         5016     5123     +107     
  Branches       408      417       +9     
===========================================
+ Hits          4321     4411      +90     
- Misses         617      623       +6     
- Partials        78       89      +11     
```

| Flag | Coverage Δ | |
|---|---|---|
| pytest | `86.06% <87.63%> (+0.07%)` | :arrow_up: |

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/720?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/720/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh#diff-cHJveHkvY29yZS9ldmVudC9tYW5hZ2VyLnB5) | `47.22% <0.00%> (+3.63%)` | :arrow_up: |
| [proxy/dashboard/inspect\_traffic.py](https://codecov.io/gh/abhinavsingh/proxy.py/pull/720/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh#diff-cHJveHkvZGFzaGJvYXJkL2luc3BlY3RfdHJhZmZpYy5weQ==) | `51.85% <0.00%> (ø)` | |
| [proxy/http/inspector/devtools.py](https://codecov.io/gh/abhinavsingh/proxy.py/pull/720/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh#diff-cHJveHkvaHR0cC9pbnNwZWN0b3IvZGV2dG9vbHMucHk=) | `45.28% <0.00%> (ø)` | |
| [proxy/http/server/web.py](https://codecov.io/gh/abhinavsingh/proxy.py/pull/720/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh#diff-cHJveHkvaHR0cC9zZXJ2ZXIvd2ViLnB5) | `63.75% <0.00%> (-0.87%)` | :arrow_down: |
| [proxy/plugin/web\_server\_route.py](https://codecov.io/gh/abhinavsingh/proxy.py/pull/720/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh#diff-cHJveHkvcGx1Z2luL3dlYl9zZXJ2ZXJfcm91dGUucHk=) | `65.21% <0.00%> (ø)` | |
| [tests/core/test\_acceptor.py](https://codecov.io/gh/abhinavsingh/proxy.py/pull/720/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh#diff-dGVzdHMvY29yZS90ZXN0X2FjY2VwdG9yLnB5) | `100.00% <ø> (ø)` | |
| [proxy/core/event/dispatcher.py](https://codecov.io/gh/abhinavsingh/proxy.py/pull/720/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh#diff-cHJveHkvY29yZS9ldmVudC9kaXNwYXRjaGVyLnB5) | `61.11% <63.63%> (+1.11%)` | :arrow_up: |
| [proxy/core/event/subscriber.py](https://codecov.io/gh/abhinavsingh/proxy.py/pull/720/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh#diff-cHJveHkvY29yZS9ldmVudC9zdWJzY3JpYmVyLnB5) | `72.41% <75.00%> (-8.24%)` | :arrow_down: |
| [proxy/core/acceptor/listener.py](https://codecov.io/gh/abhinavsingh/proxy.py/pull/720/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh#diff-cHJveHkvY29yZS9hY2NlcHRvci9saXN0ZW5lci5weQ==) | `90.38% <90.38%> (ø)` | |
| [proxy/common/flag.py](https://codecov.io/gh/abhinavsingh/proxy.py/pull/720/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh#diff-cHJveHkvY29tbW9uL2ZsYWcucHk=) | `92.00% <100.00%> (-3.17%)` | :arrow_down: |
| ... and [18 more](https://codecov.io/gh/abhinavsingh/proxy.py/pull/720/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/abhinavsingh/proxy.py/pull/720?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/720?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh). Last update [df5a668...8a0668f](https://codecov.io/gh/abhinavsingh/proxy.py/pull/720?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/README.md b/README.md
index 503a2d1..9fdb7f8 100644
--- a/README.md
+++ b/README.md
@@ -128,23 +128,35 @@
   - Made to handle `tens-of-thousands` connections / sec
 
     ```console
-    # On Macbook Pro 2015 / 2.8 GHz Intel Core i7
-    ❯ hey -n 10000 -c 100 http://localhost:8899/
+    # On Macbook Pro 2019 / 2.4 GHz 8-Core Intel Core i9 / 32 GB RAM
+    ❯ hey -n 10000 -c 100 http://localhost:8899/http-route-example
 
     Summary:
-      Total:	0.6157 secs
-      Slowest:	0.1049 secs
-      Fastest:	0.0007 secs
-      Average:	0.0055 secs
-      Requests/sec:	16240.5444
+      Total:	0.3248 secs
+      Slowest:	0.1007 secs
+      Fastest:	0.0002 secs
+      Average:	0.0028 secs
+      Requests/sec:	30784.7958
 
-      Total data:	800000 bytes
-      Size/request:	80 bytes
+      Total data:	190000 bytes
+      Size/request:	19 bytes
 
     Response time histogram:
-      0.001 [1]     |
-      0.011 [9565]	|■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
-      0.022 [332]	|■
+      0.000 [1]	|
+      0.010 [9533]	|■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
+      0.020 [384]	|■■
+
+    Latency distribution:
+      10% in 0.0004 secs
+      25% in 0.0007 secs
+      50% in 0.0013 secs
+      75% in 0.0029 secs
+      90% in 0.0057 secs
+      95% in 0.0097 secs
+      99% in 0.0185 secs
+
+    Status code distribution:
+      [200]	100000 responses
     ```
 
 - Lightweight
diff --git a/examples/pubsub_eventing.py b/examples/pubsub_eventing.py
index 16436ee..607cf45 100644
--- a/examples/pubsub_eventing.py
+++ b/examples/pubsub_eventing.py
@@ -12,29 +12,38 @@ import time
 import multiprocessing
 import logging
 
-from typing import Dict, Any
+from typing import Dict, Any, Optional
 
+from proxy.common.constants import DEFAULT_LOG_FORMAT
 from proxy.core.event import EventManager, EventQueue, EventSubscriber, eventNames
 
-# Enable debug logging to view core event logs
-logging.basicConfig(level=logging.DEBUG)
+logging.basicConfig(level=logging.DEBUG, format=DEFAULT_LOG_FORMAT)
+
+logger = logging.getLogger(__name__)
+
 
-main_publisher_request_id = '1234'
-process_publisher_request_id = '12345'
 num_events_received = [0, 0]
 
-logger = logging.getLogger(__name__)
+
+# Execute within a separate thread context
+def on_event(payload: Dict[str, Any]) -> None:
+    '''Subscriber callback.'''
+    global num_events_received
+    if payload['request_id'] == '1234':
+        num_events_received[0] += 1
+    else:
+        num_events_received[1] += 1
 
 
 def publisher_process(
     shutdown_event: multiprocessing.synchronize.Event,
     dispatcher_queue: EventQueue,
 ) -> None:
-    logger.info('publisher starting')
+    logger.info('publisher started')
     try:
         while not shutdown_event.is_set():
             dispatcher_queue.publish(
-                request_id=process_publisher_request_id,
+                request_id='12345',
                 event_name=eventNames.WORK_STARTED,
                 event_payload={'time': time.time()},
                 publisher_id='eventing_pubsub_process',
@@ -44,27 +53,19 @@ 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
-    if payload['request_id'] == main_publisher_request_id:
-        num_events_received[0] += 1
-    else:
-        num_events_received[1] += 1
-
-
 if __name__ == '__main__':
     start_time = time.time()
+
     # Start eventing core
+    subscriber: Optional[EventSubscriber] = None
     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)
+        subscriber = EventSubscriber(event_manager.queue, callback=on_event)
+        subscriber.setup()
 
         # Start a publisher process to demonstrate safe exchange
         # of messages between processes.
@@ -81,7 +82,7 @@ if __name__ == '__main__':
         try:
             while True:
                 event_manager.queue.publish(
-                    request_id=main_publisher_request_id,
+                    request_id='1234',
                     event_name=eventNames.WORK_STARTED,
                     event_payload={'time': time.time()},
                     publisher_id='eventing_pubsub_main',
@@ -89,14 +90,16 @@ if __name__ == '__main__':
         except KeyboardInterrupt:
             logger.info('bye!!!')
         finally:
-            # Stop publisher
+            # Stop publisher process
             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,
-            ),
-        )
+            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,
+                ),
+            )
+    if subscriber:
+        subscriber.shutdown(do_unsubscribe=False)
diff --git a/proxy/common/flag.py b/proxy/common/flag.py
index 88609d5..ac6f251 100644
--- a/proxy/common/flag.py
+++ b/proxy/common/flag.py
@@ -233,6 +233,9 @@ class FlagParser:
             IpAddress,
             opts.get('hostname', ipaddress.ip_address(args.hostname)),
         )
+        args.unix_socket_path = opts.get(
+            'unix_socket_path', args.unix_socket_path,
+        )
         # AF_UNIX is not available on Windows
         # See https://bugs.python.org/issue33408
         if os.name != 'nt':
diff --git a/proxy/core/acceptor/__init__.py b/proxy/core/acceptor/__init__.py
index 7bc92a8..c3fd687 100644
--- a/proxy/core/acceptor/__init__.py
+++ b/proxy/core/acceptor/__init__.py
@@ -13,6 +13,7 @@ from .pool import AcceptorPool
 from .work import Work
 from .threadless import Threadless
 from .executors import ThreadlessPool
+from .listener import Listener
 
 __all__ = [
     'Acceptor',
@@ -20,4 +21,5 @@ __all__ = [
     'Work',
     'Threadless',
     'ThreadlessPool',
+    'Listener',
 ]
diff --git a/proxy/core/acceptor/acceptor.py b/proxy/core/acceptor/acceptor.py
index 18758f5..cd56b5d 100644
--- a/proxy/core/acceptor/acceptor.py
+++ b/proxy/core/acceptor/acceptor.py
@@ -34,7 +34,7 @@ class Acceptor(multiprocessing.Process):
     """Work acceptor process.
 
     On start-up, `Acceptor` accepts a file descriptor which will be used to
-    accept new work.  File descriptor is accepted over a `work_queue` which is
+    accept new work.  File descriptor is accepted over a `fd_queue` which is
     closed immediately after receiving the descriptor.
 
     `Acceptor` goes on to listen for new work over the received server socket.
@@ -55,7 +55,7 @@ class Acceptor(multiprocessing.Process):
     def __init__(
             self,
             idd: int,
-            work_queue: connection.Connection,
+            fd_queue: connection.Connection,
             flags: argparse.Namespace,
             lock: multiprocessing.synchronize.Lock,
             executor_queues: List[connection.Connection],
@@ -72,7 +72,7 @@ class Acceptor(multiprocessing.Process):
         # to avoid concurrent accept over server socket
         self.lock = lock
         # Queue over which server socket fd is received on start-up
-        self.work_queue: connection.Connection = work_queue
+        self.fd_queue: connection.Connection = fd_queue
         # Available executors
         self.executor_queues = executor_queues
         self.executor_pids = executor_pids
@@ -101,8 +101,11 @@ class Acceptor(multiprocessing.Process):
             self.flags.log_format,
         )
         self.selector = selectors.DefaultSelector()
-        fileno = recv_handle(self.work_queue)
-        self.work_queue.close()
+        # TODO: Use selector on fd_queue so that we can
+        # dynamically accept from new fds.
+        fileno = recv_handle(self.fd_queue)
+        self.fd_queue.close()
+        # TODO: Convert to socks i.e. list of fds
         self.sock = socket.fromfd(
             fileno,
             family=self.flags.family,
diff --git a/proxy/core/acceptor/listener.py b/proxy/core/acceptor/listener.py
new file mode 100644
index 0000000..7459038
--- /dev/null
+++ b/proxy/core/acceptor/listener.py
@@ -0,0 +1,108 @@
+# -*- coding: utf-8 -*-
+"""
+    proxy.py
+    ~~~~~~~~
+    ⚡⚡⚡ Fast, Lightweight, Pluggable, TLS interception capable proxy server focused on
+    Network monitoring, controls & Application development, testing, debugging.
+
+    :copyright: (c) 2013-present by Abhinav Singh and contributors.
+    :license: BSD, see LICENSE for more details.
+"""
+import os
+import socket
+import logging
+import argparse
+
+from typing import Optional, Any
+
+from ...common.flag import flags
+from ...common.constants import DEFAULT_BACKLOG, DEFAULT_IPV6_HOSTNAME, DEFAULT_PORT
+
+
+flags.add_argument(
+    '--backlog',
+    type=int,
+    default=DEFAULT_BACKLOG,
+    help='Default: 100. Maximum number of pending connections to proxy server',
+)
+
+flags.add_argument(
+    '--hostname',
+    type=str,
+    default=str(DEFAULT_IPV6_HOSTNAME),
+    help='Default: ::1. Server IP address.',
+)
+
+flags.add_argument(
+    '--port', type=int, default=DEFAULT_PORT,
+    help='Default: 8899. Server port.',
+)
+
+flags.add_argument(
+    '--unix-socket-path',
+    type=str,
+    default=None,
+    help='Default: None. Unix socket path to use.  ' +
+    'When provided --host and --port flags are ignored',
+)
+
+logger = logging.getLogger(__name__)
+
+
+class Listener:
+
+    def __init__(self, flags: argparse.Namespace) -> None:
+        self.flags = flags
+        # Set after binding to a port.
+        # Stored here separately because ephemeral ports can be used.
+        self._port: Optional[int] = None
+        self._socket: Optional[socket.socket] = None
+
+    def __enter__(self) -> 'Listener':
+        self.setup()
+        return self
+
+    def __exit__(self, *args: Any) -> None:
+        self.shutdown()
+
+    def fileno(self) -> Optional[int]:
+        if not self._socket:
+            return None
+        return self._socket.fileno()
+
+    def setup(self) -> None:
+        if self.flags.unix_socket_path:
+            self._listen_unix_socket()
+        else:
+            self._listen_server_port()
+        if self.flags.unix_socket_path:
+            logger.info(
+                'Listening on %s' %
+                self.flags.unix_socket_path,
+            )
+        else:
+            logger.info(
+                'Listening on %s:%s' %
+                (self.flags.hostname, self._port),
+            )
+
+    def shutdown(self) -> None:
+        assert self._socket
+        self._socket.close()
+        if self.flags.unix_socket_path:
+            os.remove(self.flags.unix_socket_path)
+
+    def _listen_unix_socket(self) -> None:
+        self._socket = socket.socket(self.flags.family, socket.SOCK_STREAM)
+        self._socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        self._socket.bind(self.flags.unix_socket_path)
+        self._socket.listen(self.flags.backlog)
+        self._socket.setblocking(False)
+
+    def _listen_server_port(self) -> None:
+        self._socket = socket.socket(self.flags.family, socket.SOCK_STREAM)
+        self._socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        self._socket.bind((str(self.flags.hostname), self.flags.port))
+        self._socket.listen(self.flags.backlog)
+        self._socket.setblocking(False)
+        self._port = self._socket.getsockname()[1]
diff --git a/proxy/core/acceptor/pool.py b/proxy/core/acceptor/pool.py
index e5fa4d9..ff680f6 100644
--- a/proxy/core/acceptor/pool.py
+++ b/proxy/core/acceptor/pool.py
@@ -8,8 +8,6 @@
     :copyright: (c) 2013-present by Abhinav Singh and contributors.
     :license: BSD, see LICENSE for more details.
 """
-import os
-import socket
 import logging
 import argparse
 import multiprocessing
@@ -19,48 +17,20 @@ from multiprocessing.reduction import send_handle
 
 from typing import Any, List, Optional
 
+from .listener import Listener
 from .acceptor import Acceptor
 
 from ..event import EventQueue
 
-from ...common.utils import bytes_
 from ...common.flag import flags
-from ...common.constants import DEFAULT_BACKLOG, DEFAULT_IPV6_HOSTNAME
-from ...common.constants import DEFAULT_NUM_ACCEPTORS, DEFAULT_PORT
-from ...common.constants import DEFAULT_PID_FILE
-
-logger = logging.getLogger(__name__)
+from ...common.constants import DEFAULT_NUM_ACCEPTORS
 
 # Lock shared by acceptors for
 # sequential acceptance of work.
 LOCK = multiprocessing.Lock()
 
+logger = logging.getLogger(__name__)
 
-flags.add_argument(
-    '--pid-file',
-    type=str,
-    default=DEFAULT_PID_FILE,
-    help='Default: None. Save parent process ID to a file.',
-)
-
-flags.add_argument(
-    '--backlog',
-    type=int,
-    default=DEFAULT_BACKLOG,
-    help='Default: 100. Maximum number of pending connections to proxy server',
-)
-
-flags.add_argument(
-    '--hostname',
-    type=str,
-    default=str(DEFAULT_IPV6_HOSTNAME),
-    help='Default: ::1. Server IP address.',
-)
-
-flags.add_argument(
-    '--port', type=int, default=DEFAULT_PORT,
-    help='Default: 8899. Server port.',
-)
 
 flags.add_argument(
     '--num-acceptors',
@@ -69,14 +39,6 @@ flags.add_argument(
     help='Defaults to number of CPU cores.',
 )
 
-flags.add_argument(
-    '--unix-socket-path',
-    type=str,
-    default=None,
-    help='Default: None. Unix socket path to use.  ' +
-    'When provided --host and --port flags are ignored',
-)
-
 
 class AcceptorPool:
     """AcceptorPool is a helper class which pre-spawns `Acceptor` processes
@@ -96,23 +58,25 @@ class AcceptorPool:
     """
 
     def __init__(
-        self, flags: argparse.Namespace,
-        executor_queues: List[connection.Connection],
-        executor_pids: List[int],
-        event_queue: Optional[EventQueue] = None,
+            self,
+            flags: argparse.Namespace,
+            listener: Listener,
+            executor_queues: List[connection.Connection],
+            executor_pids: List[int],
+            event_queue: Optional[EventQueue] = None,
     ) -> None:
         self.flags = flags
-        # Eventing core queue
-        self.event_queue: Optional[EventQueue] = event_queue
         # File descriptor to use for accepting new work
-        self.socket: Optional[socket.socket] = None
-        # Acceptor process instances
-        self.acceptors: List[Acceptor] = []
-        # Work queue used to share file descriptor with acceptor processes
-        self.work_queues: List[connection.Connection] = []
+        self.listener: Listener = listener
         # Available executors
         self.executor_queues: List[connection.Connection] = executor_queues
         self.executor_pids: List[int] = executor_pids
+        # Eventing core queue
+        self.event_queue: Optional[EventQueue] = event_queue
+        # Acceptor process instances
+        self.acceptors: List[Acceptor] = []
+        # Fd queues used to share file descriptor with acceptor processes
+        self.fd_queues: List[connection.Connection] = []
 
     def __enter__(self) -> 'AcceptorPool':
         self.setup()
@@ -122,28 +86,18 @@ class AcceptorPool:
         self.shutdown()
 
     def setup(self) -> None:
-        """Setup socket and acceptors."""
-        self._write_pid_file()
-        if self.flags.unix_socket_path:
-            self._listen_unix_socket()
-        else:
-            self._listen_server_port()
-        # Override flags.port to match the actual port
-        # we are listening upon.  This is necessary to preserve
-        # the server port when `--port=0` is used.
-        assert self.socket
-        self.flags.port = self.socket.getsockname()[1]
+        """Setup acceptors."""
         self._start()
+        logger.info('Started %d acceptors' % self.flags.num_acceptors)
         # Send file descriptor to all acceptor processes.
-        assert self.socket is not None
+        fd = self.listener.fileno()
         for index in range(self.flags.num_acceptors):
             send_handle(
-                self.work_queues[index],
-                self.socket.fileno(),
+                self.fd_queues[index],
+                fd,
                 self.acceptors[index].pid,
             )
-            self.work_queues[index].close()
-        self.socket.close()
+            self.fd_queues[index].close()
 
     def shutdown(self) -> None:
         logger.info('Shutting down %d acceptors' % self.flags.num_acceptors)
@@ -151,32 +105,15 @@ class AcceptorPool:
             acceptor.running.set()
         for acceptor in self.acceptors:
             acceptor.join()
-        if self.flags.unix_socket_path:
-            os.remove(self.flags.unix_socket_path)
-        self._delete_pid_file()
         logger.debug('Acceptors shutdown')
 
-    def _listen_unix_socket(self) -> None:
-        self.socket = socket.socket(self.flags.family, socket.SOCK_STREAM)
-        self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-        self.socket.bind(self.flags.unix_socket_path)
-        self.socket.listen(self.flags.backlog)
-        self.socket.setblocking(False)
-
-    def _listen_server_port(self) -> None:
-        self.socket = socket.socket(self.flags.family, socket.SOCK_STREAM)
-        self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-        self.socket.bind((str(self.flags.hostname), self.flags.port))
-        self.socket.listen(self.flags.backlog)
-        self.socket.setblocking(False)
-
     def _start(self) -> None:
         """Start acceptor processes."""
         for acceptor_id in range(self.flags.num_acceptors):
             work_queue = multiprocessing.Pipe()
             acceptor = Acceptor(
                 idd=acceptor_id,
-                work_queue=work_queue[1],
+                fd_queue=work_queue[1],
                 flags=self.flags,
                 lock=LOCK,
                 event_queue=self.event_queue,
@@ -190,16 +127,4 @@ class AcceptorPool:
                 acceptor.pid,
             )
             self.acceptors.append(acceptor)
-            self.work_queues.append(work_queue[0])
-        logger.info('Started %d acceptors' % self.flags.num_acceptors)
-
-    def _write_pid_file(self) -> None:
-        if self.flags.pid_file is not None:
-            # NOTE: Multiple instances of proxy.py running on
-            # same host machine will currently result in overwriting the PID file
-            with open(self.flags.pid_file, 'wb') as pid_file:
-                pid_file.write(bytes_(os.getpid()))
-
-    def _delete_pid_file(self) -> None:
-        if self.flags.pid_file and os.path.exists(self.flags.pid_file):
-            os.remove(self.flags.pid_file)
+            self.fd_queues.append(work_queue[0])
diff --git a/proxy/core/event/dispatcher.py b/proxy/core/event/dispatcher.py
index f65d5bf..1bcb402 100644
--- a/proxy/core/event/dispatcher.py
+++ b/proxy/core/event/dispatcher.py
@@ -9,12 +9,12 @@
     :license: BSD, see LICENSE for more details.
 """
 import queue
-import threading
 import logging
+import threading
 
-from typing import Dict, Any, List
+from multiprocessing import connection
 
-from ...common.types import DictQueueType
+from typing import Dict, Any, List
 
 from .queue import EventQueue
 from .names import eventNames
@@ -25,26 +25,20 @@ logger = logging.getLogger(__name__)
 class EventDispatcher:
     """Core EventDispatcher.
 
-    Provides:
-    1. A dispatcher module which consumes core events and dispatches
-       them to EventQueueBasePlugin
-    2. A publish utility for publishing core events into
-       global events queue.
-
     Direct consuming from global events queue outside of dispatcher
     module is not-recommended.  Python native multiprocessing queue
     doesn't provide a fanout functionality which core dispatcher module
-    implements so that several plugins can consume same published
-    event concurrently.
+    implements so that several plugins can consume the same published
+    event concurrently (when necessary).
 
     When --enable-events is used, a multiprocessing.Queue is created and
-    attached to global argparse.  This queue can then be used for
+    attached to global flags.  This queue can then be used for
     dispatching an Event dict object into the queue.
 
     When --enable-events is used, dispatcher module is automatically
-    started. Dispatcher module also ensures that queue is not full and
-    doesn't utilize too much memory in case there are no event plugins
-    enabled.
+    started.  Most importantly, dispatcher module ensures that queue is
+    not flooded and doesn't utilize too much memory in case there are no
+    event subscriber is enabled.
     """
 
     def __init__(
@@ -54,24 +48,29 @@ class EventDispatcher:
     ) -> None:
         self.shutdown: threading.Event = shutdown
         self.event_queue: EventQueue = event_queue
-        self.subscribers: Dict[str, DictQueueType] = {}
+        # subscriber connection objects
+        self.subscribers: Dict[str, connection.Connection] = {}
 
     def handle_event(self, ev: Dict[str, Any]) -> None:
         if ev['event_name'] == eventNames.SUBSCRIBE:
             self.subscribers[ev['event_payload']['sub_id']] = \
-                ev['event_payload']['channel']
+                ev['event_payload']['conn']
+            # send ack
+            ev['event_payload']['conn'].send({
+                'event_name': eventNames.SUBSCRIBED,
+            })
         elif ev['event_name'] == eventNames.UNSUBSCRIBE:
+            # send ack
+            print('unsubscription request ack sent')
+            self.subscribers[ev['event_payload']['sub_id']].send({
+                'event_name': eventNames.UNSUBSCRIBED,
+            })
+            # close conn and delete subscriber
+            self.subscribers[ev['event_payload']['sub_id']].close()
             del self.subscribers[ev['event_payload']['sub_id']]
         else:
             # logger.info(ev)
-            unsub_ids: List[str] = []
-            for sub_id in self.subscribers:
-                try:
-                    self.subscribers[sub_id].put(ev)
-                except BrokenPipeError:
-                    unsub_ids.append(sub_id)
-            for sub_id in unsub_ids:
-                del self.subscribers[sub_id]
+            self._broadcast(ev)
 
     def run_once(self) -> None:
         ev: Dict[str, Any] = self.event_queue.queue.get(timeout=1)
@@ -91,4 +90,23 @@ class EventDispatcher:
         except KeyboardInterrupt:
             pass
         except Exception as e:
-            logger.exception('Event dispatcher exception', exc_info=e)
+            logger.exception('Dispatcher exception', exc_info=e)
+        finally:
+            # Send shutdown message to all active subscribers
+            self._broadcast({
+                'event_name': eventNames.DISPATCHER_SHUTDOWN,
+            })
+
+    def _broadcast(self, ev: Dict[str, Any]) -> None:
+        broken_pipes: List[str] = []
+        for sub_id in self.subscribers:
+            try:
+                self.subscribers[sub_id].send(ev)
+            except BrokenPipeError:
+                logger.warning(
+                    'Subscriber#%s broken pipe', sub_id,
+                )
+                self.subscribers[sub_id].close()
+                broken_pipes.append(sub_id)
+        for sub_id in broken_pipes:
+            del self.subscribers[sub_id]
diff --git a/proxy/core/event/manager.py b/proxy/core/event/manager.py
index 20d5c6e..164cf90 100644
--- a/proxy/core/event/manager.py
+++ b/proxy/core/event/manager.py
@@ -43,7 +43,6 @@ class EventManager:
         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 __enter__(self) -> 'EventManager':
         self.setup()
@@ -53,8 +52,7 @@ class EventManager:
         self.shutdown()
 
     def setup(self) -> None:
-        self.manager = multiprocessing.Manager()
-        self.queue = EventQueue(self.manager.Queue())
+        self.queue = EventQueue(multiprocessing.Queue())
         self.dispatcher_shutdown = threading.Event()
         assert self.dispatcher_shutdown
         assert self.queue
@@ -66,14 +64,13 @@ class EventManager:
             target=self.dispatcher.run,
         )
         self.dispatcher_thread.start()
-        logger.debug('Thread ID: %d', self.dispatcher_thread.ident)
+        logger.debug('Dispatcher#%d started', self.dispatcher_thread.ident)
 
     def shutdown(self) -> None:
-        assert self.dispatcher_shutdown and self.dispatcher_thread and self.manager
+        assert self.dispatcher_shutdown and self.dispatcher_thread
         self.dispatcher_shutdown.set()
         self.dispatcher_thread.join()
-        self.manager.shutdown()
         logger.debug(
-            'Shutdown of global event dispatcher thread %d successful',
+            'Dispatcher#%d shutdown',
             self.dispatcher_thread.ident,
         )
diff --git a/proxy/core/event/names.py b/proxy/core/event/names.py
index 980c1ea..5040c88 100644
--- a/proxy/core/event/names.py
+++ b/proxy/core/event/names.py
@@ -10,13 +10,17 @@
 """
 from typing import NamedTuple
 
-# Name of the events that eventing framework will support
+# Name of the events that eventing framework supports.
+#
 # Ideally this must be configurable via command line or
 # at-least extendable via plugins.
 EventNames = NamedTuple(
     'EventNames', [
         ('SUBSCRIBE', int),
+        ('SUBSCRIBED', int),
         ('UNSUBSCRIBE', int),
+        ('UNSUBSCRIBED', int),
+        ('DISPATCHER_SHUTDOWN', int),
         ('WORK_STARTED', int),
         ('WORK_FINISHED', int),
         ('REQUEST_COMPLETE', int),
@@ -25,4 +29,4 @@ EventNames = NamedTuple(
         ('RESPONSE_COMPLETE', int),
     ],
 )
-eventNames = EventNames(1, 2, 3, 4, 5, 6, 7, 8)
+eventNames = EventNames(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11)
diff --git a/proxy/core/event/queue.py b/proxy/core/event/queue.py
index 27768eb..21c0cc3 100644
--- a/proxy/core/event/queue.py
+++ b/proxy/core/event/queue.py
@@ -9,8 +9,11 @@
     :license: BSD, see LICENSE for more details.
 """
 import os
-import threading
 import time
+import threading
+
+from multiprocessing import connection
+
 from typing import Dict, Optional, Any
 
 from ...common.types import DictQueueType
@@ -19,8 +22,9 @@ from .names import eventNames
 
 
 class EventQueue:
-    """Global event queue.  Ideally the queue must come from multiprocessing.Manager,
-    specially if you intent to publish/subscribe from multiple processes.
+    """Global event queue.  Must be a multiprocessing.Manager queue because
+    subscribers need to dispatch their subscription queue over this global
+    queue.
 
     Each published event contains following schema:
     {
@@ -48,10 +52,10 @@ class EventQueue:
         publisher_id: Optional[str] = None,
     ) -> None:
         self.queue.put({
-            'request_id': request_id,
             'process_id': os.getpid(),
             'thread_id': threading.get_ident(),
             'event_timestamp': time.time(),
+            'request_id': request_id,
             'event_name': event_name,
             'event_payload': event_payload,
             'publisher_id': publisher_id,
@@ -60,19 +64,16 @@ class EventQueue:
     def subscribe(
             self,
             sub_id: str,
-            channel: DictQueueType,
+            channel: connection.Connection,
     ) -> None:
         """Subscribe to global events.
 
         sub_id is a subscription identifier which must be globally
-        unique.  channel MUST be a multiprocessing.Manager().Queue().
-        This is necessary because channel itself is dispatched
-        over the global queue for direct subscription.
-
+        unique.  channel MUST be a multiprocessing connection.
         """
         self.queue.put({
             'event_name': eventNames.SUBSCRIBE,
-            'event_payload': {'sub_id': sub_id, 'channel': channel},
+            'event_payload': {'sub_id': sub_id, 'conn': channel},
         })
 
     def unsubscribe(
diff --git a/proxy/core/event/subscriber.py b/proxy/core/event/subscriber.py
index 55f7175..3e9b705 100644
--- a/proxy/core/event/subscriber.py
+++ b/proxy/core/event/subscriber.py
@@ -8,17 +8,18 @@
     :copyright: (c) 2013-present by Abhinav Singh and contributors.
     :license: BSD, see LICENSE for more details.
 """
+import uuid
 import queue
+import logging
 import threading
 import multiprocessing
-import logging
-import uuid
 
-from typing import Dict, Optional, Any, Callable
+from multiprocessing import connection
 
-from ...common.types import DictQueueType
+from typing import Dict, Optional, Any, Callable
 
 from .queue import EventQueue
+from .names import eventNames
 
 logger = logging.getLogger(__name__)
 
@@ -47,51 +48,102 @@ class EventSubscriber:
     safe queue passed to the relay thread.
     """
 
-    def __init__(self, event_queue: EventQueue) -> None:
-        self.manager: multiprocessing.managers.SyncManager = multiprocessing.Manager()
+    def __init__(self, event_queue: EventQueue, callback: Callable[[Dict[str, Any]], None]) -> None:
         self.event_queue = event_queue
+        self.callback = callback
         self.relay_thread: Optional[threading.Thread] = None
         self.relay_shutdown: Optional[threading.Event] = None
-        self.relay_channel: Optional[DictQueueType] = None
+        self.relay_recv: Optional[connection.Connection] = None
+        self.relay_send: Optional[connection.Connection] = None
         self.relay_sub_id: Optional[str] = None
 
-    def subscribe(self, callback: Callable[[Dict[str, Any]], None]) -> None:
-        self._start_relay_thread(callback)
-        assert self.relay_sub_id and self.relay_channel
-        self.event_queue.subscribe(self.relay_sub_id, self.relay_channel)
+    def __enter__(self) -> 'EventSubscriber':
+        self.setup()
+        return self
+
+    def __exit__(self, *args: Any) -> None:
+        self.shutdown()
+
+    def setup(self, do_subscribe: bool = True) -> None:
+        """Setup subscription thread.
+
+        Call subscribe() to actually start subscription.
+        """
+        self._start_relay_thread()
+        assert self.relay_sub_id and self.relay_recv
         logger.debug(
-            'Subscribed relay sub id %s from core events',
+            'Subscriber#%s relay setup done',
             self.relay_sub_id,
         )
+        if do_subscribe:
+            self.subscribe()
+
+    def shutdown(self, do_unsubscribe: bool = True) -> None:
+        """Teardown subscription thread.
+
+        Call unsubscribe() to actually stop subscription.
+        """
+        self._stop_relay_thread()
+        logger.debug(
+            'Subscriber#%s relay shutdown done',
+            self.relay_sub_id,
+        )
+        if do_unsubscribe:
+            self.unsubscribe()
+
+    def subscribe(self) -> None:
+        assert self.relay_sub_id and self.relay_send
+        self.event_queue.subscribe(self.relay_sub_id, self.relay_send)
 
     def unsubscribe(self) -> None:
         if self.relay_sub_id is None:
-            logger.warning('Unsubscribe called without existing subscription')
+            logger.warning(
+                'Relay called unsubscribe without an active subscription',
+            )
             return
-
         try:
             self.event_queue.unsubscribe(self.relay_sub_id)
         except BrokenPipeError:
             pass
         except EOFError:
             pass
-
-        self._stop_relay_thread()
-        logger.debug(
-            'Un-subscribed relay sub id %s from core events',
-            self.relay_sub_id,
-        )
+        finally:
+            # self.relay_sub_id = None
+            pass
 
     @staticmethod
     def relay(
+            sub_id: str,
             shutdown: threading.Event,
-            channel: DictQueueType,
+            channel: connection.Connection,
             callback: Callable[[Dict[str, Any]], None],
     ) -> None:
         while not shutdown.is_set():
             try:
-                ev = channel.get(timeout=1)
-                callback(ev)
+                if channel.poll(timeout=1):
+                    ev = channel.recv()
+                    if ev['event_name'] == eventNames.SUBSCRIBED:
+                        logger.info(
+                            'Subscriber#{0} subscribe ack received'.format(
+                                sub_id,
+                            ),
+                        )
+                    elif ev['event_name'] == eventNames.UNSUBSCRIBED:
+                        logger.info(
+                            'Subscriber#{0} unsubscribe ack received'.format(
+                                sub_id,
+                            ),
+                        )
+                        break
+                    elif ev['event_name'] == eventNames.DISPATCHER_SHUTDOWN:
+                        logger.info(
+                            'Subscriber#{0} received dispatcher shutdown event'.format(
+                                sub_id,
+                            ),
+                        )
+                        break
+                    else:
+                        callback(ev)
             except queue.Empty:
                 pass
             except EOFError:
@@ -99,22 +151,32 @@ class EventSubscriber:
             except KeyboardInterrupt:
                 break
 
-    def _start_relay_thread(self, callback: Callable[[Dict[str, Any]], None]) -> None:
+    def _start_relay_thread(self) -> None:
         self.relay_sub_id = uuid.uuid4().hex
         self.relay_shutdown = threading.Event()
-        self.relay_channel = self.manager.Queue()
+        self.relay_recv, self.relay_send = multiprocessing.Pipe()
         self.relay_thread = threading.Thread(
             target=EventSubscriber.relay,
-            args=(self.relay_shutdown, self.relay_channel, callback),
+            args=(
+                self.relay_sub_id, self.relay_shutdown,
+                self.relay_recv, self.callback,
+            ),
         )
+        self.relay_thread.daemon = True
         self.relay_thread.start()
 
     def _stop_relay_thread(self) -> None:
-        assert self.relay_thread and self.relay_shutdown and self.relay_channel and self.relay_sub_id
+        assert self.relay_thread and self.relay_shutdown and self.relay_recv and self.relay_send
         self.relay_shutdown.set()
         self.relay_thread.join()
-        self.manager.shutdown()
+        self.relay_recv.close()
+        # Currently relay_send instance here in
+        # subscriber is not the same as one received
+        # by dispatcher.  This may cause file
+        # descriptor leakage.  So we make a close
+        # here explicit on our side of relay_send too.
+        self.relay_send.close()
         self.relay_thread = None
         self.relay_shutdown = None
-        self.relay_channel = None
-        self.relay_sub_id = None
+        self.relay_recv = None
+        self.relay_send = None
diff --git a/proxy/core/eventing/__init__.py b/proxy/core/eventing/__init__.py
deleted file mode 100644
index 232621f..0000000
--- a/proxy/core/eventing/__init__.py
+++ /dev/null
@@ -1,10 +0,0 @@
-# -*- coding: utf-8 -*-
-"""
-    proxy.py
-    ~~~~~~~~
-    ⚡⚡⚡ Fast, Lightweight, Pluggable, TLS interception capable proxy server focused on
-    Network monitoring, controls & Application development, testing, debugging.
-
-    :copyright: (c) 2013-present by Abhinav Singh and contributors.
-    :license: BSD, see LICENSE for more details.
-"""
diff --git a/proxy/core/eventing/publisher.py b/proxy/core/eventing/publisher.py
deleted file mode 100644
index 35bffc2..0000000
--- a/proxy/core/eventing/publisher.py
+++ /dev/null
@@ -1,32 +0,0 @@
-# -*- coding: utf-8 -*-
-"""
-    proxy.py
-    ~~~~~~~~
-    ⚡⚡⚡ Fast, Lightweight, Pluggable, TLS interception capable proxy server focused on
-    Network monitoring, controls & Application development, testing, debugging.
-
-    :copyright: (c) 2013-present by Abhinav Singh and contributors.
-    :license: BSD, see LICENSE for more details.
-"""
-
-
-class EventingPublisher:
-    """EventingTopic provides utilities around pubsub functionalities
-    within a particular topic context.
-
-    Each topic a unix socket domain file on disk.  Publishers from
-    within proxy.py instance or from outside can publish to these
-    topics.
-
-    Payload published to a topic can be consumed by multiple subscriber,
-    which can be across CPU processes.  Subscribers can ideally also
-    exist outside of proxy.py instance because topics are on-disk.
-
-    Subscribers will only receive events published after they
-    have completed the subscription phase.  No buffering/caching/redelivery/ack
-    of payload is currently implemented.
-    """
-
-    def publish(self, topic: str) -> None:
-        """Publish payload into a topic."""
-        pass
diff --git a/proxy/core/eventing/subscriber.py b/proxy/core/eventing/subscriber.py
deleted file mode 100644
index 9b16864..0000000
--- a/proxy/core/eventing/subscriber.py
+++ /dev/null
@@ -1,25 +0,0 @@
-# -*- coding: utf-8 -*-
-"""
-    proxy.py
-    ~~~~~~~~
-    ⚡⚡⚡ Fast, Lightweight, Pluggable, TLS interception capable proxy server focused on
-    Network monitoring, controls & Application development, testing, debugging.
-
-    :copyright: (c) 2013-present by Abhinav Singh and contributors.
-    :license: BSD, see LICENSE for more details.
-"""
-
-
-class EventingSubscriber:
-    """EventingSubscriber provides subscription utilities for a topic."""
-
-    def __init__(self, topic: str) -> None:
-        self.topic = topic
-
-    def subscribe(self) -> None:
-        """Subscribe to topic."""
-        pass
-
-    def unsubscribe(self) -> None:
-        """Unsubscribe from topic"""
-        pass
diff --git a/proxy/dashboard/inspect_traffic.py b/proxy/dashboard/inspect_traffic.py
index 4689d29..33fa212 100644
--- a/proxy/dashboard/inspect_traffic.py
+++ b/proxy/dashboard/inspect_traffic.py
@@ -24,7 +24,12 @@ class InspectTrafficPlugin(ProxyDashboardWebsocketPlugin):
 
     def __init__(self, *args: Any, **kwargs: Any) -> None:
         super().__init__(*args, **kwargs)
-        self.subscriber = EventSubscriber(self.event_queue)
+        self.subscriber = EventSubscriber(
+            self.event_queue,
+            callback=lambda event: InspectTrafficPlugin.callback(
+                self.client, event,
+            ),
+        )
 
     def methods(self) -> List[str]:
         return [
@@ -41,23 +46,22 @@ class InspectTrafficPlugin(ProxyDashboardWebsocketPlugin):
                         WebsocketFrame.text(
                             bytes_(
                                 json.dumps(
-                                    {'id': message['id'], 'response': 'not enabled'},
+                                    {
+                                        'id': message['id'],
+                                        'response': 'not enabled',
+                                    },
                                 ),
                             ),
                         ),
                     ),
                 )
             else:
-                self.subscriber.subscribe(
-                    lambda event: InspectTrafficPlugin.callback(
-                        self.client, event,
-                    ),
-                )
+                self.subscriber.setup()
                 self.reply(
                     {'id': message['id'], 'response': 'inspection_enabled'},
                 )
         elif message['method'] == 'disable_inspection':
-            self.subscriber.unsubscribe()
+            self.subscriber.shutdown()
             self.reply({
                 'id': message['id'],
                 'response': 'inspection_disabled',
diff --git a/proxy/http/inspector/devtools.py b/proxy/http/inspector/devtools.py
index 1d03e16..2199b40 100644
--- a/proxy/http/inspector/devtools.py
+++ b/proxy/http/inspector/devtools.py
@@ -54,7 +54,12 @@ class DevtoolsProtocolPlugin(HttpWebServerBasePlugin):
 
     def __init__(self, *args: Any, **kwargs: Any) -> None:
         super().__init__(*args, **kwargs)
-        self.subscriber = EventSubscriber(self.event_queue)
+        self.subscriber = EventSubscriber(
+            self.event_queue,
+            callback=lambda event: CoreEventsToDevtoolsProtocol.transformer(
+                self.client, event,
+            ),
+        )
 
     def routes(self) -> List[Tuple[int, str]]:
         return [
@@ -65,11 +70,7 @@ class DevtoolsProtocolPlugin(HttpWebServerBasePlugin):
         raise NotImplementedError('This should have never been called')
 
     def on_websocket_open(self) -> None:
-        self.subscriber.subscribe(
-            lambda event: CoreEventsToDevtoolsProtocol.transformer(
-                self.client, event,
-            ),
-        )
+        self.subscriber.setup()
 
     def on_websocket_message(self, frame: WebsocketFrame) -> None:
         try:
@@ -82,7 +83,7 @@ class DevtoolsProtocolPlugin(HttpWebServerBasePlugin):
         self.handle_devtools_message(message)
 
     def on_client_connection_close(self) -> None:
-        self.subscriber.unsubscribe()
+        self.subscriber.shutdown()
 
     def handle_devtools_message(self, message: Dict[str, Any]) -> None:
         frame = WebsocketFrame()
diff --git a/proxy/http/server/web.py b/proxy/http/server/web.py
index 51d599b..09c50f1 100644
--- a/proxy/http/server/web.py
+++ b/proxy/http/server/web.py
@@ -207,6 +207,9 @@ class HttpWebServerPlugin(HttpProtocolHandlerPlugin):
             if match:
                 self.route = self.routes[protocol][route]
                 self.route.handle_request(self.request)
+                if self.request.has_header(b'connection') \
+                        and self.request.header(b'connection').lower() == b'close':
+                    return True
                 return False
 
         # No-route found, try static serving if enabled
diff --git a/proxy/plugin/web_server_route.py b/proxy/plugin/web_server_route.py
index 3675db9..f310aa5 100644
--- a/proxy/plugin/web_server_route.py
+++ b/proxy/plugin/web_server_route.py
@@ -55,4 +55,4 @@ class WebServerPlugin(HttpWebServerBasePlugin):
         logger.info(frame.data)
 
     def on_client_connection_close(self) -> None:
-        logger.info('Websocket close')
+        logger.debug('Client connection close')
diff --git a/proxy/proxy.py b/proxy/proxy.py
index e20b7ec..9e4c116 100644
--- a/proxy/proxy.py
+++ b/proxy/proxy.py
@@ -8,18 +8,20 @@
     :copyright: (c) 2013-present by Abhinav Singh and contributors.
     :license: BSD, see LICENSE for more details.
 """
+import os
 import sys
 import time
 import logging
 
 from typing import List, Optional, Any
 
-from .core.acceptor import AcceptorPool, ThreadlessPool
+from .core.acceptor import AcceptorPool, ThreadlessPool, Listener
 from .core.event import EventManager
+from .common.utils import bytes_
 from .common.flag import FlagParser, flags
 from .common.constants import DEFAULT_LOG_FILE, DEFAULT_LOG_FORMAT, DEFAULT_LOG_LEVEL
 from .common.constants import DEFAULT_OPEN_FILE_LIMIT, DEFAULT_PLUGINS, DEFAULT_VERSION
-from .common.constants import DEFAULT_ENABLE_DASHBOARD, DEFAULT_WORK_KLASS
+from .common.constants import DEFAULT_ENABLE_DASHBOARD, DEFAULT_WORK_KLASS, DEFAULT_PID_FILE
 
 
 logger = logging.getLogger(__name__)
@@ -93,6 +95,13 @@ flags.add_argument(
     '.  Work klass to use for work execution.',
 )
 
+flags.add_argument(
+    '--pid-file',
+    type=str,
+    default=DEFAULT_PID_FILE,
+    help='Default: None. Save "parent" process ID to a file.',
+)
+
 
 class Proxy:
     """Context manager to control AcceptorPool, ExecutorPool & EventingCore lifecycle.
@@ -107,8 +116,9 @@ class Proxy:
 
     def __init__(self, input_args: Optional[List[str]] = None, **opts: Any) -> None:
         self.flags = FlagParser.initialize(input_args, **opts)
-        self.acceptors: Optional[AcceptorPool] = None
+        self.listener: Optional[Listener] = None
         self.executors: Optional[ThreadlessPool] = None
+        self.acceptors: Optional[AcceptorPool] = None
         self.event_manager: Optional[EventManager] = None
 
     def __enter__(self) -> 'Proxy':
@@ -132,6 +142,15 @@ class Proxy:
         # TODO: Pid watcher which watches for processes started
         # by proxy.py core.  May be alert or restart those processes
         # on failure.
+        self._write_pid_file()
+        # We setup listeners first because of flags.port override
+        # in case of ephemeral port being used
+        self.listener = Listener(flags=self.flags)
+        self.listener.setup()
+        # Override flags.port to match the actual port
+        # we are listening upon.  This is necessary to preserve
+        # the server port when `--port=0` is used.
+        self.flags.port = self.listener._port
         if self.flags.enable_events:
             logger.info('Core Event enabled')
             self.event_manager = EventManager()
@@ -146,22 +165,13 @@ class Proxy:
         self.executors.setup()
         self.acceptors = AcceptorPool(
             flags=self.flags,
-            event_queue=event_queue,
+            listener=self.listener,
             executor_queues=self.executors.work_queues,
             executor_pids=self.executors.work_pids,
+            event_queue=event_queue,
         )
         self.acceptors.setup()
-        assert self.acceptors is not None
-        if self.flags.unix_socket_path:
-            logger.info(
-                'Listening on %s' %
-                self.flags.unix_socket_path,
-            )
-        else:
-            logger.info(
-                'Listening on %s:%s' %
-                (self.acceptors.flags.hostname, self.acceptors.flags.port),
-            )
+        # TODO: May be close listener fd as we don't need it now
 
     def shutdown(self) -> None:
         assert self.acceptors
@@ -171,6 +181,20 @@ class Proxy:
         if self.flags.enable_events:
             assert self.event_manager is not None
             self.event_manager.shutdown()
+        assert self.listener
+        self.listener.shutdown()
+        self._delete_pid_file()
+
+    def _write_pid_file(self) -> None:
+        if self.flags.pid_file is not None:
+            # NOTE: Multiple instances of proxy.py running on
+            # same host machine will currently result in overwriting the PID file
+            with open(self.flags.pid_file, 'wb') as pid_file:
+                pid_file.write(bytes_(os.getpid()))
+
+    def _delete_pid_file(self) -> None:
+        if self.flags.pid_file and os.path.exists(self.flags.pid_file):
+            os.remove(self.flags.pid_file)
 
 
 def main(**opts: Any) -> None:


Test patch:
diff --git a/tests/core/test_acceptor.py b/tests/core/test_acceptor.py
index 274399c..25c092a 100644
--- a/tests/core/test_acceptor.py
+++ b/tests/core/test_acceptor.py
@@ -24,11 +24,12 @@ class TestAcceptor(unittest.TestCase):
         self.acceptor_id = 1
         self.pipe = multiprocessing.Pipe()
         self.flags = FlagParser.initialize(
-            threaded=True, work_klass=mock.MagicMock(),
+            threaded=True,
+            work_klass=mock.MagicMock(),
         )
         self.acceptor = Acceptor(
             idd=self.acceptor_id,
-            work_queue=self.pipe[1],
+            fd_queue=self.pipe[1],
             flags=self.flags,
             lock=multiprocessing.Lock(),
             executor_queues=[],
diff --git a/tests/core/test_acceptor_pool.py b/tests/core/test_acceptor_pool.py
index e2b1064..e2a17c1 100644
--- a/tests/core/test_acceptor_pool.py
+++ b/tests/core/test_acceptor_pool.py
@@ -8,82 +8,74 @@
     :copyright: (c) 2013-present by Abhinav Singh and contributors.
     :license: BSD, see LICENSE for more details.
 """
-import os
-import socket
-import tempfile
 import unittest
 
 from unittest import mock
 
-from proxy.common.utils import bytes_
 from proxy.common.flag import FlagParser
 from proxy.core.acceptor import AcceptorPool
 
 
 class TestAcceptorPool(unittest.TestCase):
 
-    @mock.patch('os.remove')
-    @mock.patch('os.path.exists')
-    @mock.patch('builtins.open')
     @mock.patch('proxy.core.acceptor.pool.send_handle')
     @mock.patch('multiprocessing.Pipe')
-    @mock.patch('socket.socket')
     @mock.patch('proxy.core.acceptor.pool.Acceptor')
+    @mock.patch('proxy.core.acceptor.Listener')
     def test_setup_and_shutdown(
             self,
+            mock_listener: mock.Mock,
             mock_acceptor: mock.Mock,
-            mock_socket: mock.Mock,
             mock_pipe: mock.Mock,
             mock_send_handle: mock.Mock,
-            mock_open: mock.Mock,
-            mock_exists: mock.Mock,
-            mock_remove: mock.Mock,
     ) -> None:
         acceptor1 = mock.MagicMock()
         acceptor2 = mock.MagicMock()
         mock_acceptor.side_effect = [acceptor1, acceptor2]
 
-        num_workers = 2
-        pid_file = os.path.join(tempfile.gettempdir(), 'pid')
-        sock = mock_socket.return_value
+        num_acceptors = 2
         flags = FlagParser.initialize(
-            num_workers=2, pid_file=pid_file, threaded=True,
+            num_acceptors=num_acceptors,
+            threaded=True,
         )
+        self.assertEqual(flags.num_acceptors, num_acceptors)
 
-        pool = AcceptorPool(flags=flags, executor_queues=[], executor_pids=[])
+        pool = AcceptorPool(
+            flags=flags, listener=mock_listener.return_value,
+            executor_queues=[], executor_pids=[],
+        )
         pool.setup()
+
+        self.assertEqual(mock_pipe.call_count, num_acceptors)
+        self.assertEqual(mock_acceptor.call_count, num_acceptors)
         mock_send_handle.assert_called()
+        self.assertEqual(mock_send_handle.call_count, num_acceptors)
 
-        mock_socket.assert_called_with(
-            socket.AF_INET6 if pool.flags.hostname.version == 6 else socket.AF_INET,
-            socket.SOCK_STREAM,
+        self.assertEqual(
+            mock_acceptor.call_args_list[0][1]['idd'], 0,
+        )
+        self.assertEqual(
+            mock_acceptor.call_args_list[0][1]['fd_queue'], mock_pipe.return_value[1],
         )
-        sock.setsockopt.assert_called_with(
-            socket.SOL_SOCKET, socket.SO_REUSEADDR, 1,
+        self.assertEqual(
+            mock_acceptor.call_args_list[0][1]['flags'], flags,
         )
-        sock.bind.assert_called_with(
-            (str(pool.flags.hostname), 8899),
+        self.assertEqual(
+            mock_acceptor.call_args_list[0][1]['event_queue'], None,
         )
-        sock.listen.assert_called_with(pool.flags.backlog)
-        sock.setblocking.assert_called_with(False)
+        # executor_queues=[],
+        # executor_pids=[]
+        self.assertEqual(
+            mock_acceptor.call_args_list[1][1]['idd'], 1,
+        )
+
+        acceptor1.start.assert_called_once()
+        acceptor2.start.assert_called_once()
+        mock_listener.return_value.fileno.assert_called_once()
 
-        self.assertTrue(mock_pipe.call_count, num_workers)
-        self.assertTrue(mock_acceptor.call_count, num_workers)
-        acceptor1.start.assert_called()
-        acceptor2.start.assert_called()
         acceptor1.join.assert_not_called()
         acceptor2.join.assert_not_called()
 
-        sock.close.assert_called()
-
         pool.shutdown()
-
-        mock_open.assert_called_with(pid_file, 'wb')
-        mock_open.return_value.__enter__.return_value.write.assert_called_with(
-            bytes_(os.getpid()),
-        )
-        mock_exists.assert_called_with(pid_file)
-        mock_remove.assert_called_with(pid_file)
-
-        acceptor1.join.assert_called()
-        acceptor2.join.assert_called()
+        acceptor1.join.assert_called_once()
+        acceptor2.join.assert_called_once()
diff --git a/tests/core/test_event_dispatcher.py b/tests/core/test_event_dispatcher.py
index 99a4de8..74e7b71 100644
--- a/tests/core/test_event_dispatcher.py
+++ b/tests/core/test_event_dispatcher.py
@@ -8,15 +8,16 @@
     :copyright: (c) 2013-present by Abhinav Singh and contributors.
     :license: BSD, see LICENSE for more details.
 """
-import multiprocessing
 import os
 import threading
 import unittest
 import queue
+import multiprocessing
+
+from multiprocessing import connection
 
 from unittest import mock
 
-from proxy.common.types import DictQueueType
 from proxy.core.event import EventDispatcher, EventQueue, eventNames
 
 
@@ -47,25 +48,32 @@ class TestEventDispatcher(unittest.TestCase):
             self.dispatcher.run_once()
 
     @mock.patch('time.time')
-    def subscribe(self, mock_time: mock.Mock) -> DictQueueType:
+    def subscribe(self, mock_time: mock.Mock) -> connection.Connection:
         mock_time.return_value = 1234567
-        # Do not use self.manager here, otherwise you may be
-        # TypeError: AutoProxy() got an unexpected keyword argument 'manager_owned'.
-        #
-        # Queue used for subscription must be acquired from a separate
-        # multiprocessing.Manager than the one used for event_queue.
-        q = multiprocessing.Manager().Queue()
-        self.event_queue.subscribe(sub_id='1234', channel=q)
+        relay_recv, relay_send = multiprocessing.Pipe()
+        self.event_queue.subscribe(sub_id='1234', channel=relay_send)
+        # consume the subscribe event
         self.dispatcher.run_once()
+        # assert subscribed ack
+        self.assertEqual(
+            relay_recv.recv(), {
+                'event_name': eventNames.SUBSCRIBED,
+            },
+        )
+        # publish event
         self.event_queue.publish(
             request_id='1234',
             event_name=eventNames.WORK_STARTED,
             event_payload={'hello': 'events'},
             publisher_id=self.__class__.__name__,
         )
+        # consume
         self.dispatcher.run_once()
+        # assert recv
+        r = relay_recv.recv()
+        print(r)
         self.assertEqual(
-            q.get(), {
+            r, {
                 'request_id': '1234',
                 'process_id': os.getpid(),
                 'thread_id': threading.get_ident(),
@@ -75,15 +83,21 @@ class TestEventDispatcher(unittest.TestCase):
                 'publisher_id': self.__class__.__name__,
             },
         )
-        return q
+        return relay_recv
 
     def test_subscribe(self) -> None:
         self.subscribe()
 
     def test_unsubscribe(self) -> None:
-        q = self.subscribe()
+        relay_recv = self.subscribe()
         self.event_queue.unsubscribe('1234')
         self.dispatcher.run_once()
+        # assert unsubscribe ack
+        self.assertEqual(
+            relay_recv.recv(), {
+                'event_name': eventNames.UNSUBSCRIBED,
+            },
+        )
         self.event_queue.publish(
             request_id='1234',
             event_name=eventNames.WORK_STARTED,
@@ -91,11 +105,11 @@ class TestEventDispatcher(unittest.TestCase):
             publisher_id=self.__class__.__name__,
         )
         self.dispatcher.run_once()
-        with self.assertRaises(queue.Empty):
-            q.get(timeout=0.1)
+        with self.assertRaises(EOFError):
+            relay_recv.recv()
 
-    def test_unsubscribe_on_broken_pipe_error(self) -> None:
-        pass
+    # def test_unsubscribe_on_broken_pipe_error(self) -> None:
+    #     pass
 
-    def test_run(self) -> None:
-        pass
+    # def test_run(self) -> None:
+    #     pass
diff --git a/tests/core/test_event_queue.py b/tests/core/test_event_queue.py
index a0125d7..174d573 100644
--- a/tests/core/test_event_queue.py
+++ b/tests/core/test_event_queue.py
@@ -50,11 +50,18 @@ class TestCoreEvent(unittest.TestCase):
 
     def test_subscribe(self) -> None:
         evq = EventQueue(self.manager.Queue())
-        q = multiprocessing.Manager().Queue()
-        evq.subscribe('1234', q)
+        _, relay_send = multiprocessing.Pipe()
+        print(relay_send)
+        evq.subscribe('1234', relay_send)
         ev = evq.queue.get()
         self.assertEqual(ev['event_name'], eventNames.SUBSCRIBE)
         self.assertEqual(ev['event_payload']['sub_id'], '1234')
+        # Not the same as the one sent over multiprocessing.Queue
+        # will be another descriptor.  We must ideally use
+        # send_handle during subscription too.
+        #
+        # self.assertEqual(ev['event_payload']
+        #                  ['conn'].fileno(), relay_send.fileno())
 
     def test_unsubscribe(self) -> None:
         evq = EventQueue(self.manager.Queue())
diff --git a/tests/core/test_event_subscriber.py b/tests/core/test_event_subscriber.py
index 5cbb303..3ce8dcb 100644
--- a/tests/core/test_event_subscriber.py
+++ b/tests/core/test_event_subscriber.py
@@ -9,6 +9,7 @@
     :license: BSD, see LICENSE for more details.
 """
 import os
+import queue
 import threading
 import unittest
 import multiprocessing
@@ -25,6 +26,7 @@ class TestEventSubscriber(unittest.TestCase):
 
     def setUp(self) -> None:
         self.manager = multiprocessing.Manager()
+        self.event_queue = EventQueue(self.manager.Queue())
 
     def tearDown(self) -> None:
         self.manager.shutdown()
@@ -33,14 +35,12 @@ class TestEventSubscriber(unittest.TestCase):
     def test_event_subscriber(self, mock_time: mock.Mock) -> None:
         mock_time.return_value = 1234567
         self.dispatcher_shutdown = threading.Event()
-        self.event_queue = EventQueue(self.manager.Queue())
         self.dispatcher = EventDispatcher(
             shutdown=self.dispatcher_shutdown,
             event_queue=self.event_queue,
         )
-        self.subscriber = EventSubscriber(self.event_queue)
-
-        self.subscriber.subscribe(self.callback)
+        self.subscriber = EventSubscriber(self.event_queue, self.callback)
+        self.subscriber.setup()
         self.dispatcher.run_once()
 
         self.event_queue.publish(
@@ -50,9 +50,11 @@ class TestEventSubscriber(unittest.TestCase):
             publisher_id=self.__class__.__name__,
         )
         self.dispatcher.run_once()
-
         self.subscriber.unsubscribe()
         self.dispatcher.run_once()
+        self.subscriber.shutdown(do_unsubscribe=False)
+        with self.assertRaises(queue.Empty):
+            self.dispatcher.run_once()
         self.dispatcher_shutdown.set()
 
     def callback(self, ev: Dict[str, Any]) -> None:
diff --git a/tests/core/test_listener.py b/tests/core/test_listener.py
new file mode 100644
index 0000000..33cf990
--- /dev/null
+++ b/tests/core/test_listener.py
@@ -0,0 +1,69 @@
+# -*- coding: utf-8 -*-
+"""
+    proxy.py
+    ~~~~~~~~
+    ⚡⚡⚡ Fast, Lightweight, Pluggable, TLS interception capable proxy server focused on
+    Network monitoring, controls & Application development, testing, debugging.
+
+    :copyright: (c) 2013-present by Abhinav Singh and contributors.
+    :license: BSD, see LICENSE for more details.
+"""
+import os
+import socket
+import tempfile
+import unittest
+
+from unittest import mock
+
+from proxy.core.acceptor import Listener
+from proxy.common.flag import FlagParser
+
+
+class TestListener(unittest.TestCase):
+
+    @mock.patch('socket.socket')
+    def test_setup_and_teardown(self, mock_socket: mock.Mock) -> None:
+        sock = mock_socket.return_value
+        flags = FlagParser.initialize(port=0)
+        listener = Listener(flags)
+        listener.setup()
+        mock_socket.assert_called_with(
+            socket.AF_INET6 if flags.hostname.version == 6 else socket.AF_INET,
+            socket.SOCK_STREAM,
+        )
+        sock.setsockopt.assert_called_with(
+            socket.SOL_SOCKET, socket.SO_REUSEADDR, 1,
+        )
+        sock.bind.assert_called_with(
+            (str(flags.hostname), 0),
+        )
+        sock.listen.assert_called_with(flags.backlog)
+        sock.setblocking.assert_called_with(False)
+
+        listener.shutdown()
+        sock.close.assert_called_once()
+
+    @unittest.skipIf(os.name == 'nt', 'AF_UNIX not available on windows')
+    @mock.patch('os.remove')
+    @mock.patch('socket.socket')
+    def test_unix_path_listener(self, mock_socket: mock.Mock, mock_remove: mock.Mock) -> None:
+        sock = mock_socket.return_value
+        sock_path = os.path.join(tempfile.gettempdir(), 'proxy.sock')
+        flags = FlagParser.initialize(unix_socket_path=sock_path)
+        listener = Listener(flags)
+        listener.setup()
+
+        mock_socket.assert_called_with(
+            socket.AF_UNIX,
+            socket.SOCK_STREAM,
+        )
+        sock.setsockopt.assert_called_with(
+            socket.SOL_SOCKET, socket.SO_REUSEADDR, 1,
+        )
+        sock.bind.assert_called_with(sock_path)
+        sock.listen.assert_called_with(flags.backlog)
+        sock.setblocking.assert_called_with(False)
+
+        listener.shutdown()
+        mock_remove.assert_called_once_with(sock_path)
+        sock.close.assert_called_once()
diff --git a/tests/test_main.py b/tests/test_main.py
index 21137b6..27ec26a 100644
--- a/tests/test_main.py
+++ b/tests/test_main.py
@@ -8,6 +8,8 @@
     :copyright: (c) 2013-present by Abhinav Singh and contributors.
     :license: BSD, see LICENSE for more details.
 """
+import os
+import tempfile
 import unittest
 
 from unittest import mock
@@ -65,44 +67,68 @@ class TestMain(unittest.TestCase):
         mock_args.enable_dashboard = DEFAULT_ENABLE_DASHBOARD
         mock_args.work_klass = DEFAULT_WORK_KLASS
 
+    @mock.patch('os.remove')
+    @mock.patch('os.path.exists')
+    @mock.patch('builtins.open')
     @mock.patch('time.sleep')
     @mock.patch('proxy.proxy.FlagParser.initialize')
     @mock.patch('proxy.proxy.EventManager')
     @mock.patch('proxy.proxy.AcceptorPool')
     @mock.patch('proxy.proxy.ThreadlessPool')
+    @mock.patch('proxy.proxy.Listener')
     def test_entry_point(
             self,
+            mock_listener: mock.Mock,
             mock_executor_pool: mock.Mock,
             mock_acceptor_pool: mock.Mock,
             mock_event_manager: mock.Mock,
             mock_initialize: mock.Mock,
             mock_sleep: mock.Mock,
+            mock_open: mock.Mock,
+            mock_exists: mock.Mock,
+            mock_remove: mock.Mock,
     ) -> None:
+        pid_file = os.path.join(tempfile.gettempdir(), 'pid')
         mock_sleep.side_effect = KeyboardInterrupt()
         mock_initialize.return_value.enable_events = False
+        mock_initialize.return_value.pid_file = pid_file
         entry_point()
         mock_event_manager.assert_not_called()
+        mock_listener.assert_called_once_with(
+            flags=mock_initialize.return_value,
+        )
         mock_executor_pool.assert_called_once_with(
             flags=mock_initialize.return_value,
             event_queue=None,
         )
         mock_acceptor_pool.assert_called_once_with(
             flags=mock_initialize.return_value,
-            event_queue=None,
+            listener=mock_listener.return_value,
             executor_queues=mock_executor_pool.return_value.work_queues,
             executor_pids=mock_executor_pool.return_value.work_pids,
+            event_queue=None,
         )
-        mock_acceptor_pool.return_value.setup.assert_called()
-        mock_acceptor_pool.return_value.shutdown.assert_called()
+        mock_acceptor_pool.return_value.setup.assert_called_once()
+        mock_acceptor_pool.return_value.shutdown.assert_called_once()
+        mock_listener.return_value.shutdown.assert_called_once()
         mock_sleep.assert_called()
 
+        mock_open.assert_called_with(pid_file, 'wb')
+        mock_open.return_value.__enter__.return_value.write.assert_called_with(
+            bytes_(os.getpid()),
+        )
+        mock_exists.assert_called_with(pid_file)
+        mock_remove.assert_called_with(pid_file)
+
     @mock.patch('time.sleep')
     @mock.patch('proxy.proxy.FlagParser.initialize')
     @mock.patch('proxy.proxy.EventManager')
     @mock.patch('proxy.proxy.AcceptorPool')
     @mock.patch('proxy.proxy.ThreadlessPool')
+    @mock.patch('proxy.proxy.Listener')
     def test_main_with_no_flags(
             self,
+            mock_listener: mock.Mock,
             mock_executor_pool: mock.Mock,
             mock_acceptor_pool: mock.Mock,
             mock_event_manager: mock.Mock,
@@ -113,18 +139,23 @@ class TestMain(unittest.TestCase):
         mock_initialize.return_value.enable_events = False
         main()
         mock_event_manager.assert_not_called()
+        mock_listener.assert_called_once_with(
+            flags=mock_initialize.return_value,
+        )
         mock_executor_pool.assert_called_once_with(
             flags=mock_initialize.return_value,
             event_queue=None,
         )
         mock_acceptor_pool.assert_called_once_with(
             flags=mock_initialize.return_value,
-            event_queue=None,
+            listener=mock_listener.return_value,
             executor_queues=mock_executor_pool.return_value.work_queues,
             executor_pids=mock_executor_pool.return_value.work_pids,
+            event_queue=None,
         )
-        mock_acceptor_pool.return_value.setup.assert_called()
-        mock_acceptor_pool.return_value.shutdown.assert_called()
+        mock_acceptor_pool.return_value.setup.assert_called_once()
+        mock_acceptor_pool.return_value.shutdown.assert_called_once()
+        mock_listener.return_value.shutdown.assert_called_once()
         mock_sleep.assert_called()
 
     @mock.patch('time.sleep')
@@ -132,8 +163,10 @@ class TestMain(unittest.TestCase):
     @mock.patch('proxy.proxy.EventManager')
     @mock.patch('proxy.proxy.AcceptorPool')
     @mock.patch('proxy.proxy.ThreadlessPool')
+    @mock.patch('proxy.proxy.Listener')
     def test_enable_events(
             self,
+            mock_listener: mock.Mock,
             mock_executor_pool: mock.Mock,
             mock_acceptor_pool: mock.Mock,
             mock_event_manager: mock.Mock,
@@ -146,18 +179,23 @@ class TestMain(unittest.TestCase):
         mock_event_manager.assert_called_once()
         mock_event_manager.return_value.setup.assert_called_once()
         mock_event_manager.return_value.shutdown.assert_called_once()
+        mock_listener.assert_called_once_with(
+            flags=mock_initialize.return_value,
+        )
         mock_executor_pool.assert_called_once_with(
             flags=mock_initialize.return_value,
             event_queue=mock_event_manager.return_value.queue,
         )
         mock_acceptor_pool.assert_called_once_with(
             flags=mock_initialize.return_value,
+            listener=mock_listener.return_value,
             event_queue=mock_event_manager.return_value.queue,
             executor_queues=mock_executor_pool.return_value.work_queues,
             executor_pids=mock_executor_pool.return_value.work_pids,
         )
-        mock_acceptor_pool.return_value.setup.assert_called()
-        mock_acceptor_pool.return_value.shutdown.assert_called()
+        mock_acceptor_pool.return_value.setup.assert_called_once()
+        mock_acceptor_pool.return_value.shutdown.assert_called_once()
+        mock_listener.return_value.shutdown.assert_called_once()
         mock_sleep.assert_called()
 
     @mock.patch('time.sleep')
@@ -166,8 +204,10 @@ class TestMain(unittest.TestCase):
     @mock.patch('proxy.proxy.EventManager')
     @mock.patch('proxy.proxy.AcceptorPool')
     @mock.patch('proxy.proxy.ThreadlessPool')
+    @mock.patch('proxy.proxy.Listener')
     def test_enable_dashboard(
             self,
+            mock_listener: mock.Mock,
             mock_executor_pool: mock.Mock,
             mock_acceptor_pool: mock.Mock,
             mock_event_manager: mock.Mock,
@@ -200,6 +240,7 @@ class TestMain(unittest.TestCase):
         mock_executor_pool.return_value.setup.assert_called_once()
         mock_acceptor_pool.assert_called_once()
         mock_acceptor_pool.return_value.setup.assert_called_once()
+        mock_listener.return_value.setup.assert_called_once()
 
     @mock.patch('time.sleep')
     @mock.patch('proxy.common.plugins.Plugins.load')
@@ -207,8 +248,10 @@ class TestMain(unittest.TestCase):
     @mock.patch('proxy.proxy.EventManager')
     @mock.patch('proxy.proxy.AcceptorPool')
     @mock.patch('proxy.proxy.ThreadlessPool')
+    @mock.patch('proxy.proxy.Listener')
     def test_enable_devtools(
             self,
+            mock_listener: mock.Mock,
             mock_executor_pool: mock.Mock,
             mock_acceptor_pool: mock.Mock,
             mock_event_manager: mock.Mock,
@@ -236,6 +279,7 @@ class TestMain(unittest.TestCase):
         mock_executor_pool.return_value.setup.assert_called_once()
         mock_acceptor_pool.assert_called_once()
         mock_acceptor_pool.return_value.setup.assert_called_once()
+        mock_listener.return_value.setup.assert_called_once()
 
     # def test_pac_file(self) -> None:
     #     pass
diff --git a/tests/test_set_open_file_limit.py b/tests/test_set_open_file_limit.py
index 35fcc30..2c55eee 100644
--- a/tests/test_set_open_file_limit.py
+++ b/tests/test_set_open_file_limit.py
@@ -18,10 +18,7 @@ if os.name != 'nt':
     import resource
 
 
-@unittest.skipIf(
-    os.name == 'nt',
-    'Open file limit tests disabled for Windows',
-)
+@unittest.skipIf(os.name == 'nt', 'Open file limit tests disabled for Windows')
 class TestSetOpenFileLimit(unittest.TestCase):
 
     @mock.patch('resource.getrlimit', return_value=(128, 1024))
diff --git a/tests/testing/test_embed.py b/tests/testing/test_embed.py
index 7fcd169..034ba68 100644
--- a/tests/testing/test_embed.py
+++ b/tests/testing/test_embed.py
@@ -21,9 +21,7 @@ from proxy.http.codes import httpStatusCodes
 from proxy.http.methods import httpMethods
 
 
-@unittest.skipIf(
-    os.name == 'nt', 'Disabled for Windows due to weird permission issues.',
-)
+@unittest.skipIf(os.name == 'nt', 'Disabled for Windows due to weird permission issues.')
 class TestProxyPyEmbedded(TestCase):
     """This test case is a demonstration of proxy.TestCase and also serves as
     integration test suite for proxy.py."""