How to use the grpclib.protocol.EventsProcessor function in grpclib

await client_stream.send_request(create_headers(), _processor=client_proc)

    initial_window = server_h2c.local_settings.initial_window_size
    assert (client_h2c.local_flow_control_window(client_stream.id)
            == initial_window)

    # data should be bigger than window size
    data = b'0' * (initial_window + 1)
    size = len(data)

    # sending less than a full message
    await client_stream.send_data(data[:initial_window - 1])

    # let server process it's events
    server_processor = EventsProcessor(DummyHandler(), server_conn)
    to_server_transport.process(server_processor)

    # checking window size was decreased
    assert client_h2c.local_flow_control_window(client_stream.id) == 1

    # simulate that server is waiting for the size of a message and should
    # acknowledge that size as soon as it will be received
    server_stream, = server_processor.streams.values()
    recv_task = loop.create_task(server_stream.recv_data(size))
    await asyncio.wait([recv_task], timeout=.01)
    assert server_stream.buffer._acked_size == initial_window - 1

    # check that server acknowledged received partial data
    assert client_h2c.local_flow_control_window(client_stream.id) > 1

    # sending remaining data and recv_task should finish

async def test_stream_release(config):
    client_h2c, server_h2c = create_connections()

    to_client_transport = TransportStub(client_h2c)
    server_conn = Connection(server_h2c, to_client_transport, config=config)

    to_server_transport = TransportStub(server_h2c)
    client_conn = Connection(client_h2c, to_server_transport, config=config)

    client_processor = EventsProcessor(DummyHandler(), client_conn)
    client_stream = client_conn.create_stream()

    server_processor = EventsProcessor(DummyHandler(), server_conn)

    assert not client_processor.streams
    client_release_stream = await client_stream.send_request(
        create_headers(), _processor=client_processor,
    )
    assert client_release_stream and client_processor.streams

    # sending data and closing stream on the client-side
    msg = b'message'
    await client_stream.send_data(msg, end_stream=True)
    events1 = to_server_transport.process(server_processor)
    assert any(isinstance(e, StreamEnded) for e in events1), events1

    # intentionally sending some stream-specific frame after stream was
    # half-closed
    client_h2c.increment_flow_control_window(10, stream_id=client_stream.id)

async def test_ping(config):
    client_h2c, server_h2c = create_connections()

    to_client_transport = TransportStub(client_h2c)
    server_conn = Connection(server_h2c, to_client_transport, config=config)

    to_server_transport = TransportStub(server_h2c)
    client_conn = Connection(client_h2c, to_server_transport, config=config)

    client_processor = EventsProcessor(DummyHandler(), client_conn)
    server_processor = EventsProcessor(DummyHandler(), server_conn)

    client_h2c.ping(b'12345678')
    client_conn.flush()

    to_server_transport.process(server_processor)
    server_conn.flush()

    ping_ack, = to_client_transport.process(client_processor)
    assert isinstance(ping_ack, PingAcknowledged)
    assert ping_ack.ping_data == b'12345678'

async def test_send_trailing_metadata_on_closed_stream(loop, config):
    client_h2c, server_h2c = create_connections()

    to_client_transport = TransportStub(client_h2c)
    to_server_transport = TransportStub(server_h2c)

    client_conn = Connection(client_h2c, to_server_transport, config=config)
    server_conn = Connection(server_h2c, to_client_transport, config=config)

    server_proc = EventsProcessor(DummyHandler(), server_conn)
    client_proc = EventsProcessor(DummyHandler(), client_conn)

    client_h2_stream = client_conn.create_stream()
    await client_h2_stream.send_request(create_headers(),
                                        _processor=client_proc)

    request = DummyRequest(value='ping')
    await send_message(client_h2_stream, ProtoCodec(), request, DummyRequest,
                       end=True)
    to_server_transport.process(server_proc)

    server_h2_stream = server_proc.handler.stream
    request_metadata = decode_metadata(server_proc.handler.headers)

    send_trailing_metadata_done = False
    async with mk_stream(server_h2_stream, request_metadata) as server_stream:

async def test_send_trailing_metadata_on_closed_stream(loop, config):
    client_h2c, server_h2c = create_connections()

    to_client_transport = TransportStub(client_h2c)
    to_server_transport = TransportStub(server_h2c)

    client_conn = Connection(client_h2c, to_server_transport, config=config)
    server_conn = Connection(server_h2c, to_client_transport, config=config)

    server_proc = EventsProcessor(DummyHandler(), server_conn)
    client_proc = EventsProcessor(DummyHandler(), client_conn)

    client_h2_stream = client_conn.create_stream()
    await client_h2_stream.send_request(create_headers(),
                                        _processor=client_proc)

    request = DummyRequest(value='ping')
    await send_message(client_h2_stream, ProtoCodec(), request, DummyRequest,
                       end=True)
    to_server_transport.process(server_proc)

    server_h2_stream = server_proc.handler.stream
    request_metadata = decode_metadata(server_proc.handler.headers)

    send_trailing_metadata_done = False
    async with mk_stream(server_h2_stream, request_metadata) as server_stream:
        await server_stream.send_trailing_metadata(status=Status.UNKNOWN)

