class_docstr: 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.
  
  Batching several messages at once helps performance when sending
  a myriad of tiny messages.
  
  Examples
  --------
  >>> stream = await connect(address)
  >>> bstream = BatchedSend(interval='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!']
class_docstr_tok: ["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",".","Batching","several","messages","at","once","helps","performance","when","sending","a","myriad","of","tiny","messages",".","Examples","--","--","--","--",">",">",">","stream","=","await","connect","(","address",")",">",">",">","bstream","=","BatchedSend","(","interval='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","!","'","]"]
code: def __init__(self, interval, loop=None, serializers=None):
      # XXX is the loop arg useful?
      self.loop = loop or IOLoop.current()
      self.interval = parse_timedelta(interval, default="ms")
      self.waker = locks.Event()
      self.stopped = locks.Event()
      self.please_stop = False
      self.buffer = []
      self.comm = None
      self.message_count = 0
      self.batch_count = 0
      self.byte_count = 0
      self.next_deadline = None
      self.recent_message_log = deque(
          maxlen=dask.config.get(
              "distributed.admin.low-level-log-length"
          )
      )
      self.serializers = serializers
      self._consecutive_failures = 0
  
imports: from __future__ import annotations
  import contextlib
  import logging
  from collections import deque
  from typing import Any
  from tornado import gen, locks
  from tornado.ioloop import IOLoop
  import dask
  from dask.utils import parse_timedelta
  from distributed.core import CommClosedError
  from distributed.metrics import time
label_desc: Use image node_id 1 to create a new BatchedSend object  with example: obj = BatchedSend(interval, loop, serializers)