def __init_

def init()

in python/rocketmq/protocol/service_pb2_grpc.py [0:0]
56 lines of code
1 McCabe index (conditional complexity)

    def __init__(self, channel):
        """Constructor.

        Args:
            channel: A grpc.Channel.
        """
        self.QueryRoute = channel.unary_unary(
                '/apache.rocketmq.v2.MessagingService/QueryRoute',
                request_serializer=apache_dot_rocketmq_dot_v2_dot_service__pb2.QueryRouteRequest.SerializeToString,
                response_deserializer=apache_dot_rocketmq_dot_v2_dot_service__pb2.QueryRouteResponse.FromString,
                )
        self.Heartbeat = channel.unary_unary(
                '/apache.rocketmq.v2.MessagingService/Heartbeat',
                request_serializer=apache_dot_rocketmq_dot_v2_dot_service__pb2.HeartbeatRequest.SerializeToString,
                response_deserializer=apache_dot_rocketmq_dot_v2_dot_service__pb2.HeartbeatResponse.FromString,
                )
        self.SendMessage = channel.unary_unary(
                '/apache.rocketmq.v2.MessagingService/SendMessage',
                request_serializer=apache_dot_rocketmq_dot_v2_dot_service__pb2.SendMessageRequest.SerializeToString,
                response_deserializer=apache_dot_rocketmq_dot_v2_dot_service__pb2.SendMessageResponse.FromString,
                )
        self.QueryAssignment = channel.unary_unary(
                '/apache.rocketmq.v2.MessagingService/QueryAssignment',
                request_serializer=apache_dot_rocketmq_dot_v2_dot_service__pb2.QueryAssignmentRequest.SerializeToString,
                response_deserializer=apache_dot_rocketmq_dot_v2_dot_service__pb2.QueryAssignmentResponse.FromString,
                )
        self.ReceiveMessage = channel.unary_stream(
                '/apache.rocketmq.v2.MessagingService/ReceiveMessage',
                request_serializer=apache_dot_rocketmq_dot_v2_dot_service__pb2.ReceiveMessageRequest.SerializeToString,
                response_deserializer=apache_dot_rocketmq_dot_v2_dot_service__pb2.ReceiveMessageResponse.FromString,
                )
        self.AckMessage = channel.unary_unary(
                '/apache.rocketmq.v2.MessagingService/AckMessage',
                request_serializer=apache_dot_rocketmq_dot_v2_dot_service__pb2.AckMessageRequest.SerializeToString,
                response_deserializer=apache_dot_rocketmq_dot_v2_dot_service__pb2.AckMessageResponse.FromString,
                )
        self.ForwardMessageToDeadLetterQueue = channel.unary_unary(
                '/apache.rocketmq.v2.MessagingService/ForwardMessageToDeadLetterQueue',
                request_serializer=apache_dot_rocketmq_dot_v2_dot_service__pb2.ForwardMessageToDeadLetterQueueRequest.SerializeToString,
                response_deserializer=apache_dot_rocketmq_dot_v2_dot_service__pb2.ForwardMessageToDeadLetterQueueResponse.FromString,
                )
        self.EndTransaction = channel.unary_unary(
                '/apache.rocketmq.v2.MessagingService/EndTransaction',
                request_serializer=apache_dot_rocketmq_dot_v2_dot_service__pb2.EndTransactionRequest.SerializeToString,
                response_deserializer=apache_dot_rocketmq_dot_v2_dot_service__pb2.EndTransactionResponse.FromString,
                )
        self.Telemetry = channel.stream_stream(
                '/apache.rocketmq.v2.MessagingService/Telemetry',
                request_serializer=apache_dot_rocketmq_dot_v2_dot_service__pb2.TelemetryCommand.SerializeToString,
                response_deserializer=apache_dot_rocketmq_dot_v2_dot_service__pb2.TelemetryCommand.FromString,
                )
        self.NotifyClientTermination = channel.unary_unary(
                '/apache.rocketmq.v2.MessagingService/NotifyClientTermination',
                request_serializer=apache_dot_rocketmq_dot_v2_dot_service__pb2.NotifyClientTerminationRequest.SerializeToString,
                response_deserializer=apache_dot_rocketmq_dot_v2_dot_service__pb2.NotifyClientTerminationResponse.FromString,
                )
        self.ChangeInvisibleDuration = channel.unary_unary(
                '/apache.rocketmq.v2.MessagingService/ChangeInvisibleDuration',
                request_serializer=apache_dot_rocketmq_dot_v2_dot_service__pb2.ChangeInvisibleDurationRequest.SerializeToString,
                response_deserializer=apache_dot_rocketmq_dot_v2_dot_service__pb2.ChangeInvisibleDurationResponse.FromString,
                )
def __init__()

def init()
