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

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,

async def test_exit_and_connection_was_broken(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)

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)

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)

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,

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

    transport = TransportStub(server_h2c)
    conn = Connection(client_h2c, transport, config=config)
    stream = conn.create_stream()

    processor = EventsProcessor(DummyHandler(), conn)

    await stream.send_request(create_headers(), _processor=processor)
    await stream.send_data(b'0' * (client_h2c.max_outbound_frame_size + 1))

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_unread_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)
    await client_stream.send_data(b'x' * data_size)
    assert client_h2c.outbound_flow_control_window == initial_window - data_size
    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)

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)