micronetcode.codec

codec.py - Interface with 3 methods: recv (socket)->bytes, decode(bytes)->list[Message] and encode(str)->bytes

 1"""codec.py - Interface with 3 methods: recv (socket)->bytes, decode(bytes)->list[Message] and encode(str)->bytes"""
 2from abc import ABC, abstractmethod
 3from typing import Any
 4import socket
 5
 6class Codec(ABC):
 7    """Interface with 3 methods: recv (socket)->bytes, decode(bytes)->list[Message] and encode(str)->bytes"""
 8    @abstractmethod
 9    def recv(self, sock: socket.socket) -> bytes | None:
10        """
11        Receive raw bytes from the wire (socket). Different protocols may have different ways to recv (len+data).
12        Returns None if the socket closed the connection. This is used to re-use the client thread for new connections.
13        """
14
15    @abstractmethod
16    def decode(self, data: bytes) -> list[Any]:
17        """Converts raw bytes data from recv(sock) into a list of payloads (may contain >1; codec specific)"""
18
19    @abstractmethod
20    def encode(self, resp: Any) -> bytes:
21        """Encodes the response of the main app back into bytes to be sent via the socket used at recv()"""
class Codec(abc.ABC): 
 7class Codec(ABC):
 8    """Interface with 3 methods: recv (socket)->bytes, decode(bytes)->list[Message] and encode(str)->bytes"""
 9    @abstractmethod
10    def recv(self, sock: socket.socket) -> bytes | None:
11        """
12        Receive raw bytes from the wire (socket). Different protocols may have different ways to recv (len+data).
13        Returns None if the socket closed the connection. This is used to re-use the client thread for new connections.
14        """
15
16    @abstractmethod
17    def decode(self, data: bytes) -> list[Any]:
18        """Converts raw bytes data from recv(sock) into a list of payloads (may contain >1; codec specific)"""
19
20    @abstractmethod
21    def encode(self, resp: Any) -> bytes:
22        """Encodes the response of the main app back into bytes to be sent via the socket used at recv()"""

Interface with 3 methods: recv (socket)->bytes, decode(bytes)->list[Message] and encode(str)->bytes

@abstractmethod
def recv(self, sock: socket.socket) -> bytes | None: 
 9    @abstractmethod
10    def recv(self, sock: socket.socket) -> bytes | None:
11        """
12        Receive raw bytes from the wire (socket). Different protocols may have different ways to recv (len+data).
13        Returns None if the socket closed the connection. This is used to re-use the client thread for new connections.
14        """

Receive raw bytes from the wire (socket). Different protocols may have different ways to recv (len+data). Returns None if the socket closed the connection. This is used to re-use the client thread for new connections.

@abstractmethod
def decode(self, data: bytes) -> list[typing.Any]: 
16    @abstractmethod
17    def decode(self, data: bytes) -> list[Any]:
18        """Converts raw bytes data from recv(sock) into a list of payloads (may contain >1; codec specific)"""

Converts raw bytes data from recv(sock) into a list of payloads (may contain >1; codec specific)

@abstractmethod
def encode(self, resp: Any) -> bytes: 
20    @abstractmethod
21    def encode(self, resp: Any) -> bytes:
22        """Encodes the response of the main app back into bytes to be sent via the socket used at recv()"""

Encodes the response of the main app back into bytes to be sent via the socket used at recv()