net: Drop unused transport/protocol methods; rename Transport->NetTransport; add NetProtocol

kafka/future.py

@@ -19,7 +19,7 @@ class Future:
     selector), which subclasses ``Future`` and adds ``__await__``. Keeping
     ``__await__`` off the base makes the invariant type-enforced -- awaiting a
     plain handoff ``Future`` raises immediately rather than silently working on
-    one backend and breaking on another. See ``kafka.net.backend.BackendFuture``
+    one backend and breaking on another. See ``kafka.net.backend.NetBackendFuture``
     for the awaitable contract.
     """
     __slots__ = ('is_done', 'value', 'exception', '_callbacks', '_errbacks', '_lock')

kafka/net/asyncio_backend.py

@@ -23,7 +23,7 @@ class AsyncioFuture(Future):
 
     Inherits ``kafka.future.Future``'s callback core and overrides only
     ``__await__`` to bridge to an ``asyncio.Future`` so an asyncio Task can
-    await it. Per the BackendFuture contract this is created and resolved on
+    await it. Per the NetBackendFuture contract this is created and resolved on
     the loop thread only; a fresh asyncio.Future is minted per awaiter, so
     fan-out (multiple awaiters / callbacks) is preserved.
     """
@@ -392,9 +392,6 @@ def data_received(self, data):
         self.transport._bump_read()
         self._conn.data_received(data)
 
-    def eof_received(self):
-        return self._conn.eof_received()
-
     def connection_lost(self, exc):
         self._conn.connection_lost(exc)
 
@@ -411,7 +408,7 @@ class _AsyncioTransport:
     Wraps an asyncio transport + its protocol adapter and exposes the surface
     KafkaConnection / the manager drive (a superset of the NetBackend Transport
     protocol): write/close/abort/is_closing, pause/resume_reading, set/get
-    protocol, host/host_port/getPeer, and last_activity for idle sweeping.
+    protocol, host/host_port/get_peer, and last_activity for idle sweeping.
     """
 
     def __init__(self, transport, host, port):
@@ -461,7 +458,7 @@ def resume_reading(self):
         except (RuntimeError, AttributeError):
             pass
 
-    def getPeer(self):
+    def get_peer(self):
         peer = self._t.get_extra_info('peername')
         return peer if peer is not None else (self.host, self._port)
 

kafka/net/backend.py

@@ -9,7 +9,7 @@
 implementation; an asyncio backend (and eventually Twisted) implements the
 same surface so it can be swapped in via ``net=`` without touching core code.
 
-The :class:`BackendFuture` contract is the surface of the
+The :class:`NetBackendFuture` contract is the surface of the
 loop-awaitable futures a backend hands out from ``net.create_future()``. The
 selector's implementation is ``kafka.net.selector.SelectorFuture``; an asyncio
 (and eventually Twisted) backend supplies its own.
@@ -43,15 +43,15 @@
 * **Timing** -- ``sleep`` (backend-specific awaitable; core coroutines await it).
 * **Connection** -- ``create_connection`` (returns a :class:`Transport`).
 * **Cross-thread bridge** -- ``run`` (schedule on the loop, block the caller).
-* **Future factory** -- ``create_future`` (see ``BackendFuture``).
+* **Future factory** -- ``create_future`` (see ``NetBackendFuture``).
 * **Cross-thread wake** -- ``wakeup``.
 """
 import importlib
 from typing import Any, Callable, Optional, Protocol, Sequence, Tuple, runtime_checkable
 
 
 @runtime_checkable
