Method __init__

mitmproxy/proxy/server.py:466–485

(
        self,
        reader: asyncio.StreamReader | mitmproxy_rs.Stream,
        writer: asyncio.StreamWriter | mitmproxy_rs.Stream,
        options: moptions.Options,
        mode: mode_specs.ProxyMode,
    )

class LiveConnectionHandler(ConnectionHandler, metaclass=abc.ABCMeta):
    def __init__(
        self,
        reader: asyncio.StreamReader | mitmproxy_rs.Stream,
        writer: asyncio.StreamWriter | mitmproxy_rs.Stream,
        options: moptions.Options,
        mode: mode_specs.ProxyMode,
    ) -> None:
        client = Client(
            transport_protocol=writer.get_extra_info("transport_protocol", "tcp"),
            peername=writer.get_extra_info("peername"),
            sockname=writer.get_extra_info("sockname"),
            timestamp_start=time.time(),
            proxy_mode=mode,
            state=ConnectionState.OPEN,
        )
        context = Context(client, options)
        super().__init__(context)
        self.transports[client] = ConnectionIO(
            handler=None, reader=reader, writer=writer
        )


class SimpleConnectionHandler(LiveConnectionHandler):  # pragma: no cover