Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
max_window_size=self.local_settings.initial_window_size
)
# When in doubt use dict-dispatch.
self._frame_dispatch_table = {
HeadersFrame: self._receive_headers_frame,
PushPromiseFrame: self._receive_push_promise_frame,
SettingsFrame: self._receive_settings_frame,
DataFrame: self._receive_data_frame,
WindowUpdateFrame: self._receive_window_update_frame,
PingFrame: self._receive_ping_frame,
RstStreamFrame: self._receive_rst_stream_frame,
PriorityFrame: self._receive_priority_frame,
GoAwayFrame: self._receive_goaway_frame,
ContinuationFrame: self._receive_naked_continuation,
AltSvcFrame: self._receive_alt_svc_frame,
ExtensionFrame: self._receive_unknown_frame
}
def test_altsvc_frame_flags(self):
f = AltSvcFrame(stream_id=0)
flags = f.parse_flags(0xFF)
assert flags == set()
def build_alt_svc_frame(self, stream_id, origin, field):
"""
Builds a single ALTSVC frame.
"""
f = AltSvcFrame(stream_id)
f.origin = origin
f.field = field
return f
def advertise_alternative_service(self, field_value):
"""
Advertise an RFC 7838 alternative service. The semantics of this are
better documented in the ``H2Connection`` class.
"""
self.config.logger.debug(
"Advertise alternative service of %r for %r", field_value, self
)
self.state_machine.process_input(StreamInputs.SEND_ALTERNATIVE_SERVICE)
asf = AltSvcFrame(self.stream_id)
asf.field = field_value
return [asf]
def test_altsvc_frame_with_origin_serializes_properly(self):
f = AltSvcFrame(stream_id=0)
f.origin = b'example.com'
f.field = b'h2="alt.example.com:8000", h2=":443"'
s = f.serialize()
assert s == self.payload_with_origin
def advertise_alternative_service(self, field_value):
"""
Advertise an RFC 7838 alternative service. The semantics of this are
better documented in the ``H2Connection`` class.
"""
self.config.logger.debug(
"Advertise alternative service of %r for %r", field_value, self
)
self.state_machine.process_input(StreamInputs.SEND_ALTERNATIVE_SERVICE)
asf = AltSvcFrame(self.stream_id)
asf.field = field_value
return [asf]
def __init__(self, stream_id, origin=b'', field=b'', **kwargs):
super(AltSvcFrame, self).__init__(stream_id, **kwargs)
if not isinstance(origin, bytes):
raise InvalidFrameError("AltSvc origin must be bytestring.")
if not isinstance(field, bytes):
raise InvalidFrameError("AltSvc field must be a bytestring.")
self.origin = origin
self.field = field
return header + self.body
_FRAME_CLASSES = [
DataFrame,
HeadersFrame,
PriorityFrame,
RstStreamFrame,
SettingsFrame,
PushPromiseFrame,
PingFrame,
GoAwayFrame,
WindowUpdateFrame,
ContinuationFrame,
AltSvcFrame,
]
#: FRAMES maps the type byte for each frame to the class used to represent that
#: frame.
FRAMES = {cls.type: cls for cls in _FRAME_CLASSES}
max_window_size=self.local_settings.initial_window_size
)
# When in doubt use dict-dispatch.
self._frame_dispatch_table = {
HeadersFrame: self._receive_headers_frame,
PushPromiseFrame: self._receive_push_promise_frame,
SettingsFrame: self._receive_settings_frame,
DataFrame: self._receive_data_frame,
WindowUpdateFrame: self._receive_window_update_frame,
PingFrame: self._receive_ping_frame,
RstStreamFrame: self._receive_rst_stream_frame,
PriorityFrame: self._receive_priority_frame,
GoAwayFrame: self._receive_goaway_frame,
ContinuationFrame: self._receive_naked_continuation,
AltSvcFrame: self._receive_alt_svc_frame,
ExtensionFrame: self._receive_unknown_frame
}