def on_data()

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

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

    def on_data(self, data: bytes, end_stream: bool) -> None:
        if self._rpc_status:
            self._rpc_status.append_description(f"Data: {data.decode('utf-8')}")
            if len(self._rpc_status.description) > 512 or end_stream:
                self._handle_transport_error(self._rpc_status)
            return

        # decode the data
        self._decoder.decode(data)