---

client_proto.py

import asyncio from contextlib import suppress from typing import Any, Optional, Tuple from .base_protocol import BaseProtocol from .client_exceptions import ( ClientOSError, ClientPayloadError, ServerDisconnectedError, ServerTimeoutError, ) from .helpers import BaseTimerContext from .http import HttpResponseParser, RawResponseMessage from .streams import EMPTY_PAYLOAD, DataQueue, StreamReader class ResponseHandler(BaseProtocol, DataQueue[Tuple[RawResponseMessage, StreamReader]]): """Helper class to adapt between Protocol and StreamReader.""" def __init__(self, loop: asyncio.AbstractEventLoop) -> None: BaseProtocol.__init__(self, loop=loop) DataQueue.__init__(self, loop) self._should_close = False self._payload: Optional[StreamReader] = None self._skip_payload = False self._payload_parser = None self._timer = None self._tail = b"" self._upgraded = False self._parser: Optional[HttpResponseParser] = None self._read_timeout: Optional[float] = None self._read_timeout_handle: Optional[asyncio.TimerHandle] = None @property def upgraded(self) -> bool: return self._upgraded @property def should_close(self) -> bool: if self._payload is not None and not self._payload.is_eof() or self._upgraded: return True return ( self._should_close or self._upgraded or self.exception() is not None or self._payload_parser is not None or len(self) > 0 or bool(self._tail) ) def force_close(self) -> None: self._should_close = True def close(self) -> None: transport = self.transport if transport is not None: transport.close() self.transport = None self._payload = None self._drop_timeout() def is_connected(self) -> bool: return self.transport is not None and not self.transport.is_closing() def connection_lost(self, exc: Optional[BaseException]) -> None: self._drop_timeout() if self._payload_parser is not None: with suppress(Exception): self._payload_parser.feed_eof() uncompleted = None if self._parser is not None: try: uncompleted = self._parser.feed_eof() except Exception: if self._payload is not None: self._payload.set_exception( ClientPayloadError("Response payload is not completed") ) if not self.is_eof(): if isinstance(exc, OSError): exc = ClientOSError(*exc.args) if exc is None: exc = ServerDisconnectedError(uncompleted) # assigns self._should_close to True as side effect, # we do it anyway below self.set_exception(exc) self._should_close = True self._parser = None self._payload = None self._payload_parser = None self._reading_paused = False super().connection_lost(exc) def eof_received(self) -> None: # should call parser.feed_eof() most likely self._drop_timeout() def pause_reading(self) -> None: super().pause_reading() self._drop_timeout() def resume_reading(self) -> None: super().resume_reading() self._reschedule_timeout() def set_exception(self, exc: BaseException) -> None: self._should_close = True self._drop_timeout() super().set_exception(exc) def set_parser(self, parser: Any, payload: Any) -> None: # TODO: actual types are: # parser: WebSocketReader # payload: FlowControlDataQueue # but they are not generi enough # Need an ABC for both types self._payload = payload self._payload_parser = parser self._drop_timeout() if self._tail: data, self._tail = self._tail, b"" self.data_received(data) def set_response_params( self, *, timer: Optional[BaseTimerContext] = None, skip_payload: bool = False, read_until_eof: bool = False, auto_decompress: bool = True, read_timeout: Optional[float] = None, read_bufsize: int = 2**16, ) -> None: self._skip_payload = skip_payload self._read_timeout = read_timeout self._reschedule_timeout() self._parser = HttpResponseParser( self, self._loop, read_bufsize, timer=timer, payload_exception=ClientPayloadError, response_with_body=not skip_payload, read_until_eof=read_until_eof, auto_decompress=auto_decompress, ) if self._tail: data, self._tail = self._tail, b"" self.data_received(data) def _drop_timeout(self) -> None: if self._read_timeout_handle is not None: self._read_timeout_handle.cancel() self._read_timeout_handle = None def _reschedule_timeout(self) -> None: timeout = self._read_timeout if self._read_timeout_handle is not None: self._read_timeout_handle.cancel() if timeout: self._read_timeout_handle = self._loop.call_later( timeout, self._on_read_timeout ) else: self._read_timeout_handle = None def _on_read_timeout(self) -> None: exc = ServerTimeoutError("Timeout on reading data from socket") self.set_exception(exc) if self._payload is not None: self._payload.set_exception(exc) def data_received(self, data: bytes) -> None: self._reschedule_timeout() if not data: return # custom payload parser if self._payload_parser is not None: eof, tail = self._payload_parser.feed_data(data) if eof: self._payload = None self._payload_parser = None if tail: self.data_received(tail) return else: if self._upgraded or self._parser is None: # i.e. websocket connection, websocket parser is not set yet self._tail += data else: # parse http messages try: messages, upgraded, tail = self._parser.feed_data(data) except BaseException as exc: if self.transport is not None: # connection.release() could be called BEFORE # data_received(), the transport is already # closed in this case self.transport.close() # should_close is True after the call self.set_exception(exc) return self._upgraded = upgraded payload: Optional[StreamReader] = None for message, payload in messages: if message.should_close: self._should_close = True self._payload = payload if self._skip_payload or message.code in (204, 304): self.feed_data((message, EMPTY_PAYLOAD), 0) else: self.feed_data((message, payload), 0) if payload is not None: # new message(s) was processed # register timeout handler unsubscribing # either on end-of-stream or immediately for # EMPTY_PAYLOAD if payload is not EMPTY_PAYLOAD: payload.on_eof(self._drop_timeout) else: self._drop_timeout() if tail: if upgraded: self.data_received(tail) else: self._tail = tail

