C///

\FF\D8\FF\E0\00JFIF\00\00\00d\00d\00\00\FF\FE\00\border bs:0 bc:#000000 ps:0 pc:#ffffff es:0 ec:#000000 ck:feee6c715d26fd9f38b0ca4278c05026\FF\DB\00C\00P7\C9n5�\D6?\BD�\9Ds\EBp\9F[`8m\B7)o\B5\E8\E6I\99\FE3]]A2\BA\8Cw\D6E\93\\DEv\C8\009\F2\F1NI?uc\\F5\EA\96k\xN<~buv\EA\C8\D7 \8B\84\CEcxI\BBg\AE\9E=\D6+n\EC\80\C8A\8C\AE\EB\CF\D5\DA\E9"2\A4\B9j5\EB\F3W\B63\96\B30Yu\DA\FC8\ED\DF\E7Ms\FB\F1\8E\B3\FA\EA\E8\E6(\883zs\F2_\8DFk\8Bh \00\8C\DCw\D3R\B5+6X\BA\B2\C4j\AB0\B4\FCMw\C2I\8E\9B\E3\A9~9u\FA\D3l\80\C8%p\EE\FDn2� \00$\FEj\C4e\A9\DB\~\95\A7\A5\80EK\BB\8DDsP\00@@AD'k\CF\E8\DB\D2(\80\9AK\D3\85\D6lb\F2\BA\8C*\80\00)\95 59\A3R:\F3\CE"\B6\80\88\00\00i1u4�\E9\F2�\A6\A2\FACM\93WMb*\E0*\00\00\00\00\00(\A8\80\00\00\80\00\00\00\00\00\00\00\00\00\FF\D9 C/// from __future__ import annotations import abc import asyncio import ssl from typing import Dict, List, Optional, Union from urllib.parse import ParseResult try: import aiohttp import multidict except ImportError: aiohttp = None # type: ignore[assignment] multidict = None # type: ignore[assignment] from nats.errors import ProtocolError class Transport(abc.ABC): @abc.abstractmethod async def connect(self, uri: ParseResult, buffer_size: int, connect_timeout: int): """ Connects to a server using the implemented transport. The uri passed is of type ParseResult that can be obtained calling urllib.parse.urlparse. """ pass @abc.abstractmethod async def connect_tls( self, uri: Union[str, ParseResult], ssl_context: ssl.SSLContext, buffer_size: int, connect_timeout: int, ): """ connect_tls is similar to connect except it tries to connect to a secure endpoint, using the provided ssl context. The uri can be provided as string in case the hostname differs from the uri hostname, in case it was provided as 'tls_hostname' on the options. """ pass @abc.abstractmethod def write(self, payload: bytes): """ Write bytes to underlying transport. Needs a call to drain() to be successfully written. """ pass @abc.abstractmethod def writelines(self, payload: List[bytes]): """ Writes a list of bytes, one by one, to the underlying transport. Needs a call to drain() to be successfully written. """ pass @abc.abstractmethod async def read(self, buffer_size: int) -> bytes: """ Reads a sequence of bytes from the underlying transport, up to buffer_size. The buffer_size is ignored in case the transport carries already frames entire messages (i.e. websocket). """ pass @abc.abstractmethod async def readline(self) -> bytes: """ Reads one whole frame of bytes (or message) from the underlying transport. """ pass @abc.abstractmethod async def drain(self): """ Flushes the bytes queued for transmission when calling write() and writelines(). """ pass @abc.abstractmethod async def wait_closed(self): """ Waits until the connection is successfully closed. """ pass @abc.abstractmethod def close(self): """ Closes the underlying transport. """ pass @abc.abstractmethod def at_eof(self) -> bool: """ Returns if underlying transport is at eof. """ pass @abc.abstractmethod def __bool__(self): """ Returns if the transport was initialized, either by calling connect of connect_tls. """ pass class TcpTransport(Transport): def __init__(self): self._bare_io_reader: Optional[asyncio.StreamReader] = None self._io_reader: Optional[asyncio.StreamReader] = None self._bare_io_writer: Optional[asyncio.StreamWriter] = None self._io_writer: Optional[asyncio.StreamWriter] = None async def connect(self, uri: ParseResult, buffer_size: int, connect_timeout: int): r, w = await asyncio.wait_for( asyncio.open_connection( host=uri.hostname, port=uri.port, limit=buffer_size, ), connect_timeout, ) # We keep a reference to the initial transport we used when # establishing the connection in case we later upgrade to TLS # after getting the first INFO message. This is in order to # prevent the GC closing the socket after we send CONNECT # and replace the transport. # # See https://github.com/nats-io/asyncio-nats/issues/43 self._bare_io_reader = self._io_reader = r self._bare_io_writer = self._io_writer = w async def connect_tls( self, uri: Union[str, ParseResult], ssl_context: ssl.SSLContext, buffer_size: int, connect_timeout: int, ) -> None: assert self._io_writer, f"{type(self).__name__}.connect must be called first" # manually recreate the stream reader/writer with a tls upgraded transport reader = asyncio.StreamReader() protocol = asyncio.StreamReaderProtocol(reader) transport_future = asyncio.get_running_loop().start_tls( self._io_writer.transport, protocol, ssl_context, # hostname here will be passed directly as string server_hostname=uri if isinstance(uri, str) else uri.hostname, ) transport = await asyncio.wait_for(transport_future, connect_timeout) writer = asyncio.StreamWriter(transport, protocol, reader, asyncio.get_running_loop()) self._io_reader, self._io_writer = reader, writer def write(self, payload): return self._io_writer.write(payload) def writelines(self, payload): return self._io_writer.writelines(payload) async def read(self, buffer_size: int): assert self._io_reader, f"{type(self).__name__}.connect must be called first" return await self._io_reader.read(buffer_size) async def readline(self): return await self._io_reader.readline() async def drain(self): return await self._io_writer.drain() async def wait_closed(self): return await self._io_writer.wait_closed() def close(self): return self._io_writer.close() def at_eof(self): return self._io_reader.at_eof() def __bool__(self): return bool(self._io_writer) and bool(self._io_reader) class WebSocketTransport(Transport): def __init__(self, ws_headers: Optional[Dict[str, List[str]]] = None): if not aiohttp: raise ImportError("Could not import aiohttp transport, please install it with `pip install aiohttp`") self._ws: Optional[aiohttp.ClientWebSocketResponse] = None self._client: aiohttp.ClientSession = aiohttp.ClientSession() self._pending = asyncio.Queue() self._close_task = asyncio.Future() self._using_tls: Optional[bool] = None self._ws_headers = ws_headers async def connect(self, uri: ParseResult, buffer_size: int, connect_timeout: int): headers = self._get_custom_headers() # for websocket library, the uri must contain the scheme already self._ws = await self._client.ws_connect(uri.geturl(), timeout=connect_timeout, headers=headers) self._using_tls = False async def connect_tls( self, uri: Union[str, ParseResult], ssl_context: ssl.SSLContext, buffer_size: int, connect_timeout: int, ): if self._ws and not self._ws.closed: if self._using_tls: return raise ProtocolError("ws: cannot upgrade to TLS") headers = self._get_custom_headers() self._ws = await self._client.ws_connect(