async def test_released_stream_data_ack(config):
    client_h2c, server_h2c = create_connections()
    initial_window = client_h2c.outbound_flow_control_window
    # should be large enough to trigger WINDOW_UPDATE frame
    data_size = initial_window - 1

    to_client_transport = TransportStub(client_h2c)
    server_handler = DummyHandler()
    server_conn = Connection(server_h2c, to_client_transport, config=config)
    server_proc = EventsProcessor(server_handler, server_conn)

    to_server_transport = TransportStub(server_h2c)
    client_conn = Connection(client_h2c, to_server_transport, config=config)
    client_proc = EventsProcessor(DummyHandler(), client_conn)
    client_stream = client_conn.create_stream()

    await client_stream.send_request(create_headers(), _processor=client_proc)
    to_server_transport.process(server_proc)

    # server_handler.stream.recv_data(data_size) intentionally not called
    await server_handler.stream.send_headers([  # trailers-only error
        (':status', '200'),
        ('content-type', 'application/grpc+proto'),
        ('grpc-status', str(Status.UNKNOWN.value)),
    ], end_stream=True)
    to_client_transport.process(client_proc)

async def test_client_reset(
    loop, caplog, handler, level, msg, exc_type, exc_text, config,
):
    caplog.set_level(logging.INFO)
    client_h2c, server_h2c = create_connections()

    to_client_transport = TransportStub(client_h2c)
    to_server_transport = TransportStub(server_h2c)

    client_conn = Connection(client_h2c, to_server_transport, config=config)
    server_conn = Connection(server_h2c, to_client_transport, config=config)

    server_proc = EventsProcessor(DummyHandler(), server_conn)
    client_proc = EventsProcessor(DummyHandler(), client_conn)

    client_h2_stream = client_conn.create_stream()
    await client_h2_stream.send_request(
        create_headers(path='/package.Service/Method'),
        _processor=client_proc,
    )
    to_server_transport.process(server_proc)

    server_h2_stream = server_proc.handler.stream

    methods = {'/package.Service/Method': Handler(
        handler,
        Cardinality.UNARY_UNARY,
        DummyRequest,
        DummyReply,
    )}

async def test_ping(config):
    client_h2c, server_h2c = create_connections()

    to_client_transport = TransportStub(client_h2c)
    server_conn = Connection(server_h2c, to_client_transport, config=config)

    to_server_transport = TransportStub(server_h2c)
    client_conn = Connection(client_h2c, to_server_transport, config=config)

    client_processor = EventsProcessor(DummyHandler(), client_conn)
    server_processor = EventsProcessor(DummyHandler(), server_conn)

    client_h2c.ping(b'12345678')
    client_conn.flush()

    to_server_transport.process(server_processor)
    server_conn.flush()

    ping_ack, = to_client_transport.process(client_processor)
    assert isinstance(ping_ack, PingAcknowledged)
    assert ping_ack.ping_data == b'12345678'

async def test_client_reset(
    loop, caplog, handler, level, msg, exc_type, exc_text, config,
):
    caplog.set_level(logging.INFO)
    client_h2c, server_h2c = create_connections()

    to_client_transport = TransportStub(client_h2c)
    to_server_transport = TransportStub(server_h2c)

    client_conn = Connection(client_h2c, to_server_transport, config=config)
    server_conn = Connection(server_h2c, to_client_transport, config=config)

    server_proc = EventsProcessor(DummyHandler(), server_conn)
    client_proc = EventsProcessor(DummyHandler(), client_conn)

    client_h2_stream = client_conn.create_stream()
    await client_h2_stream.send_request(
        create_headers(path='/package.Service/Method'),
        _processor=client_proc,
    )
    to_server_transport.process(server_proc)

    server_h2_stream = server_proc.handler.stream

    methods = {'/package.Service/Method': Handler(
        handler,
        Cardinality.UNARY_UNARY,
        DummyRequest,
        DummyReply,

def connection_made(self, transport: BaseTransport) -> None:
        sock = transport.get_extra_info('socket')
        if sock is not None:
            _set_nodelay(sock)

        h2_conn = H2Connection(config=self.h2_config)
        h2_conn.initiate_connection()

        self.connection = Connection(
            h2_conn,
            cast(Transport, transport),
            config=self.config,
        )
        self.connection.flush()

        self.processor = EventsProcessor(self.handler, self.connection)