1  import asyncore
   2  import socket
   3  import sys
   4  from abc import abstractmethod
   5  from typing import Optional, Sequence, Tuple, Union
   6  
   7  class simple_producer:
   8      def __init__(self, data: bytes, buffer_size: int = ...) -> None: ...
   9      def more(self) -> bytes: ...
  10  
  11  class async_chat(asyncore.dispatcher):
  12      ac_in_buffer_size: int
  13      ac_out_buffer_size: int
  14      def __init__(self, sock: Optional[socket.socket] = ..., map: Optional[asyncore._maptype] = ...) -> None: ...
  15      @abstractmethod
  16      def collect_incoming_data(self, data: bytes) -> None: ...
  17      @abstractmethod
  18      def found_terminator(self) -> None: ...
  19      def set_terminator(self, term: Union[bytes, int, None]) -> None: ...
  20      def get_terminator(self) -> Union[bytes, int, None]: ...
  21      def handle_read(self) -> None: ...
  22      def handle_write(self) -> None: ...
  23      def handle_close(self) -> None: ...
  24      def push(self, data: bytes) -> None: ...
  25      def push_with_producer(self, producer: simple_producer) -> None: ...
  26      def readable(self) -> bool: ...
  27      def writable(self) -> bool: ...
  28      def close_when_done(self) -> None: ...
  29      def initiate_send(self) -> None: ...
  30      def discard_buffers(self) -> None: ...
  31  
  32  if sys.version_info < (3, 0):
  33      class fifo:
  34          def __init__(self, list: Sequence[Union[bytes, simple_producer]] = ...) -> None: ...
  35          def __len__(self) -> int: ...
  36          def is_empty(self) -> bool: ...
  37          def first(self) -> bytes: ...
  38          def push(self, data: Union[bytes, simple_producer]) -> None: ...
  39          def pop(self) -> Tuple[int, bytes]: ...
  40