def handle_result()

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

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

    def handle_result(self, result: Any) -> None:
        try:
            # check if the stream is completed
            if not self._read_write_stream.can_write_more():
                return

            if not self._rpc_type.server_stream:
                # get single result
                self._read_write_stream.write(result)
            else:
                # get multi results
                for message in result:
                    self._read_write_stream.write(message)

            self._read_write_stream.done_writing()
        except Exception as e:
            self.handle_exception(e)