(
self,
headers: h3_events.Headers,
data: bytes | None = None,
trailers: h3_events.Headers | None = None,
end_stream: bool = True,
)
| 621 | self.http_event_received(http_event) |
| 622 | |
| 623 | def request( |
| 624 | self, |
| 625 | headers: h3_events.Headers, |
| 626 | data: bytes | None = None, |
| 627 | trailers: h3_events.Headers | None = None, |
| 628 | end_stream: bool = True, |
| 629 | ) -> H3Response: |
| 630 | stream_id = self._quic.get_next_available_stream_id() |
| 631 | self.http.send_headers( |
| 632 | stream_id=stream_id, |
| 633 | headers=headers, |
| 634 | end_stream=data is None and trailers is None and end_stream, |
| 635 | ) |
| 636 | if data is not None: |
| 637 | self.http.send_data( |
| 638 | stream_id=stream_id, |
| 639 | data=data, |
| 640 | end_stream=trailers is None and end_stream, |
| 641 | ) |
| 642 | if trailers is not None: |
| 643 | self.http.send_headers( |
| 644 | stream_id=stream_id, |
| 645 | headers=trailers, |
| 646 | end_stream=end_stream, |
| 647 | ) |
| 648 | waiter = self._loop.create_future() |
| 649 | response = H3Response(waiter=waiter, stream_id=stream_id) |
| 650 | self._responses[stream_id] = response |
| 651 | self.transmit() |
| 652 | return response |
| 653 | |
| 654 | |
| 655 | T = TypeVar("T", bound=QuicClient) |
no test coverage detected