# -*- coding: utf-8 -*-
"""Seekable I/O Object
=========================
.. module:: pcapkit.corekit.io
:mod:`pcapkit.corekit.io` contains seekable I/O object
:class:`~pcapkit.corekit.io.SeekableReader`, which is a customised
implementation to :class:`io.BufferedReader`.
"""
import io
import tempfile
from typing import TYPE_CHECKING, cast
from pcapkit.utilities.exceptions import (SeekError, TruncateError, UnsupportedCall,
UnsupportedOperation, stacklevel)
from pcapkit.utilities.warnings import SeekWarning, warn
if TYPE_CHECKING:
from io import BytesIO, RawIOBase
from typing import IO, Iterable, Optional
from typing_extensions import Buffer
__all__ = ['SeekableReader']
[docs]
class SeekableReader(io.BufferedReader):
"""Seekable buffered reader.
A buffered binary stream providing higher-level access to a readable, non seekable
:class:`~io.RawIOBase` raw binary stream. It inherits :class:`~io.BufferedIOBase`.
When reading data from this object, a larger amount of data may be requested from the
underlying raw stream, and kept in an internal buffer. The buffered data can then be returned
directly on subsequent reads.
The constructor creates a :class:`~io.BufferedReader` for the given readable ``raw`` stream and
``buffer_size``. If ``buffer_size`` is omitted, :data:`~io.DEFAULT_BUFFER_SIZE` is used.
Args:
raw: Underlying raw stream.
buffer_size: Buffer size.
buffer_save: Whether to save buffer to file.
buffer_path: Path to save buffer.
stream_closing: Whether the stream should be closed upon exiting.
"""
if TYPE_CHECKING:
#: Whether the stream should be closed upon exiting.
_closing: 'bool'
#: Whether the stream is closed.
_closed: 'bool'
#: Underlying raw stream.
_stream: 'IO[bytes]'
#: Current position of the stream.
_tell: 'int'
#: Buffer.
_buffer: 'BytesIO'
#: Buffer view.
_buffer_view: 'memoryview'
#: Buffer size.
_buffer_size: 'int'
#: Buffer start position.
_buffer_set: 'int'
#: Buffer current position.
_buffer_cur: 'int'
#: Path to save buffer.
_buffer_path: 'str'
#: File to save buffer.
_buffer_file: 'IO[bytes] | None'
@property
def closed(self) -> 'bool':
""":data:`True` if the stream is closed."""
return self._closed
@property
def raw(self) -> 'RawIOBase':
"""The underlying raw stream (a :class:`~io.RawIOBase` instance) that
:class:`~io.BufferedIOBase` deals with. This is not part of the :class:`~io.BufferedIOBase`
API and may not exist on some implementations."""
return cast('RawIOBase', self._stream)
@raw.setter
def raw(self, raw: 'RawIOBase', /) -> 'None':
raise UnsupportedCall("can't set attribute")
def __init__(self, raw: 'IO[bytes]', buffer_size: 'int' = io.DEFAULT_BUFFER_SIZE,
buffer_save: 'bool' = False, buffer_path: 'Optional[str]' = None, *,
stream_closing: 'bool' = True) -> 'None':
super().__init__(cast('RawIOBase', raw), buffer_size)
self._closed = False
self._closing = stream_closing
self._stream = raw
self._buffer = io.BytesIO(bytearray(buffer_size))
self._buffer_view = self._buffer.getbuffer()
self._buffer_size = buffer_size
if buffer_save:
if buffer_path is None:
self._buffer_file = tempfile.NamedTemporaryFile('wb', buffering=0)
self._buffer_path = self._buffer_file.name
else:
self._buffer_file = open(buffer_path, 'wb', buffering=0)
self._buffer_path = buffer_path
else:
self._buffer_file = None
self._buffer_path = ''
self._tell = self._buffer_set = self._buffer_cur = 0
def _write_buffer(self, buf: 'bytes', /) -> 'None':
if self._buffer_file is not None:
self._buffer_file.write(buf)
self._buffer_file.flush()
buf_len = len(buf)
old_ptr = self._buffer_cur
self._buffer_cur += buf_len
if self._buffer_cur > self._buffer_size:
if buf_len >= self._buffer_size:
# NOTE: the last ``_buffer_size`` octets, counted from the front rather
# than as ``buf[-self._buffer_size:]``, which for a buffer of no size at
# all is ``buf[-0:]`` -- the whole of ``buf``, not none of it.
self._buffer_view[:] = buf[buf_len - self._buffer_size:]
else:
self._buffer_view[:-buf_len] = self._buffer_view[old_ptr - (self._buffer_size - buf_len):old_ptr]
self._buffer_view[-buf_len:] = buf
self._buffer_set += self._buffer_cur - self._buffer_size
self._buffer_cur = self._buffer_size
# move the pointer to the end of the original contents
self._buffer.seek(-buf_len, io.SEEK_END)
else:
self._buffer_view[old_ptr:self._buffer_cur] = buf
def _seek_buffer(self) -> 'int':
"""Point the buffer at the current stream position, and say how much it can serve.
The buffer is a sliding window whose octet 0 sits at absolute ``_buffer_set`` and
whose content occupies ``[0:_buffer_cur]``, so ``_buffer_set + _buffer_cur`` is how
far the underlying stream has been consumed. Two things follow, and every buffered
read path needs both of them:
* The buffer's own cursor is a **derived** quantity, ``_tell - _buffer_set``, rather
than a fourth piece of state to be maintained. Nothing was maintaining it: a read
served from the stream writes through :attr:`_buffer_view` and leaves the cursor
untouched, :meth:`_write_buffer` rewinds it to the start of the appended octets
when the window slides, and :meth:`peek` used to advance it while leaving ``_tell``
alone. Deriving it here, at each point of use, is what makes the drift
unrepresentable instead of merely repaired afterwards by the next :meth:`seek`.
* What is available is the run from the position to the end of the content,
``_buffer_set + _buffer_cur - _tell`` -- not ``_buffer_cur``, which is measured
from the window's base and so counts octets that lie *behind* the position, and
not that count less one, which is short of both. Over-asking does not fail: the
buffer is allocated full of NUL padding, so it hands that back as though it were
data, with a plausible length that then suppresses the top-up from the stream.
Returns:
The number of octets the buffer can answer for from the current position.
Raises:
SeekError: If the position lies before the window, whose octets are then gone
for good -- the stream cannot be rewound to re-supply them. :meth:`seek`
refuses that position for the same reason; it is reachable here only through
a :meth:`_truncate_buffer` that moved the window's base past a position already
set.
"""
buf_off = self._tell - self._buffer_set
if buf_off < 0:
raise SeekError(f'cannot read before the beginning of the buffer: '
f'{self._tell} < {self._buffer_set}')
self._buffer.seek(buf_off, io.SEEK_SET)
return self._buffer_cur - buf_off
[docs]
def close(self) -> 'None':
"""Flush and close this stream. This method has no effect if the file is already closed.
Once the file is closed, any operation on the file (e.g. reading or writing) will raise
a :exc:`ValueError`.
As a convenience, it is allowed to call this method more than once; only the first call,
however, will have an effect.
"""
if self.closed:
return
self.flush()
if self._closing:
self._stream.close()
if self._buffer_file is not None:
self._buffer_file.close()
self._buffer_view.release()
self._buffer.close()
self._closed = True
[docs]
def fileno(self) -> 'int':
"""Return the underlying file descriptor (an integer) of the stream if it exists.
An :exc:`OSError` is raised if the IO object does not use a file descriptor."""
return self._stream.fileno()
[docs]
def flush(self) -> 'None':
"""Flush the write buffers of the stream if applicable. This does nothing for
read-only and non-blocking streams."""
if self._buffer_file is not None:
self._buffer_file.flush()
self._stream.flush()
[docs]
def isatty(self) -> 'bool':
"""Return :data:`True` if the stream is interactive (i.e., connected to a
terminal/tty device)."""
return self._stream.isatty()
[docs]
def readable(self) -> 'bool':
"""Return :data:`True` if the stream can be read from. If :data:`False`,
:meth:`read` will raise :exc:`OSError`."""
return self._stream.readable()
[docs]
def readline(self, size: 'int | None' = -1, /) -> 'bytes':
r"""Read and return one line from the stream. If ``size`` is specified, at most
``size`` bytes will be read.
The line terminator is always ``b'\n'`` for binary files; for text files, the
``newline`` argument to :func:`open` can be used to select the line
terminator(s) recognized.
"""
if size is None:
size = -1
if self._tell >= self._buffer_set + self._buffer_cur:
buf = self._stream.readline(size)
self._write_buffer(buf)
else:
if self._buffer_file is not None and self._tell < self._buffer_set:
with open(self._buffer_path, 'rb') as temp_file:
temp_file.seek(self._tell, io.SEEK_SET)
buf = temp_file.readline(size)
elif not size:
# NOTE: a request for no octets is answered without consulting the window,
# which has nothing to say about it. Asking anyway refuses a zero-length
# read from a position :meth:`_truncate_buffer` has left behind the window -- a
# refusal over data that was never wanted, and not what this used to do.
#
# Truthiness rather than ``size == 0`` on purpose. It is the same test for
# every value this method can be reached with -- ``size`` is an ``int`` by
# the time control arrives, and ``False`` is ``0`` -- but :meth:`peek` does
# not normalise a ``None`` its signature does not permit, and there the two
# spellings diverge: ``size == 0`` would send ``None`` on to the window and
# report a type error as a position error. Keeping it falsy keeps that call
# failing as the :exc:`TypeError` it always was.
buf = b''
else:
buf_rem = self._seek_buffer()
buf = self._buffer.readline(buf_rem if size < 0 else min(size, buf_rem))
# NOTE: an unbounded ``readline`` has to keep going until the line ends, and
# capping it at the buffer's content -- which the line need not end inside --
# is what makes the continuation necessary rather than optional here.
size_rem = -1
if not buf.endswith(b'\n') and (size < 0 or (size_rem := size - len(buf)) > 0):
buf_tmp = self._stream.readline(size_rem)
self._write_buffer(buf_tmp)
buf += buf_tmp
self._tell += len(buf)
return buf
[docs]
def readlines(self, hint: 'int' = -1, /) -> 'list[bytes]':
"""Read and return a list of lines from the stream. ``hint`` can be specified to control
the number of lines read: no more lines will be read if the total size (in
bytes/characters) of all lines so far exceeds ``hint``.
``hint`` values of ``0`` or less, as well as :obj:`None`, are treated as no hint.
Note that it's already possible to iterate on file objects using ``for line in file: ...``
without calling :meth:`file.readlines() <readlines>`.
"""
if hint is None or hint <= 0:
lines = [] # type: list[bytes]
while True:
line = self.readline()
if not line:
break
lines.append(line)
return lines
size = 0
lines = []
while size < hint:
line = self.readline(hint - size)
if not line:
break
lines.append(line)
size += len(line)
return lines
[docs]
def seek(self, offset: 'int', whence: 'int' = io.SEEK_SET, /) -> 'int':
"""Change the stream position to the given byte ``offset``. ``offset`` is interpreted
relative to the position indicated by ``whence``. The default value for ``whence`` is
:data:`~io.SEEK_SET`. Values for ``whence`` are:
* :data:`~io.SEEK_SET` or ``0`` - start of the stream (the default); ``offset`` should
be zero or positive
* :data:`~io.SEEK_CUR` or ``1`` - current stream position; ``offset`` may be negative
* :data:`~io.SEEK_END` or ``2`` - end of the stream; ``offset`` is usually negative
Return the new absolute position.
Note:
The target is computed, then validated, and only then written to ``_tell``. A
branch that assigned the position before deciding whether to accept it left a
refused seek having moved it anyway, with the resync below -- the one thing that
puts the buffer's cursor back in step -- skipped on the way out. A caller that
catches the error and reasonably takes the position to be unchanged then read
from the rejected offset instead, silently and without a second error.
The negative check applies to every ``whence``, rather than only
:data:`~io.SEEK_SET` looking at its offset. An absolute position below zero
cannot exist under any of them, and it is not the same failure as a position
that has merely slid out of the window -- which is ordinary, and recoverable
with ``buffer_save=True``. Reporting the first as ``negative seek value`` keeps
the two distinguishable, and being a property of the position rather than of the
window it holds with a saved buffer as well, where the old refusal did not
apply at all and the seek returned a negative position as if it had worked.
"""
# NOTE: we mark the end of buffer content to the end of buffer
# so that it may trigger the IO to read more data to fill in
# the content.
buf_end = self._buffer_set + self._buffer_size
#buf_end = self._buffer_set + self._buffer_cur
if whence == io.SEEK_SET:
target = offset
elif whence == io.SEEK_CUR:
target = self._tell + offset
elif whence == io.SEEK_END:
target = buf_end + offset
else:
raise SeekError(f'invalid whence ({whence}, should be {io.SEEK_SET}, {io.SEEK_CUR} or {io.SEEK_END})')
if target < 0:
raise SeekError(f'negative seek value {target}')
# NOTE: both refusals read the target rather than ``_tell``, which is what lets them
# run before the position is committed. This one is the window's, so a saved buffer
# -- which can still supply the octets from file -- is exempt from it.
if target < self._buffer_set and self._buffer_file is None:
raise SeekError(f'cannot seek before the beginning of the buffer: {target} < {self._buffer_set}')
self._tell = target
if self._tell >= self._buffer_set:
if self._tell > buf_end:
warn(f'seek beyond the end of the buffer: {self._tell} > {buf_end}',
SeekWarning, stacklevel=stacklevel())
if self._tell > (tmp_end := self._buffer_set + self._buffer_cur):
# NOTE: if we do need to seek beyond the existing contents,
# then we'll do a quick read to make up the contents; the
# size of the read is set to be 1/4 size of the buffer or
# the size of the content to be read, whichever is larger.
# However, the length to fill must not be larger than the
# buffer size itself.
tmp_len = min(max(self._tell - tmp_end, self._buffer_size // 4), self._buffer_size)
self._tell = tmp_end
tmp_buf = self.read1(tmp_len)
self._tell = tmp_end + len(tmp_buf)
self._buffer.seek(self._tell - self._buffer_set, io.SEEK_SET)
else:
# NOTE: only a saved buffer reaches here -- the refusal above has already
# turned away a position before the window when there is no file to serve it
# from, which is what the refusal used to do at this point instead, after
# ``_tell`` had been moved.
self._buffer.seek(0, io.SEEK_SET)
return self._tell
[docs]
def seekable(self) -> 'bool':
"""Return :data:`True` if the stream supports random access. If :data:`False`,
:meth:`seek`, :meth:`tell` and :meth:`truncate` will raise :exc:`OSError`."""
return True
[docs]
def tell(self) -> 'int':
"""Return the current stream position."""
return self._tell
[docs]
def truncate(self, size: 'int | None' = None, /) -> 'int':
"""Resize the stream to the given ``size`` in bytes (or the current position if ``size`` is
not specified).
Raises:
UnsupportedOperation: Always, since the reader is not writable.
Note:
:meth:`io.IOBase.writable` gates this method as well as :meth:`write` -- "If
:data:`False`, :meth:`write` and :meth:`truncate` will raise :exc:`OSError`" -- and
:meth:`writable` here returns :data:`False`, so refusing is what the contract asks
for. It is also what CPython's own read-only buffered readers do: both the
accelerated :class:`io.BufferedReader` and the pure-Python
``_pyio.BufferedReader`` raise :exc:`io.UnsupportedOperation`, the latter from
``_BufferedIOMixin.truncate``'s ``_checkWritable()`` (:issue:`645`).
The resizing this used to perform is still reachable internally, as
:meth:`_truncate_buffer`. It never touched the underlying stream in the first place
-- it resizes a private lookback window -- which is why it survives under a private
name rather than being removed with the public method's behaviour.
"""
raise UnsupportedOperation('truncate')
def _truncate_buffer(self, size: 'int | None' = None, /) -> 'int':
"""Resize the lookback buffer to the given ``size`` in bytes (or the current position if
``size`` is not specified). The current stream position isn't changed. This resizing can
extend or reduce the current buffer size. In case of extension, the new area is
zero-filled. The new buffer size is returned.
Note:
This is the internal half of what :meth:`truncate` used to do, which is all of it:
nothing here writes to the underlying stream -- :meth:`write` raises -- so what
this resizes is the buffer, not the stream behind it. That is why :meth:`truncate`
refuses (:issue:`645`) while this remains: the operation is a private-window one,
not an :class:`io.IOBase` write. The buffer is a sliding window over a stream that
cannot be seeked: its octet 0 sits at absolute offset ``_buffer_set``, its content
occupies ``[0:_buffer_cur]``, and everything past that is padding never read.
Two consequences for which octets survive. An extension appends its zero octets at
the **tail**, the new area being by definition the region past the old end. A
reduction below the content keeps the **most recent** ``size`` octets and advances
``_buffer_set`` past the ones it drops, because what this holds is lookback: the
octets it can still answer for are the ones just read, and the stream is already
beyond them. Padding is never kept in preference to content either way.
Advancing ``_buffer_set`` is what keeps ``_buffer_set + _buffer_cur`` equal to how far
the stream has actually been consumed, which :meth:`seek` relies on to decide whether
it may read ahead to fill a gap. A reduction that shrank the window without moving its
base would leave that sum short of the stream, and the next forward :meth:`seek` would
splice in octets from the wrong absolute offset without complaining.
Octets dropped by a reduction are gone for good, since the stream cannot be rewound to
re-supply them. A position left among them is then before the window, which
:meth:`seek` refuses as it refuses any other.
"""
if size is None:
# NOTE: an unspecified size means the current position, following the
# :meth:`io.IOBase.truncate` convention this used to implement. The buffer
# is indexed relative to ``_buffer_set``, and the position may sit before it
# once a saved buffer has been rewound, in which case nothing is kept.
size = max(self._tell - self._buffer_set, 0)
if size < 0:
raise TruncateError(f'negative size value {size}')
# NOTE: the position isn't changed by a truncation, but rebuilding the buffer
# resets it, so it is read here and put back below -- moved down by whatever
# the window's base moved up, so that it still denotes the same octet.
buffer_pos = self._buffer.tell()
self._buffer_view.release()
# NOTE: only ``[0:_buffer_cur]`` is content. Slicing the buffer itself would
# keep padding that was never read and count it as though it were data.
temp = self._buffer.getvalue()[:self._buffer_cur]
dropped = max(len(temp) - size, 0)
self._buffer = io.BytesIO(temp[dropped:].ljust(size, b'\x00'))
self._buffer_view = self._buffer.getbuffer()
self._buffer.seek(max(buffer_pos - dropped, 0), io.SEEK_SET)
self._buffer_size = size
# NOTE: ``_buffer_set + _buffer_cur`` is unchanged by construction -- the base
# gains exactly what the content pointer loses -- so the stream's consumption
# point still reads correctly out of the pair.
self._buffer_set += dropped
self._buffer_cur = len(temp) - dropped
return self._buffer_size
[docs]
def writable(self) -> 'bool':
"""Return :obj:`True` if the stream supports writing. If :obj:`False`, :meth:`write` and
:meth:`truncate` will raise :exc:`OSError`.
Note:
This was spelled ``writeable`` until :issue:`645`, which is not how the :mod:`io`
protocol spells it, so it overrode nothing and :mod:`io` never consulted it -- the
inherited :meth:`io.IOBase.writable` answered instead. Both returned :data:`False`,
so there was no observable divergence to notice; the coincidence is what hid it.
"""
return False
def writelines(self, lines: 'Iterable[Buffer]', /) -> 'None':
"""Write a list of lines to the stream. Line separators are not added, so it is usual for
each of the lines provided to have a line separator at the end."""
raise UnsupportedOperation('write')
[docs]
def detach(self) -> 'RawIOBase':
"""Separate the underlying raw stream from the buffer and return it.
After the raw stream has been detached, the buffer is in an unusable state.
Some buffers, like :class:`~io.BytesIO`, do not have the concept of a single raw stream to
return from this method. They raise :exc:`~io.UnsupportedOperation`.
"""
if hasattr(self._stream, 'detach'):
return self._stream.detach()
raise UnsupportedOperation('detach')
[docs]
def read(self, size: 'int | None' = -1, /) -> 'bytes':
"""Read and return ``size`` bytes, or if ``size`` is not given or negative, until EOF or if
the read call would block in non-blocking mode."""
if size is None or size < 0:
size = -1
if self._tell >= self._buffer_set + self._buffer_cur:
buf = self._stream.read(size)
self._write_buffer(buf)
else:
if self._buffer_file is not None and self._tell < self._buffer_set:
with open(self._buffer_path, 'rb') as temp_file:
temp_file.seek(self._tell, io.SEEK_SET)
buf = temp_file.read(size)
elif not size:
# NOTE: as in :meth:`readline` -- no octets wanted, so no window check.
buf = b''
else:
buf_rem = self._seek_buffer()
buf = self._buffer.read(buf_rem if size < 0 else min(size, buf_rem))
size_rem = -1
if size < 0 or (size_rem := size - len(buf)) > 0:
buf_tmp = self._stream.read(size_rem)
self._write_buffer(buf_tmp)
buf += buf_tmp
self._tell += len(buf)
return buf
[docs]
def read1(self, size: 'int | None' = -1, /) -> 'bytes':
"""Read and return up to ``size`` bytes with only one call on the raw stream. If at least
one byte is buffered, only buffered bytes are returned. Otherwise, one raw stream read call
is made."""
if size is None:
size = -1
if self._tell >= self._buffer_set + self._buffer_cur:
if hasattr(self._stream, 'read1'):
buf = self._stream.read1(size)
else:
buf = self._stream.read(size)
self._write_buffer(buf)
else:
if self._buffer_file is not None and self._tell < self._buffer_set:
with open(self._buffer_path, 'rb') as temp_file:
temp_file.seek(self._tell, io.SEEK_SET)
buf = temp_file.read1(size)
elif not size:
# NOTE: as in :meth:`readline` -- no octets wanted, so no window check.
buf = b''
else:
# NOTE: no continuation from the stream when the buffer answered, since
# :meth:`read1` is specified to return only buffered octets if any are.
buf_rem = self._seek_buffer()
buf = self._buffer.read1(buf_rem if size < 0 else min(size, buf_rem))
if not buf: # pragma: no cover
size_rem = -1
if size < 0 or (size_rem := size - len(buf)) > 0:
if hasattr(self._stream, 'read1'):
buf_tmp = self._stream.read1(size_rem)
else:
buf_tmp = self._stream.read(size_rem)
self._write_buffer(buf_tmp)
buf += buf_tmp
self._tell += len(buf)
return buf
[docs]
def readinto(self, b: 'Buffer', /) -> 'int':
"""Read bytes into a pre-allocated, writable :term:`bytes-like object` ``b`` and return the
number of bytes read. For example, ``b`` might be a :obj:`bytearray`.
Like :meth:`read`, multiple reads may be issued to the underlying raw stream, unless the
latter is interactive.
A :exc:`BlockingIOError` is raised if the underlying raw stream is in non blocking-mode,
and has no data available at the moment.
"""
if TYPE_CHECKING:
b = cast('memoryview', b)
buf = self.read(len(b))
buf_len = len(buf)
b[:buf_len] = buf
return buf_len
[docs]
def readinto1(self, b: 'Buffer', /) -> 'int':
"""Read bytes into a pre-allocated, writable :term:`bytes-like object` ``b``, using at most
one call to the underlying raw stream's :meth:`read` (or :meth:`readinto`) method. Return
the number of bytes read.
A :exc:`BlockingIOError` is raised if the underlying raw stream is in non blocking-mode,
and has no data available at the moment.
"""
if TYPE_CHECKING:
b = cast('memoryview', b)
buf = self.read1(len(b))
buf_len = len(buf)
b[:buf_len] = buf
return buf_len
[docs]
def write(self, b: 'Buffer', /) -> 'int':
"""Write the given :term:`bytes-like object`, ``b``, and return the number of bytes written
(always equal to the length of ``b`` in bytes, since if the write fails an :exc:`OSError`
will be raised). Depending on the actual implementation, these bytes may be readily written
to the underlying stream, or held in a buffer for performance and latency reasons.
When in non-blocking mode, a :exc:`BlockingIOError` is raised if the data needed to be
written to the raw stream but it couldn't accept all the data without blocking.
The caller may release or mutate ``b`` after this method returns, so the implementation
should only access ``b`` during the method call.
"""
raise UnsupportedOperation('write')
[docs]
def peek(self, size: 'int' = 0) -> 'bytes':
"""Return bytes from the stream without advancing the position.
At most one single read on the raw stream is done to satisfy the call.
The number of bytes returned may be less or more than requested.
"""
if self._tell >= self._buffer_set + self._buffer_cur:
if hasattr(self._stream, 'peek'):
buf = self._stream.peek(size)
else:
buf = self._stream.read(size)
self._write_buffer(buf)
else:
if self._buffer_file is not None and self._tell < self._buffer_set:
with open(self._buffer_path, 'rb') as temp_file:
temp_file.seek(self._tell, io.SEEK_SET)
buf = temp_file.peek(size)
elif not size:
# NOTE: as in :meth:`readline` -- no octets wanted, so no window check. Only
# the buffered branch is short-circuited: the branch above hands a bare
# ``peek()`` to the raw stream, whose own ``peek(0)`` may legitimately
# return a whole buffer's worth, and that is left as it was.
buf = b''
else:
# NOTE: read through the view rather than through ``self._buffer``, whose
# cursor an ordinary read would advance. A preview must leave the position
# alone, and the buffer's cursor is part of the position: advancing it here
# while leaving ``_tell`` untouched is what made the *next* buffered read
# start from the wrong octet, with ``tell()`` reporting the right one.
buf_rem = self._seek_buffer()
buf_off = self._buffer.tell()
buf = bytes(self._buffer_view[
buf_off:buf_off + (buf_rem if size < 0 else min(size, buf_rem))])
if not buf and len(buf) < size: # pragma: no cover
size_rem = -1
if size < 0 or (size_rem := size - len(buf)) > 0:
if hasattr(self._stream, 'peek'):
buf_tmp = self._stream.peek(size_rem)
else:
buf_tmp = self._stream.read(size_rem)
self._write_buffer(buf_tmp)
buf += buf_tmp
return buf