(self, event: Event)
| 418 | super().__init__(context, context.client) |
| 419 | |
| 420 | def _handle_event(self, event: Event) -> CommandGenerator[None]: |
| 421 | if isinstance(event, ResponseHeaders): |
| 422 | if self.is_open_for_us(event.stream_id): |
| 423 | self.h2_conn.send_headers( |
| 424 | event.stream_id, |
| 425 | headers=( |
| 426 | yield from format_h2_response_headers(self.context, event) |
| 427 | ), |
| 428 | end_stream=event.end_stream, |
| 429 | ) |
| 430 | yield SendData(self.conn, self.h2_conn.data_to_send()) |
| 431 | else: |
| 432 | yield from super()._handle_event(event) |
| 433 | |
| 434 | def handle_h2_event(self, event: h2.events.Event) -> CommandGenerator[bool]: |
| 435 | if isinstance(event, h2.events.RequestReceived): |
no test coverage detected