[BaseRemoteExecutor] Base executor for processing work remotely (#1050)

* Abstract out FD based work implementation

* No need of local/remote abstractions

* fix type

* Add `BaseLocalExecutor`

* Fix lint and tests

* Add a `BaseRemoteExecutor`

Author: abhinavsingh

proxy/core/work/__init__.py

@@ -15,6 +15,7 @@
 from .pool import ThreadlessPool
 from .work import Work
 from .local import BaseLocalExecutor
+from .remote import BaseRemoteExecutor
 from .delegate import delegate_work_to_pool
 from .threaded import start_threaded_work
 from .threadless import Threadless
@@ -27,4 +28,5 @@
     'delegate_work_to_pool',
     'start_threaded_work',
     'BaseLocalExecutor',
+    'BaseRemoteExecutor',
 ]

proxy/core/work/remote.py

@@ -0,0 +1,35 @@
+# -*- 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 asyncio
+from typing import Any, Optional
+from multiprocessing import connection
+
+from .threadless import Threadless
+
+
+class BaseRemoteExecutor(Threadless[connection.Connection]):
+    """A threadless executor implementation which receives work over a connection."""
+
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        super().__init__(*args, **kwargs)
+        self._loop: Optional[asyncio.AbstractEventLoop] = None
+
+    @property
+    def loop(self) -> Optional[asyncio.AbstractEventLoop]:
+        if self._loop is None:
+            self._loop = asyncio.get_event_loop_policy().get_event_loop()
+        return self._loop
+
+    def work_queue_fileno(self) -> Optional[int]:
+        return self.work_queue.fileno()
+
+    def close_work_queue(self) -> None:
+        self.work_queue.close()