def on_message()

in src/dubbo/protocol/triple/call/client_call.py [0:0]

• 15 lines of code
• 3 McCabe index (conditional complexity)

    def on_message(self, data: bytes) -> None:
        """
        Called when a message is received from server.
        :param data: The message data
        :type data: bytes
        """
        if self._done:
            _LOGGER.warning("Call is done, cannot receive message")
            return

        try:
            # Deserialize the message
            message = self._deserializer.deserialize(data)
            self._listener.on_message(message)
        except SerializationError as e:
            _LOGGER.error("Failed to deserialize message: %s", e)
            # close the stream
            self.cancel_by_local(e)
            # close the listener
            status = TriRpcStatus(
                code=GRpcCode.INTERNAL,
                description="Failed to deserialize message",
            )
            self._listener.on_close(status, {})