(self, loop, app_protocol, sslcontext, waiter,
server_side=False, server_hostname=None,
call_connection_made=True,
ssl_handshake_timeout=None,
ssl_shutdown_timeout=None)
| 268 | _shutdown_timeout_handle = None |
| 269 | |
| 270 | def __init__(self, loop, app_protocol, sslcontext, waiter, |
| 271 | server_side=False, server_hostname=None, |
| 272 | call_connection_made=True, |
| 273 | ssl_handshake_timeout=None, |
| 274 | ssl_shutdown_timeout=None): |
| 275 | if ssl is None: |
| 276 | raise RuntimeError("stdlib ssl module not available") |
| 277 | |
| 278 | self._ssl_buffer = bytearray(self.max_size) |
| 279 | self._ssl_buffer_view = memoryview(self._ssl_buffer) |
| 280 | |
| 281 | if ssl_handshake_timeout is None: |
| 282 | ssl_handshake_timeout = constants.SSL_HANDSHAKE_TIMEOUT |
| 283 | elif ssl_handshake_timeout <= 0: |
| 284 | raise ValueError( |
| 285 | f"ssl_handshake_timeout should be a positive number, " |
| 286 | f"got {ssl_handshake_timeout}") |
| 287 | if ssl_shutdown_timeout is None: |
| 288 | ssl_shutdown_timeout = constants.SSL_SHUTDOWN_TIMEOUT |
| 289 | elif ssl_shutdown_timeout <= 0: |
| 290 | raise ValueError( |
| 291 | f"ssl_shutdown_timeout should be a positive number, " |
| 292 | f"got {ssl_shutdown_timeout}") |
| 293 | |
| 294 | if not sslcontext: |
| 295 | sslcontext = _create_transport_context( |
| 296 | server_side, server_hostname) |
| 297 | |
| 298 | self._server_side = server_side |
| 299 | if server_hostname and not server_side: |
| 300 | self._server_hostname = server_hostname |
| 301 | else: |
| 302 | self._server_hostname = None |
| 303 | self._sslcontext = sslcontext |
| 304 | # SSL-specific extra info. More info are set when the handshake |
| 305 | # completes. |
| 306 | self._extra = dict(sslcontext=sslcontext) |
| 307 | |
| 308 | # App data write buffering |
| 309 | self._write_backlog = collections.deque() |
| 310 | self._write_buffer_size = 0 |
| 311 | |
| 312 | self._waiter = waiter |
| 313 | self._loop = loop |
| 314 | self._set_app_protocol(app_protocol) |
| 315 | self._app_transport = None |
| 316 | self._app_transport_created = False |
| 317 | # transport, ex: SelectorSocketTransport |
| 318 | self._transport = None |
| 319 | self._ssl_handshake_timeout = ssl_handshake_timeout |
| 320 | self._ssl_shutdown_timeout = ssl_shutdown_timeout |
| 321 | # SSL and state machine |
| 322 | self._incoming = ssl.MemoryBIO() |
| 323 | self._outgoing = ssl.MemoryBIO() |
| 324 | self._state = SSLProtocolState.UNWRAPPED |
| 325 | self._conn_lost = 0 # Set when connection_lost called |
| 326 | if call_connection_made: |
| 327 | self._app_state = AppProtocolState.STATE_INIT |
nothing calls this directly
no test coverage detected