Source code for wolframclient.serializers.wxfencoder.streaming
from __future__ import absolute_import, print_function, unicode_literals
from wolframclient.utils import six
from wolframclient.utils.api import zlib
from wolframclient.utils.encoding import force_bytes
[docs]class ZipCompressedWriter(object):
def __init__(self, writer):
""" Write zip compressed data to a given buffer writer. """
self._compressor = zlib.compressobj()
self._writer = writer
[docs] def flush(self):
""" Must be called when closing or destroying the object."""
self._writer.write(self._compressor.flush())
[docs] def write(self, data):
""" Write the compression of `data` to the underlying buffer writer. """
self._writer.write(self._compressor.compress(force_bytes(data)))
def __enter__(self):
return self
def __exit__(self, type, value, traceback):
self.flush()
[docs]class ExactSizeReader(object):
""" Read exactly the amount of bytes requested and fails otherwise."""
def __init__(self, reader):
self._reader = reader
[docs] def read(self, size=-1):
"""Read from an underlying readable object.
If size is negative the data is read until EOF.
If it is 0 then b'' is returned.
Otherwise the exact amount of bytes is read from the source. More than
one call may be necessary.
"""
data = self._reader.read(size)
# Negative values read until EOF and 0 returns b''. Both remain unchanged.
# Also a fast path when the requested amount of bytes is returned in one go.
if size <= 0 or len(data) == size:
return data
# need an intermediary buffer
out_len = len(data)
data = six.BytesIO(data)
while out_len < size:
chunk = self._reader.read(size - out_len)
if chunk == b"":
raise EOFError("Not enough data to read.")
data.write(chunk)
out_len = out_len + len(chunk)
return data.getvalue()
[docs]class ZipCompressedReader(object):
"""A buffer implementation reading zip compressed data from a source buffer and returning uncompressed data.
This class is instantiated from a reader, any object implementing a :meth:`~io.BufferedIOBase.read` method.
"""
CHUNK_SIZE = 8192
def __init__(self, reader):
""" Read zip compressed data from a given buffer reader."""
self._compressor = zlib.decompressobj()
self._reader = reader
[docs] def read(self, size=-1):
"""Read from a compressed stream of bytes and return the inflated byte sequence.
Parameter `size` specifies the amount of bytes to return at most. If `size` is set to -1
then the reader is read until EOF is reached.
"""
if size is None or size < 0:
chunk_size = -1
size = -1
else:
chunk_size = ZipCompressedReader.CHUNK_SIZE
out_data = six.BytesIO()
out_len = 0
while True:
# first step find try to find some data to uncompress.
# sometimes some bytes are left over. We have to send them first to zlib.
if self._compressor.unconsumed_tail != b"":
data_in = self._compressor.unconsumed_tail
else:
# read more data from input reader. Read in chunk since we can't guess how
# big the inflated result is.
data_in = self._reader.read(chunk_size)
# no more data is available.
if data_in == b"":
break
# second step, decompress the new chunk
if size > 0:
chunk = self._compressor.decompress(data_in, size - out_len)
else:
chunk = self._compressor.decompress(data_in)
# increment output len.
out_len = out_len + len(chunk)
# write to buffer
out_data.write(chunk)
# check requested size against output length.
if size > 0 and out_len == size:
break
return out_data.getvalue()