I want a FIFO queue for bytes conforming to python IOBase interface. I cannot find such a library available, nor could I produce one via trivial combinations of existiing libraries:

from io import RawIOBase
from typing import Union

class FIFOBytesIO(RawIOBase):

    def nextPowerOf2(n):
        p = 1
        if n and not (n & (n - 1)):
            return n
        while p < n:
            p <<= 1
        return p

    def __init__(self, length=1024):
        self._buf = bytearray(self.nextPowerOf2(length))
        self._start = 0
        self._end = 0

    def capacity(self):
        return len(self._buf)

    def idx_ptr(self, i):
        return (self._start + i) & (self.capacity - 1)

    def start_ptr(self):
        return self._start & (self.capacity - 1)

    def end_ptr(self):
        return self._end & (self.capacity - 1)

    def __len__(self):
        return self._end - self._start

    def readinto(self, b: bytearray, size=-1, inc_ptr=True) -> int:
        if not (0 <= size <= len(self)):
            size = len(self)
        end_ptr = self.idx_ptr(size)
        if end_ptr > self.start_ptr:
            b[:size] = self._buf[self.start_ptr:end_ptr]
        elif size > 0:
            chunk1_size = self.capacity - self.start_ptr
            chunk2_size = end_ptr
            b[:chunk1_size] = self._buf[self.start_ptr:]
            b[chunk1_size:(chunk1_size + chunk2_size)] = self._buf[:end_ptr]

        if inc_ptr:
            self._start += size
        return size

    def write(self, b: Union[bytes, bytearray]) -> int:
        new_length = len(b) + len(self)
        if new_length <= self.capacity:
            new_end_ptr = self.idx_ptr(new_length)
            if new_end_ptr > self.end_ptr:
                self._buf[self.end_ptr:new_end_ptr] = b
            elif len(b) > 0:
                chunk1_size = self.capacity - self.end_ptr
                chunk2_size = new_end_ptr
                self._buf[self.end_ptr:] = b[:chunk1_size]
                self._buf[:new_end_ptr] = b[chunk1_size:(chunk1_size + chunk2_size)]

            self._end += len(b)
            new_buf_size = self.nextPowerOf2(new_length)
            new_buf = bytearray(new_buf_size)
            self.readinto(new_buf, inc_ptr=False)
            new_buf[len(self):new_length] = b
            self._buf = new_buf
            self._start = 0
            self._end = new_length

        return len(b)

    def clear(self, length=1024):

    def read(self, size=-1, inc_ptr=True) -> bytes:
        if not size or size < 0:
            size = len(self)

        b = bytearray(size)
        self.readinto(b, size, inc_ptr)
        return bytes(b)

    def getvalue(self) -> bytes:
        return self.read(inc_ptr=False)

    def fileno(self) -> int:
        raise OSError()

    def flush(self) -> None:

    def isatty(self) -> bool:
        return False

    def readable(self) -> bool:
        return True

    def seekable(self) -> bool:
        return False

    def writable(self) -> bool:
        return True

    def closed(self) -> bool:
        return False

I am not sure if this conforms to python API. Specifically, the readline method of this class blocks if there isn't a newline in the buffer. Also, what is the desire behavior of seek, tell, and truncate for a FIFO stream?

The code I provided above works for basic read and write, I am just not sure what extra behaviors I need to implement to make it work better.


Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.