iofree package

iofree is an easy-to-use and powerful library to help you implement network protocols and binary parsers.

class iofree.LinkedNode(parser: iofree.Parser, next_: Optional[LinkedNode])[source]

Bases: object

next
parser
class iofree.Parser(gen: Generator)[source]

Bases: object

finished() → bool[source]
get_result() → Any[source]

raises NoResult exception if no result has been set

has_more_data() → bool[source]

indicate whether input has some bytes left

property has_result
parse(data: bytes, *, strict: bool = True) → Any[source]

parse bytes

read_output_bytes() → bytes[source]
readall() → bytes[source]

retrieve data from input back

respond(*, data: bytes = b'', close: bool = False, exc: Optional[Exception] = None, result: Any = <object object>) → None[source]

produce some event data to interact with a stream: data: bytes to send to the peer close: whether the socket should be closed exc: raise an exception to break the loop result: result to return

run(sock: _socket.socket) → Any[source]

reference implementation of how to deal with socket

send(data: bytes = b'') → None[source]

send data for parsing

send_event(event: Any) → None[source]
set_result(result) → None[source]
class iofree.ParserChain(*parsers: iofree.Parser)[source]

Bases: object

send(data: bytes) → None[source]
class iofree.State[source]

Bases: enum.IntEnum

An enumeration.

class iofree.Traps[source]

Bases: enum.IntEnum

An enumeration.

iofree.get_parser() → Generator[tuple, iofree.Parser, iofree.Parser][source]

get current parser object

iofree.parser(generator_func: Callable) → Callable[source]

decorator function to wrap a generator

iofree.peek(nbytes: int = 1, *, from_=None) → Generator[tuple, bytes, bytes][source]

peek many bytes without taking them away from buffer

iofree.read(nbytes: int = 0, *, from_=None) → Generator[tuple, bytes, bytes][source]

if nbytes = 0, read as many as possible, empty bytes is valid; if nbytes > 0, read exactly nbytes

iofree.read_int(nbytes: int, byteorder: str = 'big', *, signed: bool = False, from_=None) → Generator[tuple, int, int][source]

read some bytes as integer

iofree.read_more(nbytes: int = 1, *, from_=None) → Generator[tuple, bytes, bytes][source]

read at least nbytes

iofree.read_raw_struct(struct_obj: Struct, *, from_=None) → Generator[tuple, tuple, tuple][source]

read raw struct formatted data

iofree.read_struct(fmt: str, *, from_=None) → Generator[tuple, tuple, tuple][source]

read specific formatted data

iofree.read_until(data: bytes, *, return_tail: bool = True, from_=None) → Generator[tuple, bytes, bytes][source]

read until some bytes appear

iofree.wait() → Generator[tuple, bytes, Optional[object]][source]

wait for next send event

iofree.wait_event() → Generator[tuple, Any, Any][source]

wait for an event