(
self,
ws_uri: WebSocketURI,
proxy: Proxy,
user_agent_header: str | None = None,
)
| 724 | |
| 725 | class HTTPProxyConnection(asyncio.Protocol): |
| 726 | def __init__( |
| 727 | self, |
| 728 | ws_uri: WebSocketURI, |
| 729 | proxy: Proxy, |
| 730 | user_agent_header: str | None = None, |
| 731 | ): |
| 732 | self.ws_uri = ws_uri |
| 733 | self.proxy = proxy |
| 734 | self.user_agent_header = user_agent_header |
| 735 | |
| 736 | self.reader = StreamReader() |
| 737 | self.parser = Response.parse( |
| 738 | self.reader.read_line, |
| 739 | self.reader.read_exact, |
| 740 | self.reader.read_to_eof, |
| 741 | proxy=True, |
| 742 | ) |
| 743 | |
| 744 | loop = asyncio.get_running_loop() |
| 745 | self.response: asyncio.Future[Response] = loop.create_future() |
| 746 | |
| 747 | def run_parser(self) -> None: |
| 748 | try: |
nothing calls this directly
no test coverage detected