diff --git a/src/client/WFKafkaClient.cc b/src/client/WFKafkaClient.cc index 8506d3ac..9c897db2 100644 --- a/src/client/WFKafkaClient.cc +++ b/src/client/WFKafkaClient.cc @@ -28,8 +28,9 @@ #include #include #include "WFTaskError.h" -#include "WFKafkaClient.h" #include "StringUtil.h" +#include "KafkaTaskImpl.inl" +#include "WFKafkaClient.h" #define KAFKA_HEARTBEAT_INTERVAL (3 * 1000 * 1000) diff --git a/src/client/WFKafkaClient.h b/src/client/WFKafkaClient.h index 93d27e49..460a95a9 100644 --- a/src/client/WFKafkaClient.h +++ b/src/client/WFKafkaClient.h @@ -23,10 +23,9 @@ #include #include #include +#include "WFTask.h" #include "KafkaMessage.h" #include "KafkaResult.h" -#include "KafkaTaskImpl.inl" - class WFKafkaTask; class WFKafkaClient; diff --git a/src/protocol/KafkaDataTypes.cc b/src/protocol/KafkaDataTypes.cc index f0e25b7a..870366bb 100644 --- a/src/protocol/KafkaDataTypes.cc +++ b/src/protocol/KafkaDataTypes.cc @@ -16,21 +16,50 @@ Authors: Wang Zhulei (wangzhulei@sogou-inc.com) */ - #include #include +#include #include "KafkaDataTypes.h" +#define MIN(x, y) ((x) <= (y) ? (x) : (y)) + namespace protocol { -#define MIN(x, y) ((x) <= (y) ? (x) : (y)) - -static int compare_member(const void *p1, const void *p2) +std::string KafkaConfig::get_sasl_info() const { - kafka_member_t *member1 = (kafka_member_t *)p1; - kafka_member_t *member2 = (kafka_member_t *)p2; - return strcmp(member1->member_id, member2->member_id); + std::string info; + + if (strcasecmp(this->ptr->mechanisms, "plain") == 0) + { + info += this->ptr->mechanisms; + info += "|"; + info += this->ptr->username; + info += "|"; + info += this->ptr->password; + info += "|"; + } + else if (strncasecmp(this->ptr->mechanisms, "SCRAM", 5) == 0) + { + info += this->ptr->mechanisms; + info += "|"; + info += this->ptr->username; + info += "|"; + info += this->ptr->password; + info += "|"; + } + + return info; +} + +static bool compare_member(const kafka_member_t *m1, const kafka_member_t *m2) +{ + return strcmp(m1->member_id, m2->member_id) < 0; +} + +inline void KafkaMetaSubscriber::sort_by_member() +{ + std::sort(this->member_vec.begin(), this->member_vec.end(), compare_member); } static bool operator<(const KafkaMetaSubscriber& s1, const KafkaMetaSubscriber& s2) @@ -114,7 +143,7 @@ int KafkaCgroup::kafka_roundrobin_assignor(kafka_member_t **members, int next = -1; std::sort(subscribers->begin(), subscribers->end()); - qsort(members, member_elements, sizeof (kafka_member_t *), compare_member); + std::sort(members, members + member_elements, compare_member); for (const auto& subscriber : *subscribers) { diff --git a/src/protocol/KafkaDataTypes.h b/src/protocol/KafkaDataTypes.h index 2c0d9329..325b332b 100644 --- a/src/protocol/KafkaDataTypes.h +++ b/src/protocol/KafkaDataTypes.h @@ -19,15 +19,14 @@ #ifndef _KAFKA_DATATYPES_H_ #define _KAFKA_DATATYPES_H_ - #include -#include #include #include #include +#include +#include #include #include -#include #include #include #include @@ -35,7 +34,6 @@ #include "rbtree.h" #include "kafka_parser.h" - namespace protocol { @@ -553,30 +551,7 @@ public: return kafka_sasl_set_password(password, this->ptr) == 0; } - std::string get_sasl_info() const - { - std::string info; - if (strcasecmp(this->ptr->mechanisms, "plain") == 0) - { - info += this->ptr->mechanisms; - info += "|"; - info += this->ptr->username; - info += "|"; - info += this->ptr->password; - info += "|"; - } - else if (strncasecmp(this->ptr->mechanisms, "SCRAM", 5) == 0) - { - info += this->ptr->mechanisms; - info += "|"; - info += this->ptr->username; - info += "|"; - info += this->ptr->password; - info += "|"; - } - - return info; - } + std::string get_sasl_info() const; bool new_client(kafka_sasl_t *sasl) { @@ -1284,15 +1259,7 @@ public: return &this->member_vec; } - static bool cmp(const kafka_member_t *m1, const kafka_member_t *m2) - { - return strcmp(m1->member_id, m2->member_id) < 0; - } - - void sort_by_member() - { - std::sort(this->member_vec.begin(), this->member_vec.end(), cmp); - } + void sort_by_member(); private: KafkaMeta *meta; diff --git a/src/protocol/kafka_parser.c b/src/protocol/kafka_parser.c index 3822baeb..b05c070f 100644 --- a/src/protocol/kafka_parser.c +++ b/src/protocol/kafka_parser.c @@ -16,14 +16,14 @@ Authors: Wang Zhulei (wangzhulei@sogou-inc.com) */ +#include +#include +#include +#include +#include #include #include #include -#include -#include -#include -#include -#include #include "kafka_parser.h" static kafka_api_version_t kafka_api_version_queryable[] = { @@ -610,13 +610,13 @@ int kafka_parser_append_message(const void *buf, size_t *size, if (s > parser->message_size - parser->cur_size) { - memcpy(parser->msgbuf + parser->cur_size, buf, + memcpy((char *)parser->msgbuf + parser->cur_size, buf, parser->message_size - parser->cur_size); parser->cur_size = parser->message_size; } else { - memcpy(parser->msgbuf + parser->cur_size, buf, s); + memcpy((char *)parser->msgbuf + parser->cur_size, buf, s); parser->cur_size += s; }