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