mirror of
https://github.com/varun-r-mallya/py-libp2p.git
synced 2025-12-31 20:36:24 +00:00
85 lines
1.6 KiB
Python
85 lines
1.6 KiB
Python
from abc import (
|
|
ABC,
|
|
abstractmethod,
|
|
)
|
|
from typing import (
|
|
Optional,
|
|
)
|
|
|
|
|
|
class Closer(ABC):
|
|
@abstractmethod
|
|
async def close(self) -> None:
|
|
...
|
|
|
|
|
|
class Reader(ABC):
|
|
@abstractmethod
|
|
async def read(self, n: int = None) -> bytes:
|
|
...
|
|
|
|
|
|
class Writer(ABC):
|
|
@abstractmethod
|
|
async def write(self, data: bytes) -> None:
|
|
...
|
|
|
|
|
|
class WriteCloser(Writer, Closer):
|
|
pass
|
|
|
|
|
|
class ReadCloser(Reader, Closer):
|
|
pass
|
|
|
|
|
|
class ReadWriter(Reader, Writer):
|
|
pass
|
|
|
|
|
|
class ReadWriteCloser(Reader, Writer, Closer):
|
|
@abstractmethod
|
|
def get_remote_address(self) -> Optional[tuple[str, int]]:
|
|
"""
|
|
Return the remote address of the connected peer.
|
|
|
|
:return: A tuple of (host, port) or None if not available
|
|
"""
|
|
...
|
|
|
|
|
|
class MsgReader(ABC):
|
|
@abstractmethod
|
|
async def read_msg(self) -> bytes:
|
|
...
|
|
|
|
|
|
class MsgWriter(ABC):
|
|
@abstractmethod
|
|
async def write_msg(self, msg: bytes) -> None:
|
|
...
|
|
|
|
|
|
class MsgReadWriteCloser(MsgReader, MsgWriter, Closer):
|
|
pass
|
|
|
|
|
|
class Encrypter(ABC):
|
|
@abstractmethod
|
|
def encrypt(self, data: bytes) -> bytes:
|
|
...
|
|
|
|
@abstractmethod
|
|
def decrypt(self, data: bytes) -> bytes:
|
|
...
|
|
|
|
|
|
class EncryptedMsgReadWriter(MsgReadWriteCloser, Encrypter):
|
|
"""Read/write message with encryption/decryption."""
|
|
|
|
def get_remote_address(self) -> Optional[tuple[str, int]]:
|
|
"""Get remote address if supported by the underlying connection."""
|
|
if hasattr(self, "conn") and hasattr(self.conn, "get_remote_address"):
|
|
return self.conn.get_remote_address()
|
|
return None
|