Upgrade an existing stream-based connection to TLS.
(self, sslcontext, *,
server_hostname=None,
ssl_handshake_timeout=None,
ssl_shutdown_timeout=None)
| 386 | await self._protocol._drain_helper() |
| 387 | |
| 388 | async def start_tls(self, sslcontext, *, |
| 389 | server_hostname=None, |
| 390 | ssl_handshake_timeout=None, |
| 391 | ssl_shutdown_timeout=None): |
| 392 | """Upgrade an existing stream-based connection to TLS.""" |
| 393 | server_side = self._protocol._client_connected_cb is not None |
| 394 | protocol = self._protocol |
| 395 | await self.drain() |
| 396 | new_transport = await self._loop.start_tls( # type: ignore |
| 397 | self._transport, protocol, sslcontext, |
| 398 | server_side=server_side, server_hostname=server_hostname, |
| 399 | ssl_handshake_timeout=ssl_handshake_timeout, |
| 400 | ssl_shutdown_timeout=ssl_shutdown_timeout) |
| 401 | self._transport = new_transport |
| 402 | protocol._replace_transport(new_transport) |
| 403 | |
| 404 | def __del__(self, warnings=warnings): |
| 405 | if not self._transport.is_closing(): |
nothing calls this directly
no test coverage detected