(
self, key: ConnectionKeyT, uri: URI, endpoint: HostnameEndpoint
)
| 68 | return self._new_connection(key, uri, endpoint) |
| 69 | |
| 70 | def _new_connection( |
| 71 | self, key: ConnectionKeyT, uri: URI, endpoint: HostnameEndpoint |
| 72 | ) -> Deferred[H2ClientProtocol]: |
| 73 | self._pending_requests[key] = deque() |
| 74 | |
| 75 | conn_lost_deferred: Deferred[list[BaseException]] = Deferred() |
| 76 | conn_lost_deferred.addCallback(self._remove_connection, key) |
| 77 | |
| 78 | factory = H2ClientFactory( |
| 79 | uri, |
| 80 | self.settings, |
| 81 | conn_lost_deferred, |
| 82 | tls_verbose_logging=self._tls_verbose_logging, |
| 83 | ) |
| 84 | conn_d = endpoint.connect(factory) |
| 85 | conn_d.addCallback(self.put_connection, key) |
| 86 | |
| 87 | d: Deferred[H2ClientProtocol] = Deferred() |
| 88 | self._pending_requests[key].append(d) |
| 89 | return d |
| 90 | |
| 91 | def put_connection( |
| 92 | self, conn: H2ClientProtocol, key: ConnectionKeyT |
no test coverage detected