0
\$\begingroup\$

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):

    @staticmethod
    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

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

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

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

    @property
    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)
        else:
            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):
        self.__init__(length)

    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:
        pass

    def isatty(self) -> bool:
        return False

    def readable(self) -> bool:
        return True

    def seekable(self) -> bool:
        return False

    def writable(self) -> bool:
        return True

    @property
    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.

\$\endgroup\$

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.