DescriptorsHandlerMixin and Descriptors, SelectableEvents types (#938)

* Add `Descriptors` type

* Add a `DescriptorsHandlerMixin` class used throughout the http framework

* Remove dependency upon `HasFileno` ie `typing_extension` too

* Define `SelectableEvents` type

* Fix doc

* Blank line

* Remove dep on `typing-extensions`

* Discover base plugin class

* await on now async handlers

Author: abhinavsingh

Makefile

@@ -98,7 +98,6 @@ lib-clean:
 lib-dep:
 	pip install --upgrade pip && \
 	pip install \
-		-r requirements.txt \
 		-r requirements-testing.txt \
 		-r requirements-release.txt \
 		-r requirements-tunnel.txt && \

docs/conf.py

@@ -299,6 +299,7 @@
     (_py_class_role, 'proxy.core.acceptor.threadless.T'),
     (_py_class_role, 'proxy.core.acceptor.work.T'),
     (_py_class_role, 'queue.Queue[Any]'),
+    (_py_class_role, 'SelectableEvents'),
     (_py_class_role, 'TcpClientConnection'),
     (_py_class_role, 'TcpServerConnection'),
     (_py_class_role, 'unittest.case.TestCase'),

examples/web_scraper.py

@@ -10,12 +10,10 @@
 """
 import time
 
-from typing import Dict
-
 from proxy import Proxy
 from proxy.core.acceptor import Work
 from proxy.core.connection import TcpClientConnection
-from proxy.common.types import Readables, Writables
+from proxy.common.types import Readables, SelectableEvents, Writables
 
 
 class WebScraper(Work[TcpClientConnection]):
@@ -40,7 +38,7 @@ class WebScraper(Work[TcpClientConnection]):
     only PUBSUB protocol.
     """
 
-    async def get_events(self) -> Dict[int, int]:
+    async def get_events(self) -> SelectableEvents:
         """Return sockets and events (read or write) that we are interested in."""
         return {}
 

helper/monitor_open_files.sh

@@ -27,8 +27,8 @@ pgrep -P "$PROXY_PY_PID" | while read -r acceptorPid; do
   OPEN_FILES_BY_ACCEPTOR=$(lsof -p "$acceptorPid" | wc -l)
   echo "[$acceptorPid] Acceptor process: $OPEN_FILES_BY_ACCEPTOR"
 
-  pgrep -P "$acceptorPid" | while read -r threadlessPid; do
-    OPEN_FILES_BY_THREADLESS=$(lsof -p "$threadlessPid" | wc -l)
-    echo "  [$threadlessPid] Threadless process: $OPEN_FILES_BY_THREADLESS"
+  pgrep -P "$acceptorPid" | while read -r childPid; do
+    OPEN_FILES_BY_CHILD_PROC=$(lsof -p "$childPid" | wc -l)
+    echo "  [$childPid] child process: $OPEN_FILES_BY_CHILD_PROC"
   done
 done

proxy/common/plugins.py

@@ -9,7 +9,6 @@
     :license: BSD, see LICENSE for more details.
 """
 import os
-import abc
 import logging
 import inspect
 import itertools
@@ -72,20 +71,19 @@ def load(
             klass, module_name = Plugins.importer(plugin_)
             assert klass and module_name
             mro = list(inspect.getmro(klass))
-            mro.reverse()
-            iterator = iter(mro)
-            try:
-                while next(iterator) is not abc.ABC:
-                    pass
-                base_klass = next(iterator)
-                if klass not in p[bytes_(base_klass.__name__)]:
-                    p[bytes_(base_klass.__name__)].append(klass)
-                logger.info('Loaded plugin %s.%s', module_name, klass.__name__)
-            except StopIteration:
-                logger.warn(
-                    '%s is NOT a valid plugin',
-                    text_(plugin_),
-                )
+            # Find the base plugin class that
+            # this plugin_ is implementing
+            found = False
+            for base_klass in mro:
+                if bytes_(base_klass.__name__) in p:
+                    found = True
+                    break
+            if not found:
+                raise ValueError('%s is NOT a valid plugin' % text_(plugin_))
+            if klass not in p[bytes_(base_klass.__name__)]:
+                p[bytes_(base_klass.__name__)].append(klass)
+            logger.info('Loaded plugin %s.%s', module_name, klass.__name__)
+        # print(p)
         return p
 
     @staticmethod

proxy/common/types.py

@@ -10,17 +10,8 @@
 """
 import queue
 import ipaddress
-import sys
 
-from typing import TYPE_CHECKING, Dict, Any, List, Union
-
-# NOTE: Using try/except causes linting problems which is why it's necessary
-# NOTE: to use this mypy/pylint idiom for py36-py38 compatibility
-# Ref: https://github.com/python/typeshed/issues/3500#issuecomment-560958608
-if sys.version_info >= (3, 8):
-    from typing import Protocol
-else:
-    from typing_extensions import Protocol
+from typing import TYPE_CHECKING, Dict, Any, List, Tuple, Union
 
 
 if TYPE_CHECKING:
@@ -29,11 +20,11 @@
     DictQueueType = queue.Queue
 
 
-class HasFileno(Protocol):
-    def fileno(self) -> int:
-        ...     # pragma: no cover
-
+Selectable = int
+Selectables = List[Selectable]
+SelectableEvents = Dict[Selectable, int]    # Values are event masks
+Readables = Selectables
+Writables = Selectables
+Descriptors = Tuple[Readables, Writables]
 
-Readables = List[Union[int, HasFileno]]
-Writables = List[Union[int, HasFileno]]
 IpAddress = Union[ipaddress.IPv4Address, ipaddress.IPv6Address]

proxy/core/acceptor/threadless.py

@@ -21,7 +21,7 @@
 from typing import Any, Dict, Optional, Tuple, List, Set, Generic, TypeVar, Union
 
 from ...common.logger import Logger
-from ...common.types import Readables, Writables
+from ...common.types import Readables, SelectableEvents, Writables
 from ...common.constants import DEFAULT_INACTIVE_CONN_CLEANUP_TIMEOUT, DEFAULT_SELECTOR_SELECT_TIMEOUT
 from ...common.constants import DEFAULT_WAIT_FOR_TASKS_TIMEOUT
 
@@ -82,7 +82,7 @@ def __init__(
             # work_id
             int,
             # fileno, mask
-            Dict[int, int],
+            SelectableEvents,
         ] = {}
         self.wait_timeout: float = DEFAULT_WAIT_FOR_TASKS_TIMEOUT
         self.cleanup_inactive_timeout: float = DEFAULT_INACTIVE_CONN_CLEANUP_TIMEOUT
@@ -288,9 +288,9 @@ async def _selected_events(self) -> Tuple[
             if key.data not in work_by_ids:
                 work_by_ids[key.data] = ([], [])
             if mask & selectors.EVENT_READ:
-                work_by_ids[key.data][0].append(key.fileobj)
+                work_by_ids[key.data][0].append(key.fd)
             if mask & selectors.EVENT_WRITE:
-                work_by_ids[key.data][1].append(key.fileobj)
+                work_by_ids[key.data][1].append(key.fd)
         return (work_by_ids, new_work_available)
 
     async def _wait_for_tasks(self) -> Set['asyncio.Task[bool]']:

proxy/core/acceptor/work.py

@@ -14,12 +14,12 @@
 """
 import argparse
 
-from abc import ABC, abstractmethod
 from uuid import uuid4
+from abc import ABC, abstractmethod
 from typing import Optional, Dict, Any, TypeVar, Generic, TYPE_CHECKING
 
 from ..event import eventNames, EventQueue
-from ...common.types import Readables, Writables
+from ...common.types import Readables, SelectableEvents, Writables
 
 if TYPE_CHECKING:
     from ..connection import UpstreamConnectionPool
@@ -48,7 +48,7 @@ def __init__(
         self.upstream_conn_pool = upstream_conn_pool
 
     @abstractmethod
-    async def get_events(self) -> Dict[int, int]:
+    async def get_events(self) -> SelectableEvents:
         """Return sockets and events (read or write) that we are interested in."""
         return {}   # pragma: no cover
 

proxy/core/base/tcp_server.py

@@ -16,11 +16,11 @@
 import selectors
 
 from abc import abstractmethod
-from typing import Dict, Any, Optional
+from typing import Any, Optional
 
 from ...core.acceptor import Work
 from ...core.connection import TcpClientConnection
-from ...common.types import Readables, Writables
+from ...common.types import Readables, SelectableEvents, Writables
 
 logger = logging.getLogger(__name__)
 
@@ -61,7 +61,7 @@ def handle_data(self, data: memoryview) -> Optional[bool]:
         """Optionally return True to close client connection."""
         pass    # pragma: no cover
 
-    async def get_events(self) -> Dict[int, int]:
+    async def get_events(self) -> SelectableEvents:
         events = {}
         # We always want to read from client
         # Register for EVENT_READ events

proxy/core/base/tcp_tunnel.py

@@ -12,10 +12,10 @@
 import selectors
 
 from abc import abstractmethod
-from typing import Any, Optional, Dict
+from typing import Any, Optional
 
 from ...http.parser import HttpParser, httpParserTypes
-from ...common.types import Readables, Writables
+from ...common.types import Readables, SelectableEvents, Writables
 from ...common.utils import text_
 
 from ..connection import TcpServerConnection
@@ -60,9 +60,9 @@ def shutdown(self) -> None:
             self.upstream.close()
         super().shutdown()
 
-    async def get_events(self) -> Dict[int, int]:
+    async def get_events(self) -> SelectableEvents:
         # Get default client events
-        ev: Dict[int, int] = await super().get_events()
+        ev: SelectableEvents = await super().get_events()
         # Read from server if we are connected
         if self.upstream and self.upstream._conn is not None:
             ev[self.upstream.connection.fileno()] = selectors.EVENT_READ