Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
transport: asyncio.BaseTransport,
protocol: str,
) -> None:
self.loop = loop
self.config = config
self.transport = transport
self.protocol = protocol
self.closed = False
self._can_write = asyncio.Event(loop=loop)
self._can_write.set()
self.start_keep_alive_timeout()
socket = self.transport.get_extra_info("socket")
self.client = parse_socket_addr(socket.family, socket.getpeername())
self.server = parse_socket_addr(socket.family, socket.getsockname())
self.ssl_info = self.transport.get_extra_info("ssl_object")
def __init__(self, stream: trio.abc.Stream, protocol: str) -> None:
self.stream = stream
self.protocol = protocol
self._is_ssl = isinstance(self.stream, trio.SSLStream)
if self._is_ssl:
socket = self.stream.transport_stream.socket
else:
socket = self.stream.socket
self.client = parse_socket_addr(socket.family, socket.getpeername())
self.server = parse_socket_addr(socket.family, socket.getsockname())
try:
try:
with trio.fail_after(self.config.ssl_handshake_timeout):
await self.stream.do_handshake()
except (trio.BrokenResourceError, trio.TooSlowError):
return # Handshake failed
alpn_protocol = self.stream.selected_alpn_protocol()
socket = self.stream.transport_stream.socket
ssl = True
except AttributeError: # Not SSL
alpn_protocol = "http/1.1"
socket = self.stream.socket
ssl = False
try:
client = parse_socket_addr(socket.family, socket.getpeername())
server = parse_socket_addr(socket.family, socket.getsockname())
async with trio.open_nursery() as nursery:
self.nursery = nursery
self.protocol = ProtocolWrapper(
self.config,
ssl,
client,
server,
self.protocol_send,
partial(spawn_app, nursery, self.app, self.config),
EventWrapper,
alpn_protocol,
)
await self.protocol.initiate()
nursery.start_soon(self.protocol.send_task)
async def run(self) -> None:
socket = self.writer.get_extra_info("socket")
client = parse_socket_addr(socket.family, socket.getpeername())
server = parse_socket_addr(socket.family, socket.getsockname())
ssl_object = self.writer.get_extra_info("ssl_object")
if ssl_object is not None:
ssl = True
alpn_protocol = ssl_object.selected_alpn_protocol()
else:
ssl = False
alpn_protocol = "http/1.1"
self.protocol = ProtocolWrapper(
self.config,
ssl,
client,
server,
self.protocol_send,
partial(spawn_app, self.app, self.loop, self.config),
async def run(self) -> None:
socket = self.writer.get_extra_info("socket")
client = parse_socket_addr(socket.family, socket.getpeername())
server = parse_socket_addr(socket.family, socket.getsockname())
ssl_object = self.writer.get_extra_info("ssl_object")
if ssl_object is not None:
ssl = True
alpn_protocol = ssl_object.selected_alpn_protocol()
else:
ssl = False
alpn_protocol = "http/1.1"
self.protocol = ProtocolWrapper(
self.config,
ssl,
client,
server,
self.protocol_send,
partial(spawn_app, self.app, self.loop, self.config),
EventWrapper,
try:
with trio.fail_after(self.config.ssl_handshake_timeout):
await self.stream.do_handshake()
except (trio.BrokenResourceError, trio.TooSlowError):
return # Handshake failed
alpn_protocol = self.stream.selected_alpn_protocol()
socket = self.stream.transport_stream.socket
ssl = True
except AttributeError: # Not SSL
alpn_protocol = "http/1.1"
socket = self.stream.socket
ssl = False
try:
client = parse_socket_addr(socket.family, socket.getpeername())
server = parse_socket_addr(socket.family, socket.getsockname())
async with trio.open_nursery() as nursery:
self.nursery = nursery
self.protocol = ProtocolWrapper(
self.config,
ssl,
client,
server,
self.protocol_send,
partial(spawn_app, nursery, self.app, self.config),
EventWrapper,
alpn_protocol,
)
await self.protocol.initiate()
nursery.start_soon(self.protocol.send_task)
await self._update_keep_alive_timeout()
def connection_made(self, transport: asyncio.DatagramTransport) -> None: # type: ignore
# h3/Quic is an optional part of Hypercorn
from ..protocol.quic import QuicProtocol # noqa: F811
self.transport = transport
socket = self.transport.get_extra_info("socket")
server = parse_socket_addr(socket.family, socket.getsockname())
self.protocol = QuicProtocol(
self.config,
server,
partial(spawn_app, self.app, self.loop, self.config),
self.protocol_send,
self._call_at,
self.loop.time,
)
config: Config,
transport: asyncio.BaseTransport,
protocol: str,
) -> None:
self.loop = loop
self.config = config
self.transport = transport
self.protocol = protocol
self.closed = False
self._can_write = asyncio.Event(loop=loop)
self._can_write.set()
self.start_keep_alive_timeout()
socket = self.transport.get_extra_info("socket")
self.client = parse_socket_addr(socket.family, socket.getpeername())
self.server = parse_socket_addr(socket.family, socket.getsockname())
self.ssl_info = self.transport.get_extra_info("ssl_object")
def __init__(
self,
app: ASGIFramework,
config: Config,
socket: trio.socket.socket,
nursery: trio._core._run.Nursery,
) -> None:
from ..protocol.quic import QuicProtocol # h3/Quic is an optional part of Hypercorn
self.app = app
self.config = config
self.nursery = nursery
self.socket = trio.socket.from_stdlib_socket(socket)
server = parse_socket_addr(socket.family, socket.getsockname())
self.protocol = QuicProtocol(
config,
server,
partial(spawn_app, self.nursery, self.app, self.config),
self.protocol_send,
self._call_at,
trio.current_time,
)
def __init__(self, stream: trio.abc.Stream, protocol: str) -> None:
self.stream = stream
self.protocol = protocol
self._is_ssl = isinstance(self.stream, trio.SSLStream)
if self._is_ssl:
socket = self.stream.transport_stream.socket
else:
socket = self.stream.socket
self.client = parse_socket_addr(socket.family, socket.getpeername())
self.server = parse_socket_addr(socket.family, socket.getsockname())