docstr: Async queue that accepts individual items and returns batches.
  
  Respects max_batch_size and timeout_s; a batch will be returned when
  max_batch_size elements are available or the timeout has passed since
  the previous get.
  
  If handle_batch_func is passed in, a background coroutine will run to
  poll from the queue and call handle_batch_func on the results.
  
  Cannot be pickled.
  
  Arguments:
      max_batch_size: max number of elements to return in a batch.
      timeout_s: time to wait before returning an incomplete
          batch.
      handle_batch_func(Optional[Callable]): callback to run in the
          background to handle batches if provided.
docstr_tok: ["Async","queue","that","accepts","individual","items","and","returns","batches",".","Respects","max_batch_size","and","timeout_s",";","a","batch","will","be","returned","when","max_batch_size","elements","are","available","or","the","timeout","has","passed","since","the","previous","get",".","If","handle_batch_func","is","passed","in",",","a","background","coroutine","will","run","to","poll","from","the","queue","and","call","handle_batch_func","on","the","results",".","Can","not","be","pickled",".","Arguments",":","max_batch_size",":","max","number","of","elements","to","return","in","a","batch",".","timeout_s",":","time","to","wait","before","returning","an","incomplete","batch",".","handle_batch_func","(","Optional","[","Callable","]",")",":","callback","to","run","in","the","background","to","handle","batches","if","provided","."]
code: def __init__(
      self,
      max_batch_size: int,
      batch_wait_timeout_s: float,
      handle_batch_func: Optional[Callable] = None,
  ) -> None:
      """Async queue that accepts individual items and returns batches.
  
      Respects max_batch_size and timeout_s; a batch will be returned when
      max_batch_size elements are available or the timeout has passed since
      the previous get.
  
      If handle_batch_func is passed in, a background coroutine will run to
      poll from the queue and call handle_batch_func on the results.
  
      Cannot be pickled.
  
      Arguments:
          max_batch_size: max number of elements to return in a batch.
          timeout_s: time to wait before returning an incomplete
              batch.
          handle_batch_func(Optional[Callable]): callback to run in the
              background to handle batches if provided.
      """
      self.queue: asyncio.Queue[_SingleRequest] = asyncio.Queue()
      self.max_batch_size = max_batch_size
      self.batch_wait_timeout_s = batch_wait_timeout_s
      self.queue_put_event = asyncio.Event()
  
      self._handle_batch_task = None
      if handle_batch_func is not None:
          self._handle_batch_task = (
              get_or_create_event_loop().create_task(
                  self._process_batches(handle_batch_func)
              )
          )
  
imports: import asyncio
  import time
  from dataclasses import dataclass
  from functools import wraps
  from inspect import isasyncgenfunction, iscoroutinefunction
  from typing import Any, AsyncGenerator, Callable, Dict, Iterable, List, Optional, Tuple, Type, TypeVar, overload
  from ray._private.signature import extract_signature, flatten_args, recover_args
  from ray._private.utils import get_or_create_event_loop
  from ray.serve._private.utils import extract_self_if_method_call
  from ray.serve.exceptions import RayServeException
  from ray.util.annotations import PublicAPI
label_desc: Use image node_id 1 to create a new _BatchQueue object  with example: obj = _BatchQueue(max_batch_size, batch_wait_timeout_s, handle_batch_func)