def send_headers()

in src/dubbo/remoting/aio/http2/stream.py [0:0]

• 8 lines of code
• 4 McCabe index (conditional complexity)

    def send_headers(self, headers: Http2Headers, end_stream: bool = False) -> None:
        if self.local_closed:
            raise StreamError("The stream has been closed locally.")
        elif self._headers_sent and not end_stream:
            raise StreamError("Trailers must be the last frame of the stream(end_stream must be True).")

        self._headers_sent = True
        headers_frame = HeadersFrame(self.id, headers, end_stream=end_stream)
        self._outbound_controller.write_headers(headers_frame)