def send_message()

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

• 11 lines of code
• 2 McCabe index (conditional complexity)

    def send_message(self, message: Any) -> None:
        if not self._headers_sent:
            headers = Http2Headers()
            headers.status = HttpStatus.OK.value
            headers.add(
                TripleHeaderName.CONTENT_TYPE.value,
                TripleHeaderValue.APPLICATION_GRPC_PROTO.value,
            )
            self._server_stream.send_headers(headers)

        serialized_data = FunctionHelper.call_func(self._serializer.serialize, message)
        # TODO support compression
        self._server_stream.send_message(serialized_data, False)