[Middleware] HttpClientConnection preparation (#995)

* Turn usual suspects to warnings, not error

* Add `HttpClientConnection` skeleton

* Fix doc build

* Update references in http tests

* Make `work` core agnostic to work object construction by adding an abstract static method to `Work` interface called `create`

* Make mypy happy

* Fix tests broken due to change in how work objects are now constructed

* Doc ko bhi happy karo

Author: abhinavsingh

docs/conf.py

@@ -308,12 +308,14 @@
     (_py_class_role, 'unittest.result.TestResult'),
     (_py_class_role, 'UUID'),
     (_py_class_role, 'UpstreamConnectionPool'),
+    (_py_class_role, 'HttpClientConnection'),
     (_py_class_role, 'Url'),
     (_py_class_role, 'WebsocketFrame'),
     (_py_class_role, 'Work'),
     (_py_class_role, 'proxy.core.acceptor.work.Work'),
     (_py_class_role, 'connection.Connection'),
     (_py_class_role, 'EventQueue'),
+    (_py_class_role, 'T'),
     (_py_obj_role, 'proxy.core.work.threadless.T'),
     (_py_obj_role, 'proxy.core.work.work.T'),
     (_py_obj_role, 'proxy.core.base.tcp_server.T'),

examples/ssl_echo_server.py

@@ -9,7 +9,7 @@
     :license: BSD, see LICENSE for more details.
 """
 import time
-from typing import Optional
+from typing import Any, Optional
 
 from proxy import Proxy
 from proxy.core.base import BaseTcpServerHandler
@@ -20,6 +20,10 @@
 class EchoSSLServerHandler(BaseTcpServerHandler[TcpClientConnection]):
     """Wraps client socket during initialization."""
 
+    @staticmethod
+    def create(**kwargs: Any) -> TcpClientConnection:
+        return TcpClientConnection(**kwargs)
+
     def initialize(self) -> None:
         # Acceptors don't perform TLS handshake.  Perform the same
         # here using wrap_socket() utility.

examples/tcp_echo_server.py

@@ -9,7 +9,7 @@
     :license: BSD, see LICENSE for more details.
 """
 import time
-from typing import Optional
+from typing import Any, Optional
 
 from proxy import Proxy
 from proxy.core.base import BaseTcpServerHandler
@@ -19,6 +19,10 @@
 class EchoServerHandler(BaseTcpServerHandler[TcpClientConnection]):
     """Sets client socket to non-blocking during initialization."""
 
+    @staticmethod
+    def create(**kwargs: Any) -> TcpClientConnection:
+        return TcpClientConnection(**kwargs)
+
     def initialize(self) -> None:
         self.work.connection.setblocking(False)
 

proxy/core/acceptor/acceptor.py

@@ -224,7 +224,8 @@ def _work(self, conn: socket.socket, addr: Optional[Tuple[str, int]]) -> None:
                 ),
             )
             thread.start()
