def _on_headers_received()

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

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

    def _on_headers_received(self, headers: Http2Headers) -> None:
        """
        Handle the headers received.
        :param headers: The headers.
        :type headers: Http2Headers
        """
        self._headers_received = True

        # validate headers
        self._validate_headers(headers)
        if self._rpc_status:
            return

        # get messageEncoding
        decompressor: Optional[Decompressor] = None
        message_encoding = headers.get(TripleHeaderName.GRPC_ENCODING.value, Identity.get_message_encoding())
        if message_encoding != Identity.get_message_encoding():
            try:
                # get decompressor by messageEncoding
                decompressor = extensionLoader.get_extension(Decompressor, message_encoding)()
            except ExtensionError:
                # unsupported
                self._rpc_status = TriRpcStatus(
                    GRpcCode.UNIMPLEMENTED,
                    description="Unsupported message encoding",
                )
                return

        self._decoder = TriDecoder(self, decompressor)