Custom Protocol

custom_protocol.py
 1import asyncio
 2import typing
 3
 4from zcached.asyncio import AsyncZCached
 5
 6
 7class MyProtocol(asyncio.StreamReaderProtocol):
 8   # Our protocol must inherit from the StreamReaderProtocol.
 9
10    def connection_made(self, transport: asyncio.BaseTransport) -> None:
11        """Called when we made a connection to the server."""
12        super().connection_made(transport)  # Calling the original method.
13
14        # Our custom logic:
15        host, port = transport.get_extra_info("peername")
16        print(f"Connected with: {host}:{port}")
17
18    def connection_lost(self, exc: typing.Optional[Exception]) -> None:
19        """Called when we lost the connection."""
20        super().connection_lost(exc)  # Calling the original method.
21
22        # Our custom logic:
23        print("Connection lost!")
24        if exc is not None:
25            print(f"Raised exception: {exc}")
26
27
28async def main():
29    client: AsyncZCached = AsyncZCached(
30        host="127.0.0.1", port=7556, protocol_type=MyProtocol
31    )
32    await client.run()
33
34
35if __name__ == "__main__":
36    asyncio.run(main())