def _on_trailers_received()

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

• 12 lines of code
• 5 McCabe index (conditional complexity)

    def _on_trailers_received(self, trailers: Http2Headers) -> None:
        """
        Handle the trailers received.
        :param trailers: The trailers.
        :type trailers: Http2Headers
        """
        if not self._rpc_status and not self._headers_received:
            self._validate_headers(trailers)

        if self._rpc_status:
            self._rpc_status.append_description(f"Trailers: {trailers}")
        else:
            self._rpc_status = self._get_status_from_trailers(trailers)
            self._trailers = trailers

        if self._decoder:
            self._decoder.close()
        else:
            self._listener.on_complete(self._rpc_status, trailers.to_dict())