async_example_program.tools.async_offline_buffer.AsyncOfflineBuffer

class documentation

class AsyncOfflineBuffer: (source)

Constructor: AsyncOfflineBuffer(offline_path, name)

View In Hierarchy

This class handles an offline message buffer.

Each message that should be stored needs to be a dict before it can be appended to the offline buffer file. The buffer will grow indefinitely until it is emptied.

The buffer can be retrieved as a list of dict messages, and the buffer is regarded as empty after that.

Buffer extension will always be .json, regardless of the initial buffer name.

Method __init__ The class constructor.
Method __len__ Return number of buffered messages.
Async Method append Append messages to an existing buffer with 'utf-8' encoding.
Async Method retrieve Return the content of the current buffer and regard the buffer empty.
Instance Variable buffer_empty Keeps track of buffer empty status.
Instance Variable count Keep track on number of stored messages.
Instance Variable filename Name of offline file.
Property is_empty Return buffer empty status.
def __init__(self, offline_path: str, name: Optional[str] = None): (source)

The class constructor.

Parameters
offline_path:strOffline path used for storing failed messages.
name:Optional[str]Filename used for storing failed messages.
def __len__(self) -> int: (source)

Return number of buffered messages.

Returns
intNumber of buffered messages.
async def append(self, msg: dict): (source)

Append messages to an existing buffer with 'utf-8' encoding.

If the buffer is regarded as empty, create a new one.

Parameters
msg:dictMessage to buffer.
async def retrieve(self) -> list: (source)

Return the content of the current buffer and regard the buffer empty.

Returns
listBuffer messages.
buffer_empty: bool = (source)

Keeps track of buffer empty status.

count: int = (source)

Keep track on number of stored messages.

filename: Path = (source)

Name of offline file.

@property
is_empty: bool = (source)

Return buffer empty status.