-class BackendFuture(Protocol):
+class NetBackendFuture(Protocol):
     """Contract for the awaitable futures returned by ``net.create_future()``.
 
     A pluggable async backend (the kafka.net selector, asyncio, Twisted, ...)
@@ -60,7 +60,7 @@ class BackendFuture(Protocol):
     backends. The selector's ``SelectorFuture`` is the reference implementation:
     it subclasses the thread-safe ``kafka.future.Future`` (the portable callback
     core) and adds ``__await__``. A plain ``Future`` is deliberately NOT a
-    BackendFuture -- it has no ``__await__`` -- so awaiting a cross-thread
+    NetBackendFuture -- it has no ``__await__`` -- so awaiting a cross-thread
     handoff future fails loudly instead of silently working on one backend.
 
     Pinned semantics -- the three axes where backends could otherwise diverge:
@@ -93,28 +93,27 @@ class BackendFuture(Protocol):
     exception: Any
 
     def __await__(self): ...
-    def success(self, value: Any) -> 'BackendFuture': ...
-    def failure(self, e: Any) -> 'BackendFuture': ...
-    def add_callback(self, f: Callable, *args: Any, **kwargs: Any) -> 'BackendFuture': ...
-    def add_errback(self, f: Callable, *args: Any, **kwargs: Any) -> 'BackendFuture': ...
-    def add_both(self, f: Callable, *args: Any, **kwargs: Any) -> 'BackendFuture': ...
-    def chain(self, future: 'BackendFuture') -> 'BackendFuture': ...
+    def success(self, value: Any) -> 'NetBackendFuture': ...
+    def failure(self, e: Any) -> 'NetBackendFuture': ...
+    def add_callback(self, f: Callable, *args: Any, **kwargs: Any) -> 'NetBackendFuture': ...
+    def add_errback(self, f: Callable, *args: Any, **kwargs: Any) -> 'NetBackendFuture': ...
+    def add_both(self, f: Callable, *args: Any, **kwargs: Any) -> 'NetBackendFuture': ...
+    def chain(self, future: 'NetBackendFuture') -> 'NetBackendFuture': ...
     def succeeded(self) -> bool: ...
     def failed(self) -> bool: ...
 
 
 @runtime_checkable
-class Transport(Protocol):
-    """The transport surface a backend's ``create_connection`` returns.
+class NetTransport(Protocol):
+    """The transport surface used by NetProtocol / NetBackend.
 
     The subset of the ``asyncio.Transport`` / Twisted ``ITransport`` surface
     that ``KafkaConnection`` actually drives (plus ``last_activity``, which the
     manager reads to sweep idle connections). A backend's ``create_connection``
     builds one and wires it to the conn. The selector's ``KafkaTCPTransport``
     and an asyncio-transport adapter both satisfy it.
     """
-
-    # Monotonic timestamp of the last read/write; manager idle-sweeping reads it.
+    # Monotonic timestamp of the last read/write; used for idle-sweeping
     last_activity: float
 
     def write(self, data: bytes) -> None: ...
@@ -124,6 +123,20 @@ def is_closing(self) -> bool: ...
     def pause_reading(self) -> None: ...
     def resume_reading(self) -> None: ...
     def host_port(self) -> Tuple[str, int]: ...
+    def get_protocol(self) -> 'NetProtocol': ...
+    def set_protocol(self, protocol: 'NetProtocol') -> None: ...
+
+
+@runtime_checkable
+class NetProtocol(Protocol):
+    # Monotonic timestamp of the last read/write; used for idle-sweeping
+    last_activity: float
+
+    def connection_made(self, transport: NetTransport) -> None: ...
+    def data_received(self, data: bytes) -> None: ...
+    def connection_lost(self, exc: Any = None) -> None: ...
+    def pause_writing(self) -> None: ...
+    def resume_writing(self) -> None: ...
 
 
 @runtime_checkable
@@ -155,7 +168,7 @@ def call_soon(self, task: Any) -> Any:
     def call_soon_threadsafe(self, callback: Any) -> Any:
         """``call_soon`` from another thread; wakes the loop."""
 
-    def call_soon_with_future(self, coro: Any, *args: Any) -> BackendFuture:
+    def call_soon_with_future(self, coro: Any, *args: Any) -> NetBackendFuture:
         """Schedule ``coro`` and return a future that resolves with its result."""
 
     def call_at(self, when: float, task: Any) -> Any:
