Merge pull request #15 from dapper91/dev

- aliveness checking extracted to mixins.

Author: dapper91

CHANGELOG.rst

@@ -1,6 +1,12 @@
 Changelog
 =========
 
+0.4.1 (2023-08-16)
+------------------
+
+- aliveness checking extracted to mixins.
+
+
 0.4.0 (2023-08-15)
 ------------------
 

generic_connection_pool/contrib/socket.py

@@ -7,7 +7,7 @@
 import socket
 from ipaddress import IPv4Address, IPv6Address
 from ssl import SSLContext, SSLSocket
-from typing import Generator, Optional, Tuple, Union
+from typing import Generator, Generic, Optional, Tuple, Union
 
 from generic_connection_pool.threading import BaseConnectionManager, EndpointT
 
@@ -46,7 +46,29 @@ def socket_timeout(sock: socket.socket, timeout: Optional[float]) -> Generator[N
         sock.settimeout(orig_timeout)
 
 
-class TcpSocketConnectionManager(BaseConnectionManager[TcpEndpoint, socket.socket]):
+class SocketAlivenessCheckingMixin(Generic[EndpointT]):
+    """
+    Socket aliveness checking mix-in.
+    """
+
+    def check_aliveness(self, endpoint: EndpointT, conn: socket.socket, timeout: Optional[float] = None) -> bool:
+        try:
+            with socket_nonblocking(conn):
+                if conn.recv(1, socket.MSG_PEEK) == b'':
+                    return False
+        except BlockingIOError as exc:
+            if exc.errno != errno.EAGAIN:
+                raise
+        except OSError:
+            return False
+
+        return True
+
+
+class TcpSocketConnectionManager(
+    SocketAlivenessCheckingMixin[TcpEndpoint],
+    BaseConnectionManager[TcpEndpoint, socket.socket],
+):
     """
     TCP socket connection manager.
     """
@@ -76,10 +98,17 @@ def dispose(self, endpoint: TcpEndpoint, conn: socket.socket, timeout: Optional[
 
         conn.close()
 
-    def check_aliveness(self, endpoint: EndpointT, conn: socket.socket, timeout: Optional[float] = None) -> bool:
+
+class SslSocketAlivenessCheckingMixin(Generic[EndpointT]):
+    """
+    SSL socket aliveness checking mix-in.
+    """
+
+    def check_aliveness(self, endpoint: EndpointT, conn: SSLSocket, timeout: Optional[float] = None) -> bool:
         try:
             with socket_nonblocking(conn):
-                if conn.recv(1, socket.MSG_PEEK) == b'':
+                # peek into the plain socket since ssl socket doesn't support flags
+                if socket.socket.recv(conn, 1, socket.MSG_PEEK) == b'':
                     return False
         except BlockingIOError as exc:
             if exc.errno != errno.EAGAIN:
@@ -93,7 +122,10 @@ def check_aliveness(self, endpoint: EndpointT, conn: socket.socket, timeout: Opt
 SslEndpoint = Tuple[Hostname, Port]
 
 
-class SslSocketConnectionManager(BaseConnectionManager[SslEndpoint, SSLSocket]):
+class SslSocketConnectionManager(
+    SslSocketAlivenessCheckingMixin[SslEndpoint],
+    BaseConnectionManager[SslEndpoint, SSLSocket],
+):
     """
     SSL socket connection manager.
     """
@@ -118,17 +150,3 @@ def dispose(self, endpoint: SslEndpoint, conn: SSLSocket, timeout: Optional[floa
             pass
 
         conn.close()
-
-    def check_aliveness(self, endpoint: EndpointT, conn: socket.socket, timeout: Optional[float] = None) -> bool:
-        try:
-            with socket_nonblocking(conn):
-                # peek into the plain socket since ssl socket doesn't support flags
-                if socket.socket.recv(conn, 1, socket.MSG_PEEK) == b'':
-                    return False
-        except BlockingIOError as exc:
-            if exc.errno != errno.EAGAIN:
-                raise
-        except OSError:
-            return False
-
-        return True

generic_connection_pool/contrib/socket_async.py

@@ -7,16 +7,37 @@
 import socket
 from ipaddress import IPv4Address, IPv6Address
 from ssl import SSLContext
-from typing import Tuple, Union
+from typing import Generic, Tuple, Union
 
-from generic_connection_pool.asyncio import BaseConnectionManager
+from generic_connection_pool.asyncio import BaseConnectionManager, EndpointT
 
 IpAddress = Union[IPv4Address, IPv6Address]
 Port = int
 TcpEndpoint = Tuple[IpAddress, Port]
 
 
-class TcpSocketConnectionManager(BaseConnectionManager[TcpEndpoint, socket.socket]):
+class SocketAlivenessCheckingMixin(Generic[EndpointT]):
+    """
+    Nonblocking socket aliveness checking mix-in.
+    """
+
+    async def check_aliveness(self, endpoint: EndpointT, conn: socket.socket) -> bool:
+        try:
+            if conn.recv(1, socket.MSG_PEEK) == b'':
+                return False
+        except BlockingIOError as exc:
+            if exc.errno != errno.EAGAIN:
+                raise
+        except OSError:
+            return False
+
+        return True
+
+
+class TcpSocketConnectionManager(
+    SocketAlivenessCheckingMixin[TcpEndpoint],
+    BaseConnectionManager[TcpEndpoint, socket.socket],
+):
     """
     TCP socket connection manager.
     """
@@ -43,33 +64,40 @@ async def dispose(self, endpoint: TcpEndpoint, conn: socket.socket) -> None:
         conn.shutdown(socket.SHUT_RDWR)
         conn.close()
 
-    async def check_aliveness(self, endpoint: TcpEndpoint, conn: socket.socket) -> bool:
+
+Hostname = str
+TcpStreamEndpoint = Tuple[Hostname, Port]
+Stream = Tuple[asyncio.StreamReader, asyncio.StreamWriter]
+
+
+class StreamAlivenessCheckingMixin(Generic[EndpointT]):
+    """
+    Asynchronous stream aliveness checking mix-in.
+    """
+
+    async def check_aliveness(self, endpoint: EndpointT, conn: Stream) -> bool:
+        reader, writer = conn
+
         try:
-            if conn.recv(1, socket.MSG_PEEK) == b'':
-                return False
-        except BlockingIOError as exc:
-            if exc.errno != errno.EAGAIN:
-                raise
+            await reader.read(0)
         except OSError:
             return False
 
-        return True
-
-
-Hostname = str
-TcpStreamEndpoint = Tuple[Hostname, Port]
-TcpStream = Tuple[asyncio.StreamReader, asyncio.StreamWriter]
+        return not writer.is_closing() and not reader.at_eof()
 
 
-class TcpStreamConnectionManager(BaseConnectionManager[TcpStreamEndpoint, TcpStream]):
+class TcpStreamConnectionManager(
+    StreamAlivenessCheckingMixin[TcpStreamEndpoint],
+    BaseConnectionManager[TcpStreamEndpoint, Stream],
+):
     """
     TCP stream connection manager.
     """
 
     def __init__(self, ssl: Union[None, bool, SSLContext] = None):
         self._ssl = ssl
 
-    async def create(self, endpoint: TcpStreamEndpoint) -> TcpStream:
+    async def create(self, endpoint: TcpStreamEndpoint) -> Stream:
         hostname, port = endpoint
         server_hostname = hostname if self._ssl is not None else None
 
@@ -82,20 +110,10 @@ async def create(self, endpoint: TcpStreamEndpoint) -> TcpStream:
 
         return reader, writer
 
-    async def dispose(self, endpoint: TcpStreamEndpoint, conn: TcpStream) -> None:
+    async def dispose(self, endpoint: TcpStreamEndpoint, conn: Stream) -> None:
         reader, writer = conn
         if writer.can_write_eof():
             writer.write_eof()
 
         writer.close()
         await writer.wait_closed()
-
-    async def check_aliveness(self, endpoint: TcpStreamEndpoint, conn: TcpStream) -> bool:
-        reader, writer = conn
-
-        try:
-            await reader.read(0)
-        except OSError:
-            return False
-
-        return not writer.is_closing() and not reader.at_eof()

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "generic-connection-pool"
-version = "0.4.0"
+version = "0.4.1"
 description = "generic connection pool"
 authors = ["Dmitry Pershin <[email protected]>"]
 license = "Unlicense"