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

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[micronetcode.message.Message]: 
18    @abstractmethod
19    def decode(self, data: bytes) -> list[Message]:
20        """Converts raw bytes data fro recv(sock) into a list of messages for the main app"""

Converts raw bytes data fro recv(sock) into a list of messages for the main app

@abstractmethod
def encode(self, resp: Any) -> bytes: 
22    @abstractmethod
23    def encode(self, resp: Any) -> bytes:
24        """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()