@@ -174,7 +187,7 @@ def sleep(self, delay: float) -> Any:
     # --- connection seam --------------------------------------------------
     async def create_connection(
         self,
-        protocol: Any,
+        protocol: NetProtocol,
         host: str,
         port: int,
         *,
@@ -211,8 +224,8 @@ def run(self, coro: Any, *args: Any, timeout_ms: Optional[float] = None) -> Any:
         """
 
     # --- future factory ---------------------------------------------------
-    def create_future(self) -> BackendFuture:
-        """Create a loop-awaitable future (see ``BackendFuture``)."""
+    def create_future(self) -> NetBackendFuture:
+        """Create a loop-awaitable future (see ``NetBackendFuture``)."""
 
     # --- misc -------------------------------------------------------------
     def wakeup(self) -> None:

kafka/net/connection.py

@@ -221,15 +221,6 @@ def data_received(self, data):
             self.unpause('max_in_flight')
         self._reauth.on_response_processed()
 
-    def eof_received(self):
-        """ Called when the other end calls write_eof() or equivalent.
-
-        If this returns a false value (including None), the transport
-        will close itself.  If it returns a true value, closing the
-        transport is up to the protocol.
-        """
-        return False
-
     def connection_lost(self, exc):
         """ Called when the connection is lost or closed.
 
@@ -283,7 +274,7 @@ def connection_made(self, transport):
         self.initializing = True
         self.transport.resume_reading()
         try:
-            log_prefix = 'node=%s[%s:%s]' % (self.node_id, *self.transport.getPeer()[0:2])
+            log_prefix = 'node=%s[%s:%s]' % (self.node_id, *self.transport.get_peer()[0:2])
         except Exception:
             log.exception('Failed to build connection log_prefix')
             log_prefix = ''
@@ -456,7 +447,7 @@ async def _sasl_authenticate(self, timeout_at=None):
         # Prefer the configured hostname (stored on the transport) so that
         # mechanisms like GSSAPI construct service principals against the
         # user-supplied name, not whichever IP getaddrinfo handed us.
-        sasl_host = self.transport.host if self.transport.host else self.transport.getPeer()[0]
+        sasl_host = self.transport.host if self.transport.host else self.transport.get_peer()[0]
         mechanism = get_sasl_mechanism(self.config['sasl_mechanism'])(
             host=sasl_host, **self.config)
 

kafka/net/selector.py

@@ -43,7 +43,7 @@ def _initialize_coro(maybe_coro):
 
 
 class SelectorFuture(Future):
-    """The NetworkSelector's loop-awaitable future (see backend.BackendFuture).
+    """The NetworkSelector's loop-awaitable future (see backend.NetBackendFuture).
 
     ``kafka.future.Future`` is the thread-safe callback/handoff core with no
     ``__await__``; ``SelectorFuture`` adds it: ``yield self`` suspends the
@@ -550,7 +550,7 @@ def reschedule(self, when, task):
         return task
 
     def create_future(self):
-        """Create a loop-awaitable future (see backend.BackendFuture).
+        """Create a loop-awaitable future (see backend.NetBackendFuture).
 
         Portability seam for pluggable backends: core coroutines call this
         instead of constructing ``Future`` directly, so an alternate backend

kafka/net/transport.py

@@ -32,7 +32,6 @@ def __init__(self, net, sock, host=None):
     def last_activity(self):
         return max(self.last_write, self.last_read)
 
-    # AsyncIO
     def is_closing(self):
         """Return True if the transport is closing or closed."""
         return self._closed
@@ -128,39 +127,6 @@ def _sock_recv(self):
         recvd_data = b''.join(recvd)
         return recvd_data, err
 
-    """Interface for write-only transports."""
-
-    def set_write_buffer_limits(self, high=None, low=None):
-        """Set the high- and low-water limits for write flow control.
-
-        These two values control when to call the protocol's
-        pause_writing() and resume_writing() methods.  If specified,
-        the low-water limit must be less than or equal to the
-        high-water limit.  Neither value can be negative.
-
-        The defaults are implementation-specific.  If only the
-        high-water limit is given, the low-water limit defaults to an
-        implementation-specific value less than or equal to the
-        high-water limit.  Setting high to zero forces low to zero as
-        well, and causes pause_writing() to be called whenever the
-        buffer becomes non-empty.  Setting low to zero causes
-        resume_writing() to be called only once the buffer is empty.
-        Use of zero for either limit is generally sub-optimal as it
-        reduces opportunities for doing I/O and computation
-        concurrently.
-        """
-        raise NotImplementedError
-
-    def get_write_buffer_size(self):
-        """Return the current size of the write buffer."""
-        raise NotImplementedError
-
-    def get_write_buffer_limits(self):
-        """Get the high and low watermarks for write flow control.
-        Return a tuple (low, high) where low and high are
-        positive number of bytes."""
-        raise NotImplementedError
-
     def write(self, data):
         """Write some data bytes to the transport.
 
@@ -177,16 +143,6 @@ def write(self, data):
             self._write_task =  self._net.call_soon(self._write_to_sock)
         return len(data)
 
-    def writelines(self, list_of_data):
-        """Write a list (or any iterable) of data bytes to the transport."""
-        if not self._write or self._closed:
-            raise RuntimeError('Transport closed for writes')
-        self._write_buffer.extend(list_of_data)
-        if not self._writing:
-            self._writing = True
-            self._write_task = self._net.call_soon(self._write_to_sock)
-        return sum(len(data) for data in list_of_data)
-
     async def _write_to_sock(self):
         try:
             while self._write_buffer:
@@ -229,22 +185,6 @@ def _sock_send(self):
                     return total_bytes, Errors.KafkaConnectionError(e)
         return total_bytes, None
 
-    def write_eof(self):
-        """Close the write end after flushing buffered data.
-
-        (This is like typing ^D into a UNIX program reading from stdin.)
-
-        Data may still be received.
-        """
-        log.debug('%s: write_eof', self)
-        self._write = False
-        if not self._write_buffer:
-            self._sock.shutdown(socket.SHUT_WR)
-
-    def can_write_eof(self):
-        """Return True if this transport supports write_eof(), False if not."""
-        return True
-
     def abort(self, error=None):
         """Close the transport immediately.
 
@@ -282,19 +222,7 @@ def _close(self, error=None):
         if proto is not None:
             proto.connection_lost(error)
 
-  # Twisted
-    def abortConnection(self):
-        """Close the connection abruptly."""
-        return self.abort()
-
-    def getHost(self):
-        """Similar to getPeer, but returns an address describing this side of the connection.
-
-        Returns IPv4Address or IPv6Address.
-        """
-        return self._sock.getsockname()
-
-    def getPeer(self):
+    def get_peer(self):
         """Get the remote address of this connection.
 
         Treat this method with caution. It is the unfortunate result of the CGI and Jabber standards,
@@ -305,49 +233,6 @@ def getPeer(self):
         """
         return self._sock.getpeername()
 
-    def getTcpKeepAlive(self):
-        """Return if SO_KEEPALIVE is enabled."""
-        return self._sock.getsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE)
-
-    def getTcpNoDelay(self):
-        """Return if TCP_NODELAY is enabled."""
-        return self._sock.getsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY)
-
-    def loseWriteConnection(self):
-        """Half-close the write side of a TCP connection."""
-        return self.write_eof()
-
-    def setTcpKeepAlive(self, enabled):
-        """Enable/disable SO_KEEPALIVE."""
-        return self._sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, enabled)
-
-    def setTcpNoDelay(self, enabled):
-        """Enable/disable TCP_NODELAY."""
-        return self._sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, enabled)
-
-    def loseConnection(self):
-        """Close my connection, after writing all pending data.
-
-        Note that if there is a registered producer on a transport it will not be closed until the producer has been unregistered.
-        """
-        return self.close()
-
-    #def write(self, data):
-    #    """Write some data to the physical connection, in sequence, in a non-blocking fashion.
-    #
-    #    If possible, make sure that it is all written. No data will ever be lost,
-    #    although (obviously) the connection may be closed before it all gets through.
-    #    """
-    #    pass
-
-    def writeSequence(self, data):
-        """Write an iterable of byte strings to the physical connection.
-
-        If possible, make sure that all of the data is written to the socket at once,
-        without first copying it all into a single byte string.
-        """
-        return self.writelines(data)
-
     async def handshake(self):
         log.info('%s: connected to %s', self, self._sock)