-            logger.debug(
+            # TODO: Move me into target method
+            logger.debug(   # pragma: no cover
                 'Dispatched work#{0}.{1}.{2} to worker#{3}'.format(
                     conn.fileno(), self.idd, self._total, index,
                 ),
@@ -237,6 +238,7 @@ def _work(self, conn: socket.socket, addr: Optional[Tuple[str, int]]) -> None:
                 event_queue=self.event_queue,
                 publisher_id=self.__class__.__name__,
             )
+            # TODO: Move me into target method
             logger.debug(   # pragma: no cover
                 'Started work#{0}.{1}.{2} in thread#{3}'.format(
                     conn.fileno(), self.idd, self._total, thread.ident,

proxy/core/base/tcp_tunnel.py

@@ -47,6 +47,10 @@ def __init__(self, *args: Any, **kwargs: Any) -> None:
     def handle_data(self, data: memoryview) -> Optional[bool]:
         pass    # pragma: no cover
 
+    @staticmethod
+    def create(**kwargs: Any) -> TcpClientConnection:
+        return TcpClientConnection(**kwargs)
+
     def initialize(self) -> None:
         self.work.connection.setblocking(False)
 

proxy/core/connection/pool.py

@@ -16,7 +16,7 @@
 import logging
 import selectors
 
-from typing import TYPE_CHECKING, Set, Dict, Tuple
+from typing import TYPE_CHECKING, Any, Set, Dict, Tuple
 
 from ...common.flag import flags
 from ...common.types import Readables, SelectableEvents, Writables
@@ -77,6 +77,10 @@ def __init__(self) -> None:
         self.connections: Dict[int, TcpServerConnection] = {}
         self.pools: Dict[Tuple[str, int], Set[TcpServerConnection]] = {}
 
+    @staticmethod
+    def create(**kwargs: Any) -> TcpServerConnection:
+        return TcpServerConnection(**kwargs)
+
     def acquire(self, addr: Tuple[str, int]) -> Tuple[bool, TcpServerConnection]:
         """Returns a reusable connection from the pool.
 
@@ -152,7 +156,7 @@ def add(self, addr: Tuple[str, int]) -> TcpServerConnection:
 
         NOTE: You must not use the returned connection, instead use `acquire`.
         """
-        new_conn = TcpServerConnection(addr[0], addr[1])
+        new_conn = self.create(host=addr[0], port=addr[1])
         new_conn.connect()
         self._add(new_conn)
         logger.debug(

proxy/core/ssh/handler.py

@@ -12,7 +12,7 @@
 
 from typing import TYPE_CHECKING, Tuple
 
-if TYPE_CHECKING:
+if TYPE_CHECKING:   # pragma: no cover
     try:
         from paramiko.channel import Channel
     except ImportError:

proxy/core/ssh/listener.py

@@ -16,7 +16,7 @@
 try:
     from paramiko import SSHClient, AutoAddPolicy
     from paramiko.transport import Transport
-    if TYPE_CHECKING:
+    if TYPE_CHECKING:   # pragma: no cover
         from paramiko.channel import Channel
 except ImportError:
     pass

proxy/core/work/threaded.py

@@ -12,25 +12,27 @@
 import argparse
 import threading
 
-from typing import TYPE_CHECKING, Optional, Tuple
+from typing import TYPE_CHECKING, Optional, Tuple, TypeVar
 
-from ..connection import TcpClientConnection
 from ..event import EventQueue, eventNames
 
 if TYPE_CHECKING:   # pragma: no cover
     from .work import Work
 
+T = TypeVar('T')
 
+
+# TODO: Add generic T
 def start_threaded_work(
         flags: argparse.Namespace,
         conn: socket.socket,
         addr: Optional[Tuple[str, int]],
         event_queue: Optional[EventQueue] = None,
         publisher_id: Optional[str] = None,
-) -> Tuple['Work[TcpClientConnection]', threading.Thread]:
+) -> Tuple['Work[T]', threading.Thread]:
     """Utility method to start a work in a new thread."""
     work = flags.work_klass(
-        TcpClientConnection(conn, addr),
+        flags.work_klass.create(conn=conn, addr=addr),
         flags=flags,
         event_queue=event_queue,
         upstream_conn_pool=None,

proxy/core/work/threadless.py

@@ -25,7 +25,6 @@
 from ...common.constants import DEFAULT_INACTIVE_CONN_CLEANUP_TIMEOUT, DEFAULT_SELECTOR_SELECT_TIMEOUT
 from ...common.constants import DEFAULT_WAIT_FOR_TASKS_TIMEOUT
 
-from ..connection import TcpClientConnection
 from ..event import eventNames
 
 if TYPE_CHECKING:   # pragma: no cover
@@ -138,7 +137,7 @@ def work_on_tcp_conn(
         )
         uid = '%s-%s-%s' % (self.iid, self._total, fileno)
         self.works[fileno] = self.flags.work_klass(
-            TcpClientConnection(
+            self.flags.work_klass.create(
                 conn=conn,
                 addr=addr,
             ),