Initialize manager in main thread for event queue

.gitignore

@@ -12,16 +12,16 @@
 
 coverage.xml
 proxy.py.iml
-*.pyc
 
+*.pyc
+*.egg-info
 *.csr
 *.crt
 *.key
-
 venv*
+
 cover
 htmlcov
 dist
 build
-
-*.egg-info
+proxy/public

proxy/common/constants.py

@@ -68,7 +68,7 @@
 DEFAULT_PLUGINS = ''
 DEFAULT_PORT = 8899
 DEFAULT_SERVER_RECVBUF_SIZE = DEFAULT_BUFFER_SIZE
-DEFAULT_STATIC_SERVER_DIR = PROXY_PY_DIR
+DEFAULT_STATIC_SERVER_DIR = os.path.join(PROXY_PY_DIR, "public")
 DEFAULT_THREADLESS = False
 DEFAULT_TIMEOUT = 10
 DEFAULT_VERSION = False

proxy/core/acceptor/pool.py

@@ -46,8 +46,11 @@ def __init__(self, flags: Flags, work_klass: Type[ThreadlessWork]) -> None:
         self.event_dispatcher: Optional[EventDispatcher] = None
         self.event_dispatcher_thread: Optional[threading.Thread] = None
         self.event_dispatcher_shutdown: Optional[threading.Event] = None
+        self.manager: Optional[multiprocessing.managers.SyncManager] = None
+
         if self.flags.enable_events:
-            self.event_queue = EventQueue()
+            self.manager = multiprocessing.Manager()
+            self.event_queue = EventQueue(self.manager.Queue())
 
     def listen(self) -> None:
         self.socket = socket.socket(self.flags.family, socket.SOCK_STREAM)

proxy/core/event.py

@@ -56,10 +56,8 @@ class EventQueue:
     7. Publisher ID (optional)  - Optionally, publishing entity unique name / ID
     """
 
-    def __init__(self) -> None:
-        super().__init__()
-        self.manager = multiprocessing.Manager()
-        self.queue = self.manager.Queue()
+    def __init__(self, queue: DictQueueType) -> None:
+        self.queue = queue
 
     def publish(
         self,

tests/core/test_event_dispatcher.py

@@ -24,7 +24,7 @@ class TestEventDispatcher(unittest.TestCase):
 
     def setUp(self) -> None:
         self.dispatcher_shutdown = threading.Event()
-        self.event_queue = EventQueue()
+        self.event_queue = EventQueue(multiprocessing.Manager().Queue())
         self.dispatcher = EventDispatcher(
             shutdown=self.dispatcher_shutdown,
             event_queue=self.event_queue)

tests/core/test_event_queue.py

@@ -17,13 +17,15 @@
 
 from proxy.core.event import EventQueue, eventNames
 
+MANAGER = multiprocessing.Manager()
+
 
 class TestCoreEvent(unittest.TestCase):
 
     @mock.patch('time.time')
     def test_publish(self, mock_time: mock.Mock) -> None:
         mock_time.return_value = 1234567
-        evq = EventQueue()
+        evq = EventQueue(MANAGER.Queue())
         evq.publish(
             request_id='1234',
             event_name=eventNames.WORK_STARTED,
@@ -41,15 +43,15 @@ def test_publish(self, mock_time: mock.Mock) -> None:
         })
 
     def test_subscribe(self) -> None:
-        evq = EventQueue()
+        evq = EventQueue(MANAGER.Queue())
         q = multiprocessing.Manager().Queue()
         evq.subscribe('1234', q)
         ev = evq.queue.get()
         self.assertEqual(ev['event_name'], eventNames.SUBSCRIBE)
         self.assertEqual(ev['event_payload']['sub_id'], '1234')
 
     def test_unsubscribe(self) -> None:
-        evq = EventQueue()
+        evq = EventQueue(MANAGER.Queue())
         evq.unsubscribe('1234')
         ev = evq.queue.get()
         self.assertEqual(ev['event_name'], eventNames.UNSUBSCRIBE)

tests/core/test_event_subscriber.py

@@ -11,6 +11,7 @@
 import os
 import threading
 import unittest
+import multiprocessing
 from typing import Dict, Any
 
 from unittest import mock
@@ -26,7 +27,7 @@ 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.event_queue = EventQueue(multiprocessing.Manager().Queue())
         self.dispatcher = EventDispatcher(
             shutdown=self.dispatcher_shutdown,
             event_queue=self.event_queue)