@@ -0,0 +1,1522 @@
+from abc import ABC, abstractmethod%0Afrom typing import Generic, List, Iterable%0Aimport asyncio%0A%0Afrom google.cloud.pubsublite.internal.wire.connection import Request, Response%0Afrom google.cloud.pubsublite.internal.wire.work_item import WorkItem%0A%0A%0Aclass BatchTester(Generic%5BRequest%5D, ABC):%0A  %22%22%22A BatchTester determines whether a given batch of messages must be sent.%22%22%22%0A  @abstractmethod%0A  def test(self, requests: Iterable%5BRequest%5D) -%3E bool:%0A    %22%22%22%0A    Args:%0A      requests: The current outstanding batch.%0A%0A    Returns: Whether that batch must be sent.%0A    %22%22%22%0A    raise NotImplementedError()%0A%0A%0Aclass SerialBatcher(Generic%5BRequest, Response%5D):%0A  _tester: BatchTester%5BRequest%5D%0A  _requests: List%5BWorkItem%5BRequest%5D%5D  # A list of outstanding requests%0A%0A  def __init__(self, tester: BatchTester%5BRequest%5D):%0A    self._tester = tester%0A    self._requests = %5B%5D%0A%0A  def add(self, request: Request) -%3E 'asyncio.Future%5BResponse%5D':%0A    %22%22%22Add a new request to this batcher. Callers must always call should_flush() after add, and flush() if that returns%0A    true.%0A%0A    Args:%0A      request: The request to send.%0A%0A    Returns:%0A      A future that will resolve to the response or a GoogleAPICallError.%0A    %22%22%22%0A    item = WorkItem%5BRequest%5D(request)%0A    self._requests.append(item)%0A    return item.response_future%0A%0A  def should_flush(self) -%3E bool:%0A    return self._tester.test(item.request for item in self._requests)%0A%0A  def flush(self) -%3E Iterable%5BWorkItem%5BRequest%5D%5D:%0A    requests = self._requests%0A    self._requests = %5B%5D%0A    return requests%0A

Implement SerialBatcher which helps with transforming single writes into batch writes. (#7)