mercurial-scm/hg: comparison mercurial/wireprotoframing.py

equal deleted inserted replaced

-:1a612f9ec2c4
+:068398a8c9cb
 from .utils import (
 cborutil,
 stringutil,
 )
+if typing.TYPE_CHECKING:
+from typing import (
+Iterator,
+)
+HandleSendFramesReturnT = tuple[bytes, dict[bytes, Iterator[bytearray]]]
 FRAME_HEADER_SIZE = 8
 DEFAULT_MAX_FRAME_SIZE = 32768
 STREAM_FLAG_BEGIN_STREAM = 0x01
 STREAM_FLAG_END_STREAM = 0x02
 STREAM_FLAG_ENCODING_APPLIED = 0x04
-STREAM_FLAGS = {
+STREAM_FLAGS: dict[bytes, int] = {
 b'stream-begin': STREAM_FLAG_BEGIN_STREAM,
 b'stream-end': STREAM_FLAG_END_STREAM,
 b'encoded': STREAM_FLAG_ENCODING_APPLIED,
 }
 FRAME_TYPE_TEXT_OUTPUT = 0x06
 FRAME_TYPE_PROGRESS = 0x07
 FRAME_TYPE_SENDER_PROTOCOL_SETTINGS = 0x08
 FRAME_TYPE_STREAM_SETTINGS = 0x09
-FRAME_TYPES = {
+FRAME_TYPES: dict[bytes, int] = {
 b'command-request': FRAME_TYPE_COMMAND_REQUEST,
 b'command-data': FRAME_TYPE_COMMAND_DATA,
 b'command-response': FRAME_TYPE_COMMAND_RESPONSE,
 b'error-response': FRAME_TYPE_ERROR_RESPONSE,
 b'text-output': FRAME_TYPE_TEXT_OUTPUT,
 FLAG_COMMAND_REQUEST_NEW = 0x01
 FLAG_COMMAND_REQUEST_CONTINUATION = 0x02
 FLAG_COMMAND_REQUEST_MORE_FRAMES = 0x04
 FLAG_COMMAND_REQUEST_EXPECT_DATA = 0x08
-FLAGS_COMMAND_REQUEST = {
+FLAGS_COMMAND_REQUEST: dict[bytes, int] = {
 b'new': FLAG_COMMAND_REQUEST_NEW,
 b'continuation': FLAG_COMMAND_REQUEST_CONTINUATION,
 b'more': FLAG_COMMAND_REQUEST_MORE_FRAMES,
 b'have-data': FLAG_COMMAND_REQUEST_EXPECT_DATA,
 }
 FLAG_COMMAND_DATA_CONTINUATION = 0x01
 FLAG_COMMAND_DATA_EOS = 0x02
-FLAGS_COMMAND_DATA = {
+FLAGS_COMMAND_DATA: dict[bytes, int] = {
 b'continuation': FLAG_COMMAND_DATA_CONTINUATION,
 b'eos': FLAG_COMMAND_DATA_EOS,
 }
 FLAG_COMMAND_RESPONSE_CONTINUATION = 0x01
 FLAG_COMMAND_RESPONSE_EOS = 0x02
-FLAGS_COMMAND_RESPONSE = {
+FLAGS_COMMAND_RESPONSE: dict[bytes, int] = {
 b'continuation': FLAG_COMMAND_RESPONSE_CONTINUATION,
 b'eos': FLAG_COMMAND_RESPONSE_EOS,
 }
 FLAG_SENDER_PROTOCOL_SETTINGS_CONTINUATION = 0x01
 FLAG_SENDER_PROTOCOL_SETTINGS_EOS = 0x02
-FLAGS_SENDER_PROTOCOL_SETTINGS = {
+FLAGS_SENDER_PROTOCOL_SETTINGS: dict[bytes, int] = {
 b'continuation': FLAG_SENDER_PROTOCOL_SETTINGS_CONTINUATION,
 b'eos': FLAG_SENDER_PROTOCOL_SETTINGS_EOS,
 }
 FLAG_STREAM_ENCODING_SETTINGS_CONTINUATION = 0x01
 FLAG_STREAM_ENCODING_SETTINGS_EOS = 0x02
-FLAGS_STREAM_ENCODING_SETTINGS = {
+FLAGS_STREAM_ENCODING_SETTINGS: dict[bytes, int] = {
 b'continuation': FLAG_STREAM_ENCODING_SETTINGS_CONTINUATION,
 b'eos': FLAG_STREAM_ENCODING_SETTINGS_EOS,
 }
 # Maps frame types to their available flags.
-FRAME_TYPE_FLAGS = {
+FRAME_TYPE_FLAGS: dict[int, dict[bytes, int]] = {
 FRAME_TYPE_COMMAND_REQUEST: FLAGS_COMMAND_REQUEST,
 FRAME_TYPE_COMMAND_DATA: FLAGS_COMMAND_DATA,
 FRAME_TYPE_COMMAND_RESPONSE: FLAGS_COMMAND_RESPONSE,
 FRAME_TYPE_ERROR_RESPONSE: {},
 FRAME_TYPE_TEXT_OUTPUT: {},
 }
 ARGUMENT_RECORD_HEADER = struct.Struct('<HH')
-def humanflags(mapping, value):
+def humanflags(mapping: dict[bytes, int], value: int) -> bytes:
 """Convert a numeric flags value to a human value, using a mapping table."""
 namemap = {v: k for k, v in mapping.items()}
 flags = []
 val = 1
 while value >= val:
 @attr.s(slots=True)
 class frameheader:
 """Represents the data in a frame header."""
-length = attr.ib()
+length = attr.ib(type=int)
-requestid = attr.ib()
+requestid = attr.ib(type=int)
-streamid = attr.ib()
+streamid = attr.ib(type=int)
-streamflags = attr.ib()
+streamflags = attr.ib(type=int)
-typeid = attr.ib()
+typeid = attr.ib(type=int)
-flags = attr.ib()
+flags = attr.ib(type=int)
 @attr.s(slots=True, repr=False)
 class frame:
 """Represents a parsed frame."""
-requestid = attr.ib()
+requestid = attr.ib(type=int)
-streamid = attr.ib()
+streamid = attr.ib(type=int)
-streamflags = attr.ib()
+streamflags = attr.ib(type=int)
-typeid = attr.ib()
+typeid = attr.ib(type=int)
-flags = attr.ib()
+flags = attr.ib(type=int)
-payload = attr.ib()
+payload = attr.ib(type=bytes)
 @encoding.strmethod
 def __repr__(self):
 typename = b'<unknown 0x%02x>' % self.typeid
 for name, value in FRAME_TYPES.items():
 humanflags(FRAME_TYPE_FLAGS.get(self.typeid, {}), self.flags),
 )
 )
-def makeframe(requestid, streamid, streamflags, typeid, flags, payload):
+def makeframe(
+requestid: int,
+streamid: int,
+streamflags: int,
+typeid: int,
+flags: int,
+payload: bytes,
+) -> bytearray:
 """Assemble a frame into a byte array."""
 # TODO assert size of payload.
 frame = bytearray(FRAME_HEADER_SIZE + len(payload))
 # 24 bits length
 frame[8:] = payload
 return frame
-def makeframefromhumanstring(s):
+def makeframefromhumanstring(s: bytes) -> bytearray:
 """Create a frame from a human readable string
 Strings have the form:
 <request-id> <stream-id> <stream-flags> <type> <flags> <payload>
 flags=finalflags,
 payload=payload,
 )
-def parseheader(data):
+def parseheader(data: bytes) -> frameheader:
 """Parse a unified framing protocol frame header from a buffer.
 The header is expected to be in the buffer at offset 0 and the
 buffer is expected to be large enough to hold a full header.
 """
 return frameheader(
 framelength, requestid, streamid, streamflags, frametype, frameflags
 )
-def readframe(fh):
+def readframe(fh) -> frame | None:
 """Read a unified framing protocol frame from a file object.
 Returns a 3-tuple of (type, flags, payload) for the decoded frame or
 None if no frame is available. May raise if a malformed frame is
 seen.
 h.requestid, h.streamid, h.streamflags, h.typeid, h.flags, payload
 )
 def createcommandframes(
-stream,
+stream: stream,
-requestid,
+requestid: int,
 cmd,
 args,
 datafh=None,
-maxframesize=DEFAULT_MAX_FRAME_SIZE,
+maxframesize: int = DEFAULT_MAX_FRAME_SIZE,
 redirect=None,
-):
+) -> Iterator[bytearray]:
 """Create frames necessary to transmit a request to run a command.
 This is a generator of bytearrays. Each item represents a frame
 ready to be sent over the wire to a peer.
 """
 if done:
 break
-def createcommandresponseokframe(stream, requestid):
+def createcommandresponseokframe(
+stream: outputstream, requestid: int
+) -> bytearray | None:
 overall = b''.join(cborutil.streamencode({b'status': b'ok'}))
 if stream.streamsettingssent:
 overall = stream.encode(overall)
 encoded = True
 encoded=encoded,
 )
 def createcommandresponseeosframes(
-stream, requestid, maxframesize=DEFAULT_MAX_FRAME_SIZE
+stream: outputstream,
-):
+requestid: int,
+maxframesize: int = DEFAULT_MAX_FRAME_SIZE,
+) -> Iterator[bytearray]:
 """Create an empty payload frame representing command end-of-stream."""
 payload = stream.flush()
 offset = 0
 while True:
 if done:
 break
-def createalternatelocationresponseframe(stream, requestid, location):
+def createalternatelocationresponseframe(
+stream: outputstream, requestid: int, location
+) -> bytearray:
 data = {
 b'status': b'redirect',
 b'location': {
 b'url': location.url,
 b'mediatype': location.mediatype,
 payload=payload,
 encoded=encoded,
 )
-def createcommanderrorresponse(stream, requestid, message, args=None):
+def createcommanderrorresponse(
+stream: stream, requestid: int, message: bytes, args=None
+) -> Iterator[bytearray]:
 # TODO should this be using a list of {'msg': ..., 'args': {}} so atom
 # formatting works consistently?
 m = {
 b'status': b'error',
 b'error': {
 flags=FLAG_COMMAND_RESPONSE_EOS,
 payload=overall,
 )
-def createerrorframe(stream, requestid, msg, errtype):
+def createerrorframe(
+stream: stream, requestid: int, msg: bytes, errtype: bytes
+) -> Iterator[bytearray]:
 # TODO properly handle frame size limits.
 assert len(msg) <= DEFAULT_MAX_FRAME_SIZE
 payload = b''.join(
 cborutil.streamencode(
 payload=payload,
 )
 def createtextoutputframe(
-stream, requestid, atoms, maxframesize=DEFAULT_MAX_FRAME_SIZE
+stream: stream,
-):
+requestid: int,
+atoms,
+maxframesize: int = DEFAULT_MAX_FRAME_SIZE,
+) -> Iterator[bytearray]:
 """Create a text output frame to render text to people.
 ``atoms`` is a 3-tuple of (formatting string, args, labels).
 The formatting string contains ``%s`` tokens to be replaced by the
 TODO we'll need something like this when compression is supported.
 So it might make sense to implement this functionality at the stream
 level.
 """
-def __init__(self, stream, requestid, maxframesize=DEFAULT_MAX_FRAME_SIZE):
+def __init__(
+self, stream, requestid: int, maxframesize: int = DEFAULT_MAX_FRAME_SIZE
+) -> None:
 self._stream = stream
 self._requestid = requestid
 self._maxsize = maxframesize
 self._chunks = []
 self._chunkssize = 0
 # this module.
 STREAM_ENCODERS: dict[bytes, tuple[EncoderT, DecoderT]] = {}
 STREAM_ENCODERS_ORDER: list[bytes] = []
-def populatestreamencoders():
+def populatestreamencoders() -> None:
 if STREAM_ENCODERS:
 return
 try:
 from . import zstd  # pytype: disable=import-error
 class stream:
 """Represents a logical unidirectional series of frames."""
-def __init__(self, streamid, active=False):
+streamid: int
+_active: bool
+def __init__(self, streamid: int, active: bool = False) -> None:
 self.streamid = streamid
 self._active = active
-def makeframe(self, requestid, typeid, flags, payload):
+def makeframe(
+self, requestid: int, typeid: int, flags: int, payload: bytes
+) -> bytearray:
 """Create a frame to be sent out over this stream.
 Only returns the frame instance. Does not actually send it.
 """
 streamflags = 0
 class inputstream(stream):
 """Represents a stream used for receiving data."""
-def __init__(self, streamid, active=False):
+_decoder: Decoder | None
+def __init__(self, streamid: int, active: bool = False) -> None:
 super().__init__(streamid, active=active)
 self._decoder = None
-def setdecoder(self, ui, name, extraobjs):
+def setdecoder(self, ui, name: bytes, extraobjs) -> None:
 """Set the decoder for this stream.
 Receives the stream profile name and any additional CBOR objects
 decoded from the stream encoding settings frame payloads.
 """
 if name not in STREAM_ENCODERS:
 raise error.Abort(_(b'unknown stream decoder: %s') % name)
 self._decoder = STREAM_ENCODERS[name][1](ui, extraobjs)
-def decode(self, data):
+def decode(self, data) -> bytes:
 # Default is identity decoder. We don't bother instantiating one
 # because it is trivial.
 if not self._decoder:
 return data
 return self._decoder.decode(data)
-def flush(self):
+def flush(self) -> bytes:
 if not self._decoder:
 return b''
+# TODO: this looks like a bug- no decoder class defines flush(), so
+#  either no decoders are used, or no inputstream is flushed.
 return self._decoder.flush()
 class outputstream(stream):
 """Represents a stream used for sending data."""
-def __init__(self, streamid, active=False):
+streamsettingssent: bool
+_encoder: Encoder | None
+_encodername: bytes | None
+def __init__(self, streamid: int, active: bool = False) -> None:
 super().__init__(streamid, active=active)
 self.streamsettingssent = False
 self._encoder = None
 self._encodername = None
-def setencoder(self, ui, name):
+def setencoder(self, ui, name: bytes) -> None:
 """Set the encoder for this stream.
 Receives the stream profile name.
 """
 if name not in STREAM_ENCODERS:
 raise error.Abort(_(b'unknown stream encoder: %s') % name)
 self._encoder = STREAM_ENCODERS[name][0](ui)
 self._encodername = name
-def encode(self, data):
+def encode(self, data) -> bytes:
 if not self._encoder:
 return data
 return self._encoder.encode(data)
-def flush(self):
+def flush(self) -> bytes:
 if not self._encoder:
 return b''
 return self._encoder.flush()
-def finish(self):
+# TODO: was this supposed to return the result of finish()?
+def finish(self):  # -> bytes:
 if not self._encoder:
 return b''
 self._encoder.finish()
-def makeframe(self, requestid, typeid, flags, payload, encoded=False):
+def makeframe(
+self,
+requestid: int,
+typeid: int,
+flags: int,
+payload: bytes,
+encoded: bool = False,
+) -> bytearray:
 """Create a frame to be sent out over this stream.
 Only returns the frame instance. Does not actually send it.
 """
 streamflags = 0
 return makeframe(
 requestid, self.streamid, streamflags, typeid, flags, payload
 )
-def makestreamsettingsframe(self, requestid):
+def makestreamsettingsframe(self, requestid: int) -> bytearray | None:
 """Create a stream settings frame for this stream.
 Returns frame data or None if no stream settings frame is needed or has
 already been sent.
 """
 FLAG_STREAM_ENCODING_SETTINGS_EOS,
 payload,
 )
-def ensureserverstream(stream):
+def ensureserverstream(stream: stream) -> None:
 if stream.streamid % 2:
 raise error.ProgrammingError(
 b'server should only write to even '
 b'numbered streams; %d is not even' % stream.streamid
 )
-DEFAULT_PROTOCOL_SETTINGS = {
+DEFAULT_PROTOCOL_SETTINGS: dict[bytes, list[bytes]] = {
 b'contentencodings': [b'identity'],
 }
 class serverreactor:
 that if we receive a command and instruct its dispatch, another command
 with its request ID can come in over the wire and there will be a race
 between who responds to what.
 """
-def __init__(self, ui, deferoutput=False):
+_bufferedframegens: list[Iterator[bytearray]]
+def __init__(self, ui, deferoutput: bool = False) -> None:
 """Construct a new server reactor.
 ``deferoutput`` can be used to indicate that no output frames should be
 instructed to be sent until input has been exhausted. In this mode,
 events that would normally generate output frames (such as a command
 # Sender protocol settings are optional. Set implied default values.
 self._sendersettings = dict(DEFAULT_PROTOCOL_SETTINGS)
 populatestreamencoders()
-def onframerecv(self, frame):
+def onframerecv(self, frame: frame):
 """Process a frame that has been received off the wire.
 Returns a dict with an ``action`` key that details what action,
 if any, the consumer should take next.
 """
 if not meth:
 raise error.ProgrammingError(b'unhandled state: %s' % self._state)
 return meth(frame)
-def oncommandresponsereadyobjects(self, stream, requestid, objs):
+def oncommandresponsereadyobjects(
+self, stream, requestid: int, objs
+) -> HandleSendFramesReturnT:
 """Signal that objects are ready to be sent to the client.
 ``objs`` is an iterable of objects (typically a generator) that will
 be encoded via CBOR and added to frames, which will be sent to the
 client.
 self._activecommands.remove(requestid)
 return self._handlesendframes(sendframes())
-def oninputeof(self):
+def oninputeof(self) -> tuple[bytes, dict[bytes, Iterator[bytearray]]]:
 """Signals that end of input has been received.
 No more frames will be received. All pending activity should be
 completed.
 """
 return b'sendframes', {
 b'framegen': makegen(),
 }
-def _handlesendframes(self, framegen):
+def _handlesendframes(
+self, framegen: Iterator[bytearray]
+) -> HandleSendFramesReturnT:
 if self._deferoutput:
 self._bufferedframegens.append(framegen)
 return b'noop', {}
 else:
 return b'sendframes', {
 b'framegen': framegen,
 }
-def onservererror(self, stream, requestid, msg):
+def onservererror(
+self, stream: stream, requestid: int, msg: bytes
+) -> HandleSendFramesReturnT:
 ensureserverstream(stream)
 def sendframes():
 yield from createerrorframe(
 stream, requestid, msg, errtype=b'server'
 self._activecommands.remove(requestid)
 return self._handlesendframes(sendframes())
-def oncommanderror(self, stream, requestid, message, args=None):
+def oncommanderror(
+self, stream: stream, requestid: int, message: bytes, args=None
+) -> HandleSendFramesReturnT:
 """Called when a command encountered an error before sending output."""
 ensureserverstream(stream)
 def sendframes():
 yield from createcommanderrorresponse(
 self._activecommands.remove(requestid)
 return self._handlesendframes(sendframes())
-def makeoutputstream(self):
+def makeoutputstream(self) -> outputstream:
 """Create a stream to be used for sending data to the client.
 If this is called before protocol settings frames are received, we
 don't know what stream encodings are supported by the client and
 we will default to identity.
 s.setencoder(self._ui, name)
 break
 return s
-def _makeerrorresult(self, msg):
+def _makeerrorresult(self, msg: bytes) -> tuple[bytes, dict[bytes, bytes]]:
 return b'error', {
 b'message': msg,
 }
-def _makeruncommandresult(self, requestid):
+def _makeruncommandresult(self, requestid: int):
 entry = self._receivingcommands[requestid]
 if not entry[b'requestdone']:
 self._state = b'errored'
 raise error.ProgrammingError(
 b'redirect': request.get(b'redirect'),
 b'data': entry[b'data'].getvalue() if entry[b'data'] else None,
 },
 )
-def _makewantframeresult(self):
+def _makewantframeresult(self) -> tuple[bytes, dict[bytes, bytes]]:
 return b'wantframe', {
 b'state': self._state,
 }
-def _validatecommandrequestframe(self, frame):
+def _validatecommandrequestframe(self, frame: frame):
 new = frame.flags & FLAG_COMMAND_REQUEST_NEW
 continuation = frame.flags & FLAG_COMMAND_REQUEST_CONTINUATION
 if new and continuation:
 self._state = b'errored'
 b'received command request frame with neither new nor '
 b'continuation flags set'
 )
 )
-def _onframeinitial(self, frame):
+def _onframeinitial(self, frame: frame):
 # Called when we receive a frame when in the "initial" state.
 if frame.typeid == FRAME_TYPE_SENDER_PROTOCOL_SETTINGS:
 self._state = b'protocol-settings-receiving'
 self._protocolsettingsdecoder = cborutil.bufferingdecoder()
 return self._onframeprotocolsettings(frame)
 b'frame; got %d'
 )
 % frame.typeid
 )
-def _onframeprotocolsettings(self, frame):
+def _onframeprotocolsettings(self, frame: frame):
 assert self._state == b'protocol-settings-receiving'
 assert self._protocolsettingsdecoder is not None
 if frame.typeid != FRAME_TYPE_SENDER_PROTOCOL_SETTINGS:
 self._state = b'errored'
 self._state = b'idle'
 return self._makewantframeresult()
-def _onframeidle(self, frame):
+def _onframeidle(self, frame: frame):
 # The only frame type that should be received in this state is a
 # command request.
 if frame.typeid != FRAME_TYPE_COMMAND_REQUEST:
 self._state = b'errored'
 return self._makeerrorresult(
 assert moreframes or expectingdata
 self._state = b'command-receiving'
 return self._makewantframeresult()
-def _onframecommandreceiving(self, frame):
+def _onframecommandreceiving(self, frame: frame):
 if frame.typeid == FRAME_TYPE_COMMAND_REQUEST:
 # Process new command requests as such.
 if frame.flags & FLAG_COMMAND_REQUEST_NEW:
 return self._onframeidle(frame)
 self._state = b'errored'
 return self._makeerrorresult(
 _(b'received unexpected frame type: %d') % frame.typeid
 )
-def _handlecommanddataframe(self, frame, entry):
+def _handlecommanddataframe(self, frame: frame, entry):
 assert frame.typeid == FRAME_TYPE_COMMAND_DATA
 # TODO support streaming data instead of buffering it.
 entry[b'data'].write(frame.payload)
 return self._makeruncommandresult(frame.requestid)
 else:
 self._state = b'errored'
 return self._makeerrorresult(_(b'command data frame without flags'))
-def _onframeerrored(self, frame):
+def _onframeerrored(self, frame: frame):
 return self._makeerrorresult(_(b'server already errored'))
 class commandrequest:
 """Represents a request to run a command."""
-def __init__(self, requestid, name, args, datafh=None, redirect=None):
+state: bytes
+def __init__(
+self, requestid: int, name, args, datafh=None, redirect=None
+) -> None:
 self.requestid = requestid
 self.name = name
 self.args = args
 self.datafh = datafh
 self.redirect = redirect
 ``expectmore`` and ``eos`` evaluate to True when more response data
 is expected to follow or we're at the end of the response stream,
 respectively.
 """
+_hasmultiplesend: bool
+_buffersends: bool
 def __init__(
 self,
 ui,
-hasmultiplesend=False,
+hasmultiplesend: bool = False,
-buffersends=True,
+buffersends: bool = True,
 clientcontentencoders=None,
-):
+) -> None:
 """Create a new instance.
 ``hasmultiplesend`` indicates whether multiple sends are supported
 by the transport. When True, it is possible to send commands immediately
 instead of buffering until the caller signals an intent to finish a
 {
 b'framegen': self._makecommandframes(request),
 },
 )
-def flushcommands(self):
+def flushcommands(self) -> tuple[bytes, dict[bytes, Iterator[bytearray]]]:
 """Request that all queued commands be sent.
 If any commands are buffered, this will instruct the caller to send
 them over the wire. If no commands are buffered it instructs the client
 to no-op.
 return b'sendframes', {
 b'framegen': makeframes(),
 }
-def _makecommandframes(self, request):
+def _makecommandframes(
+self, request: commandrequest
+) -> Iterator[bytearray]:
 """Emit frames to issue a command request.
 As a side-effect, update request accounting to reflect its changed
 state.
 """
 yield from res
 request.state = b'sent'
-def onframerecv(self, frame):
+def onframerecv(self, frame: frame):
 """Process a frame that has been received off the wire.
 Returns a 2-tuple of (action, meta) describing further action the
 caller needs to take as a result of receiving this frame.
 """
 b'unhandled frame type: %d' % frame.typeid
 )
 return meth(request, frame)
-def _onstreamsettingsframe(self, frame):
+def _onstreamsettingsframe(self, frame: frame):
 assert frame.typeid == FRAME_TYPE_STREAM_SETTINGS
 more = frame.flags & FLAG_STREAM_ENCODING_SETTINGS_CONTINUATION
 eos = frame.flags & FLAG_STREAM_ENCODING_SETTINGS_EOS
 },
 )
 return b'noop', {}
-def _oncommandresponseframe(self, request, frame):
+def _oncommandresponseframe(self, request: commandrequest, frame: frame):
 if frame.flags & FLAG_COMMAND_RESPONSE_EOS:
 request.state = b'received'
 del self._activerequests[request.requestid]
 return (
 b'eos': frame.flags & FLAG_COMMAND_RESPONSE_EOS,
 b'data': frame.payload,
 },
 )
-def _onerrorresponseframe(self, request, frame):
+def _onerrorresponseframe(self, request: commandrequest, frame: frame):
 request.state = b'errored'
 del self._activerequests[request.requestid]
 # The payload should be a CBOR map.
 m = cborutil.decodeall(frame.payload)[0]

changeset 52699	068398a8c9cb
parent 52698	1a612f9ec2c4