Merge pull request #17 from dapper91/hooks-example

- hooks usage example added.

Author: dapper91

docs/source/pages/api.rst

@@ -35,3 +35,6 @@ _______
 
 .. automodule:: generic_connection_pool.contrib.unix
     :members:
+
+.. automodule:: generic_connection_pool.contrib.unix_async
+    :members:

docs/source/pages/quickstart.rst

@@ -125,6 +125,14 @@ The following example illustrate how to implement a custom connection manager fo
   :language: python
 
 
+Connection manager allows to define methods to be called on connection acquire, release or when
+a connection determined to be dead. That helps to log pool actions or collect metrics.
+The following examples illustrate how to collect pool metrics and export them to prometheus.
+
+.. literalinclude:: ../../../examples/manager_hooks.py
+  :language: python
+
+
 Examples
 ________
 

examples/manager_hooks.py

@@ -0,0 +1,46 @@
+import ssl
+import time
+from ssl import SSLSocket
+from typing import Any, Dict, Tuple
+
+import prometheus_client as prom
+
+from generic_connection_pool.contrib.socket import SslSocketConnectionManager
+from generic_connection_pool.threading import ConnectionPool
+
+Hostname = str
+Port = int
+Endpoint = Tuple[Hostname, Port]
+
+acquire_latency_hist = prom.Histogram('acquire_latency', 'Connections acquire latency', labelnames=['hostname'])
+acquire_total = prom.Counter('acquire_total', 'Connections acquire count', labelnames=['hostname'])
+dead_conn_total = prom.Counter('dead_conn_total', 'Dead connections count', labelnames=['hostname'])
+
+
+class ObservableConnectionManager(SslSocketConnectionManager):
+
+    def __init__(self, *args: Any, **kwargs: Any):
+        super().__init__(*args, **kwargs)
+        self._acquires: Dict[SSLSocket, float] = {}
+
+    def on_acquire(self, endpoint: Endpoint, conn: SSLSocket) -> None:
+        hostname, port = endpoint
+
+        acquire_total.labels(hostname).inc()
+        self._acquires[conn] = time.time()
+
+    def on_release(self, endpoint: Endpoint, conn: SSLSocket) -> None:
+        hostname, port = endpoint
+
+        acquired_at = self._acquires.pop(conn)
+        acquire_latency_hist.labels(hostname).observe(time.time() - acquired_at)
+
+    def on_connection_dead(self, endpoint: Endpoint, conn: SSLSocket) -> None:
+        hostname, port = endpoint
+
+        dead_conn_total.labels(hostname).inc()
+
+
+http_pool = ConnectionPool[Endpoint, SSLSocket](
+    ObservableConnectionManager(ssl.create_default_context()),
+)

examples/ssl_pool.py

@@ -1,4 +1,3 @@
-import socket
 import ssl
 import urllib.parse
 from http.client import HTTPResponse
@@ -10,7 +9,7 @@
 Hostname = str
 Port = int
 Endpoint = Tuple[Hostname, Port]
-Connection = socket.socket
+Connection = ssl.SSLSocket
 
 
 http_pool = ConnectionPool[Endpoint, Connection](

examples/tcp_pool.py

@@ -1,12 +1,13 @@
 import socket
-from ipaddress import IPv4Address
-from typing import Tuple
+from ipaddress import IPv4Address, IPv6Address
+from typing import Tuple, Union
 
 from generic_connection_pool.contrib.socket import TcpSocketConnectionManager
 from generic_connection_pool.threading import ConnectionPool
 
 Port = int
-Endpoint = Tuple[IPv4Address, Port]
+IpAddress = Union[IPv4Address, IPv6Address]
+Endpoint = Tuple[IpAddress, Port]
 Connection = socket.socket
 
 
@@ -21,7 +22,7 @@
 )
 
 
-def command(addr: IPv4Address, port: int, cmd: str) -> None:
+def command(addr: IpAddress, port: int, cmd: str) -> None:
     with redis_pool.connection(endpoint=(addr, port), timeout=5.0) as sock:
         sock.sendall(cmd.encode() + b'\n')
         response = sock.recv(1024)