def heart_beat()

in datahub/implement.py [0:0]

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

    def heart_beat(self, project_name, topic_name, consumer_group, consumer_id, version_id, hold_shard_list, read_end_shard_list):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')
        if consumer_group is None or len(consumer_group) == 0:
            raise InvalidParameterException("Consumer group format is invalid")

        if hold_shard_list is None:
            raise InvalidParameterException("Hold shard list is none")

        url = Path.SUBSCRIPTION % (project_name, topic_name, consumer_group)
        request_param = HeartBeatParams(consumer_id, version_id, hold_shard_list, read_end_shard_list)
        content = self._rest_client.post(url, data=request_param.content())
        result = HeartBeatResult.parse_content(content)
        return result