blob: 1d92b705e6de05e2cf56476698560511410df3c7 (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
|
from typing import Optional
from urllib.parse import urlsplit, parse_qsl
def session_established(session):
path: Optional[bytes] = None
for key, value in session.request_headers:
if key == b':path':
path = value
assert path is not None
qs = dict(parse_qsl(urlsplit(path).query))
code = qs[b'code']
if code is None:
raise Exception('code is missing, path = {}'.format(path))
session.dict_for_handlers['code'] = int(code)
def stream_data_received(session,
stream_id: int,
data: bytes,
stream_ended: bool):
code: int = session.dict_for_handlers['code']
if session.stream_is_unidirectional(stream_id):
session.stop_stream(stream_id, code)
else:
session.stop_stream(stream_id, code)
session.reset_stream(stream_id, code)
|
