file_buffer

Implements FileBuffer, a WAL-based buffer using local files.

class unistream.buffers.file_buffer.FileBuffer(record_class: type[AbcRecord], path_wal: Path, max_records: int = 1000, max_bytes: int = 1000000)

Use local log file as write-ahead-log (WAL) to persist the buffer.

Parameters:

• record_class – the record class.

• path_wal – the path of the WAL file. it must have a filename and a file extension, for example: my_buffer.log.

• max_records – The maximum number of records that can be stored in the buffer.

• max_bytes – The maximum total size of records (in bytes) that can be stored in the buffer.

• n_records – This variable tracks the number of records in the memory queue.

• n_bytes – This variable tracks the number of bytes in the memory queue.

• memory_queue

• memory_serialization_queue

• storage_queue – This queue tracks the older WAL files that have not been emitted.

Note

For factory method parameter definition, see factory method at FileBuffer.new().

classmethod new(record_class: type[AbcRecord], path_wal: Path, max_records: int = 1000, max_bytes: int = 1000000)

Create a new instance of FileBuffer.

Parameters:

• record_class – the record class.

• path_wal – the path of the WAL file. it must have a filename and a file extension, for example: my_buffer.log.

• max_records – The maximum number of records that can be stored in the buffer.

• max_bytes – The maximum total size of records (in bytes) that can be stored in the buffer.

clear_memory_queue()

Clear the in-memory queue. Including the queue for original records and the queue for serialized records. Also reset the records and bytes counter.

clear_wal()

Clear all WAL file. Including the current one and old files.

put(record: AbcRecord)

Put one record to the buffer.

It automatically add the record to the memory queue, write it to the WAL file, and move the WAL file to the storage queue if the buffer is full, which indicates that the buffer is ready to emit records.

todo: when putting lots of records, avoid open and close file every time,

maybe create a put_many method. However, I did a test, it seems that open and close file 1000 times is not a big deal comparing to network IO.

should_i_emit() → bool

Since we immediately move the WAL file to the storage queue when it is full, if the storage queue is not empty, it means we should emit records.

emit() → list[AbcRecord]

Emit the records due to the buffer is full. Similar to _emit(), it leverages the cache to reduce IO.

commit()

When the emitted records are successfully processed, we can remove it from the storage queue. If the emitted records are from the memory queue, then we can clear the memory queue and delete the current WAL file. We also clear the emitted records cache.