Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def __init__(self, config=None):
self.state_machine = H2ConnectionStateMachine()
self.streams = {}
self.highest_inbound_stream_id = 0
self.highest_outbound_stream_id = 0
self.encoder = Encoder()
self.decoder = Decoder()
# This won't always actually do anything: for versions of HPACK older
# than 2.3.0 it does nothing. However, we have to try!
self.decoder.max_header_list_size = self.DEFAULT_MAX_HEADER_LIST_SIZE
#: The configuration for this HTTP/2 connection object.
#:
#: .. versionadded:: 2.5.0
self.config = config
if self.config is None:
self.config = H2Configuration(
client_side=True,
)
# Objects that store settings, including defaults.
#
def __init__(
self,
tcp_handler=None,
rfile=None,
wfile=None,
is_server=False,
dump_frames=False,
encoder=None,
decoder=None,
unhandled_frame_cb=None,
):
self.tcp_handler = tcp_handler or TCPHandler(rfile, wfile)
self.is_server = is_server
self.dump_frames = dump_frames
self.encoder = encoder or Encoder()
self.decoder = decoder or Decoder()
self.unhandled_frame_cb = unhandled_frame_cb
self.http2_settings = self.HTTP2_DEFAULT_SETTINGS.copy()
self.current_stream_id = None
self.connection_preface_performed = False
flags=FLAG_NO_FLAGS,
stream_id=0x0):
valid_flags = 0
for flag in self.VALID_FLAGS:
valid_flags |= flag
if flags | valid_flags != valid_flags:
raise ValueError('invalid flags detected.')
if state is None:
class State(object):
pass
state = State()
state.http2_settings = HTTP2_DEFAULT_SETTINGS.copy()
state.encoder = Encoder()
state.decoder = Decoder()
self.state = state
self.length = length
self.type = self.TYPE
self.flags = flags
self.stream_id = stream_id
def __init__(self, client_side=True, header_encoding='utf-8', config=None):
self.state_machine = H2ConnectionStateMachine()
self.streams = {}
self.highest_inbound_stream_id = 0
self.highest_outbound_stream_id = 0
self.encoder = Encoder()
self.decoder = Decoder()
# This won't always actually do anything: for versions of HPACK older
# than 2.3.0 it does nothing. However, we have to try!
self.decoder.max_header_list_size = self.DEFAULT_MAX_HEADER_LIST_SIZE
#: The configuration for this HTTP/2 connection object.
#:
#: .. versionadded:: 2.5.0
self.config = config
if self.config is None:
self.config = H2Configuration(
client_side=client_side,
header_encoding=header_encoding,
)
# Objects that store settings, including defaults.
def __init__(
self,
tcp_handler=None,
rfile=None,
wfile=None,
is_server=False,
dump_frames=False,
encoder=None,
decoder=None,
unhandled_frame_cb=None,
):
self.tcp_handler = tcp_handler or TCPHandler(rfile, wfile)
self.is_server = is_server
self.dump_frames = dump_frames
self.encoder = encoder or Encoder()
self.decoder = decoder or Decoder()
self.unhandled_frame_cb = unhandled_frame_cb
self.http2_settings = self.HTTP2_DEFAULT_SETTINGS.copy()
self.current_stream_id = None
self.connection_preface_performed = False
def __init__(self, config=None):
self.state_machine = H2ConnectionStateMachine()
self.streams = {}
self.highest_inbound_stream_id = 0
self.highest_outbound_stream_id = 0
self.encoder = Encoder()
self.decoder = Decoder()
# This won't always actually do anything: for versions of HPACK older
# than 2.3.0 it does nothing. However, we have to try!
self.decoder.max_header_list_size = self.DEFAULT_MAX_HEADER_LIST_SIZE
#: The configuration for this HTTP/2 connection object.
#:
#: .. versionadded:: 2.5.0
self.config = config
if self.config is None:
self.config = H2Configuration(
client_side=True,
)
# Objects that store settings, including defaults.
#
flags=FLAG_NO_FLAGS,
stream_id=0x0):
valid_flags = 0
for flag in self.VALID_FLAGS:
valid_flags |= flag
if flags | valid_flags != valid_flags:
raise ValueError('invalid flags detected.')
if state is None:
class State(object):
pass
state = State()
state.http2_settings = HTTP2_DEFAULT_SETTINGS.copy()
state.encoder = Encoder()
state.decoder = Decoder()
self.state = state
self.length = length
self.type = self.TYPE
self.flags = flags
self.stream_id = stream_id