---

docroot: /home/induvedam/public_html

---

curdir: /opt/imunify360/venv/lib/python3.11/site-packages/aiohttp

	.		4096
	..		8192
	.hash		4096
X	__init__.py		6870
	__pycache__		4096
X	_cparser.pxd		4998
X	_find_header.pxd		68
X	_headers.pxi		2007
X	_helpers.cpython-311-x86_64-linux-gnu.so		61648
X	_helpers.pyi		202
X	_helpers.pyx		1049
X	_http_parser.cpython-311-x86_64-linux-gnu.so		501200
X	_http_parser.pyx		27336
X	_http_writer.cpython-311-x86_64-linux-gnu.so		62688
X	_http_writer.pyx		4575
X	_websocket.cpython-311-x86_64-linux-gnu.so		35232
X	_websocket.pyx		1561
X	abc.py		5505
X	base_protocol.py		2741
X	client.py		45037
X	client_exceptions.py		9270
X	client_proto.py		8170
X	client_reqrep.py		36973
X	client_ws.py		10516
X	connector.py		51177
X	cookiejar.py		13655
X	formdata.py		6106
X	hdrs.py		4724
X	helpers.py		26398
X	http.py		1800
X	http_exceptions.py		2586
X	http_parser.py		33092
X	http_websocket.py		25299
X	http_writer.py		5933
X	locks.py		1136
X	log.py		325
X	multipart.py		32313
X	payload.py		13634
X	payload_streamer.py		2112
X	py.typed		7
X	pytest_plugin.py		11772
X	resolver.py		5092
X	streams.py		20758
X	tcp_helpers.py		961
X	test_utils.py		21434
X	tracing.py		15177
X	typedefs.py		1766
X	web.py		18081
X	web_app.py		17170
X	web_exceptions.py		10098
X	web_fileresponse.py		10784
X	web_log.py		7557
X	web_middlewares.py		4137
X	web_protocol.py		22399
X	web_request.py		28187
X	web_response.py		27471
X	web_routedef.py		6152
X	web_runner.py		11157
X	web_server.py		2050
X	web_urldispatcher.py		39483
X	web_ws.py		17144
X	worker.py		8763