<commit_before>import tornado
import tornado.gen
import tornado.ioloop
from tornado.iostream import PipeIOStream, StreamClosedError
from ..exceptions import StreamingException
from ..messages.common import StreamState
from ..messages import common

try:
    from tornado.locks import Condition
except ImportError:  # pragma: no cover
    from toro import Condition

from collections import deque


class Stream(object):

    def __init__(self):
        self._state = StreamState.init

    def read(self):
        """Async read from internal stream buffer

        if it reaches the end of the stream, it will return empty(""), caller
        will depend on the return value to tell if it reaches the end of the
        stream.

        It doesn't support seek functionality, which means it will only read
        the stream data once in single direction.

        NOTE: The data which has been read will be discarded in the stream.

        :return:chunk of bytes read from stream
        """
        raise NotImplementedError()

    def write(self, chunk):
        """Async write to internal stream buffer

        :raises StreamingException:
            if stream has been closed, it will raise StreamingException
        """
        raise NotImplementedError()

    def close(self):
        raise NotImplementedError()


class InMemStream(Stream):

    def __init__(self, buf=None, auto_close=True):
        """In-Memory based stream

        :param buf: the buffer for the in memory stream
        """
        super(InMemStream, self).__init__()
        self._stream = deque()
        if buf:
            self._stream.append(buf)
        self._condition = Condition()
        self.auto_close = auto_close

    @tornado.gen.coroutine
    def read(self):
        if self._state != StreamState.completed and len(self._stream) == 0:
            yield self._condition.wait()

        chunk = ""
        while len(self._stream) > 0 and len(chunk) < common.MAX_PAYLOAD_SIZE:
            chunk += self._stream.popleft()

        raise tornado.gen.Return(chunk)

    @tornado.gen.coroutine
    def write(self, chunk):
        if self._state == StreamState.completed:
            raise StreamingException("Stream has been closed.")

        self._stream.append(chunk)
        self._condition.notify()

    @tornado.gen.coroutine
    def close(self):
        self._state = StreamState.completed
        self._condition.notify()


class PipeStream(Stream):

    def __init__(self, rpipe, wpipe=None, auto_close=False):
        """Pipe-based stream

        NOTE: reading from or writing to files, use os.open to get the file
        descriptor instead of python's open. Socket file descriptors and
        others are fine.

        when you use os.pipe to generate one write pipe and one read pipe, you
        need to pass both of them into init method.

        :param rpipe: an integer file descriptor which supports read ops
        :param wpipe: an integer file descriptor which supports write ops
        :param auto: flag to indicate to close the stream automatically or not
        """
        super(PipeStream, self).__init__()
        assert rpipe is not None
        self._rpipe = rpipe
        self._wpipe = wpipe
<commit_msg>add none check for buf to write<commit_after>import tornado
import tornado.gen
import tornado.ioloop
from tornado.iostream import PipeIOStream, StreamClosedError
from ..exceptions import StreamingException
from ..messages.common import StreamState
from ..messages import common

try:
    from tornado.locks import Condition
except ImportError:  # pragma: no cover
    from toro import Condition

from collections import deque


class Stream(object):

    def __init__(self):
        self._state = StreamState.init

    def read(self):
        """Async read from internal stream buffer

        if it reaches the end of the stream, it will return empty(""), caller
        will depend on the return value to tell if it reaches the end of the
        stream.

        It doesn't support seek functionality, which means it will only read
        the stream data once in single direction.

        NOTE: The data which has been read will be discarded in the stream.

        :return:chunk of bytes read from stream
        """
        raise NotImplementedError()

    def write(self, chunk):
        """Async write to internal stream buffer

        :raises StreamingException:
            if stream has been closed, it will raise StreamingException
        """
        raise NotImplementedError()

    def close(self):
        raise NotImplementedError()


class InMemStream(Stream):

    def __init__(self, buf=None, auto_close=True):
        """In-Memory based stream

        :param buf: the buffer for the in memory stream
        """
        super(InMemStream, self).__init__()
        self._stream = deque()
        if buf:
            self._stream.append(buf)
        self._condition = Condition()
        self.auto_close = auto_close

    @tornado.gen.coroutine
    def read(self):
        if self._state != StreamState.completed and len(self._stream) == 0:
            yield self._condition.wait()

        chunk = ""
        while len(self._stream) > 0 and len(chunk) < common.MAX_PAYLOAD_SIZE:
            chunk += self._stream.popleft()

        raise tornado.gen.Return(chunk)

    @tornado.gen.coroutine
    def write(self, chunk):
        if self._state == StreamState.completed:
            raise StreamingException("Stream has been closed.")
        if chunk:
            self._stream.append(chunk)
            self._condition.notify()

    @tornado.gen.coroutine
    def close(self):
        self._state = StreamState.completed
        self._condition.notify()


class PipeStream(Stream):

    def __init__(self, rpipe, wpipe=None, auto_close=False):
        """Pipe-based stream

        NOTE: reading from or writing to files, use os.open to get the file
        descriptor instead of python's open. Socket file descriptors and
        others are fine.

        when you use os.pipe to generate one write pipe and one read pipe, you
        need to pass both of them into init method.

        :param rpipe: an integer file descriptor which supports read ops
        :param wpipe: an integer file descriptor which supports write ops
        :param auto: flag to indicate to close the stream automatically or not
        """
        super(PipeStream, self).__init__()
        assert rpipe is not None
        self._rpipe = rpipe
        self._wpipe = wpipe

{'fraction_non_alphanumeric': 0.07273566046440104, 'fraction_numerical': 0.0006151007227433493, 'mean_word_length': 2.481798715203426, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7228318', 'n_tokens_mistral': 1729, 'n_tokens_neox': 1651, 'n_words': 648}