<commit_before>from __future__ import print_function, division, absolute_import

import logging
from timeit import default_timer

from tornado import gen
from tornado.queues import Queue
from tornado.iostream import StreamClosedError
from tornado.ioloop import PeriodicCallback, IOLoop

from .core import read, write
from .utils import log_errors


logger = logging.getLogger(__name__)


class BatchedSend(object):
    """ Batch messages in batches on a stream

    This takes an IOStream and an interval (in ms) and ensures that we send no
    more than one message every interval milliseconds.  We send lists of
    messages.

    Example
    -------
    >>> stream = yield connect(ip, port)
    >>> bstream = BatchedSend(interval=10)  # 10 ms
    >>> bstream.start(stream)
    >>> bstream.send('Hello,')
    >>> bstream.send('world!')

    On the other side, the recipient will get a message like the following::

        ['Hello,', 'world!']
    """
    def __init__(self, interval, loop=None):
        self.loop = loop or IOLoop.current()
        self.interval = interval / 1000.
        self.last_transmission = 0
        self.buffer = []
        self.stream = None
        self.last_payload = []
        self.last_send = gen.sleep(0)

    def start(self, stream):
        self.stream = stream
        if self.buffer:
            self.send_next()

    def __str__(self):
        return '<BatchedSend: %d in buffer>' % len(self.buffer)

    __repr__ = __str__

    @gen.coroutine
    def send_next(self, wait=True):
        try:
            now = default_timer()
            if wait:
                wait_time = min(self.last_transmission + self.interval - now,
                                self.interval)
                yield gen.sleep(wait_time)
            yield self.last_send
            self.buffer, payload = [], self.buffer
            self.last_payload = payload
            self.last_transmission = now
            self.last_send = write(self.stream, payload)
        except Exception as e:
            logger.exception(e)

    @gen.coroutine
    def _write(self, payload):
        yield gen.sleep(0)
        yield write(self.stream, payload)

    def send(self, msg):
        """ Send a message to the other side

        This completes quickly and synchronously
        """
        try:
            if self.stream is None:  # not yet started
                self.buffer.append(msg)<commit_msg>Put timeout on last_send in batched<commit_after>from __future__ import print_function, division, absolute_import

import logging
from timeit import default_timer

from tornado import gen
from tornado.queues import Queue
from tornado.iostream import StreamClosedError
from tornado.ioloop import PeriodicCallback, IOLoop

from .core import read, write
from .utils import log_errors


logger = logging.getLogger(__name__)


class BatchedSend(object):
    """ Batch messages in batches on a stream

    This takes an IOStream and an interval (in ms) and ensures that we send no
    more than one message every interval milliseconds.  We send lists of
    messages.

    Example
    -------
    >>> stream = yield connect(ip, port)
    >>> bstream = BatchedSend(interval=10)  # 10 ms
    >>> bstream.start(stream)
    >>> bstream.send('Hello,')
    >>> bstream.send('world!')

    On the other side, the recipient will get a message like the following::

        ['Hello,', 'world!']
    """
    def __init__(self, interval, loop=None):
        self.loop = loop or IOLoop.current()
        self.interval = interval / 1000.
        self.last_transmission = 0
        self.buffer = []
        self.stream = None
        self.last_payload = []
        self.last_send = gen.sleep(0)

    def start(self, stream):
        self.stream = stream
        if self.buffer:
            self.send_next()

    def __str__(self):
        return '<BatchedSend: %d in buffer>' % len(self.buffer)

    __repr__ = __str__

    @gen.coroutine
    def send_next(self, wait=True):
        try:
            now = default_timer()
            if wait:
                wait_time = min(self.last_transmission + self.interval - now,
                                self.interval)
                yield gen.sleep(wait_time)
            while self.stream._write_buffer:
                yield gen.with_timeout(timedelta(milliseconds=10),
                                       self.last_send)  # hangs otherwise?
            self.buffer, payload = [], self.buffer
            self.last_payload = payload
            self.last_transmission = now
            self.last_send = write(self.stream, payload)
        except Exception as e:
            logger.exception(e)

    @gen.coroutine
    def _write(self, payload):
        yield gen.sleep(0)
        yield write(self.stream, payload)

    def send(self, msg):
        """ Send a message to the other side

        This completes quickly and synchronously
        """
        try:
            if self.stream is None:  # not yet started
                self.buffer.append(msg)
{'fraction_non_alphanumeric': 0.09072741806554756, 'fraction_numerical': 0.004796163069544364, 'mean_word_length': 2.1617182564750475, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 35, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '475861', 'n_tokens_mistral': 1422, 'n_tokens_neox': 1352, 'n_words': 421}