def on_headers()

in src/dubbo/protocol/triple/stream/client_stream.py [0:0]

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

    def on_headers(self, headers: Http2Headers, end_stream: bool) -> None:
        if not end_stream:
            # handle headers
            self._on_headers_received(headers)
        else:
            # handle trailers
            self._on_trailers_received(headers)

        if end_stream and not self._headers_received:
            self._handle_transport_error(self._rpc_status)