Merge pull request #329 from Barenboim/master

remove std::tuple in kafka client

Author: Barenboim

src/client/WFKafkaClient.cc

@@ -678,13 +678,20 @@ void ComplexKafkaTask::kafka_meta_callback(__WFKafkaTask *task)
 	WFTaskFactory::count_by_name(name, (unsigned int)-1);
 }
 
+struct __broker_status
+{
+	KafkaBroker *broker;
+	int state;
+	int error;
+};
+
 void ComplexKafkaTask::kafka_broker_api_callback(__WFKafkaTask *task)
 {
-	using KafkaTuple = std::tuple<int, int, void *>;
-	KafkaTuple * state_error_broker = new KafkaTuple{task->get_state(),
-													 task->get_error(),
-													 task->user_data};
-	series_of(task)->set_context(state_error_broker);
+	struct __broker_status *status = new struct __broker_status;
+	status->broker = (KafkaBroker *)task->user_data;
+	status->state = task->get_state();
+	status->error = task->get_error();
+	series_of(task)->set_context(status);
 }
 
 void ComplexKafkaTask::kafka_broker_callback(const ParallelWork *pwork)
@@ -694,21 +701,20 @@ void ComplexKafkaTask::kafka_broker_callback(const ParallelWork *pwork)
 	t->error = 0;
 
 	t->lock_status.get_mutex()->lock();
-	using KafkaTuple = std::tuple<int, int, KafkaBroker *>;
-	KafkaTuple *state_error_broker;
+	struct __broker_status *status;
 
 	for (size_t i = 0; i < pwork->size(); i++)
 	{
-		state_error_broker = (KafkaTuple *)pwork->series_at(i)->get_context();
-		if (std::get<0>(*state_error_broker) != WFT_STATE_SUCCESS)
+		status = (struct __broker_status *)pwork->series_at(i)->get_context();
+		if (status->state != WFT_STATE_SUCCESS)
 		{
-			t->state = std::get<0>(*state_error_broker);
-			t->error = std::get<1>(*state_error_broker);
+			t->state = status->state;
+			t->error = status->error;
 		}
 		else
-			t->client_broker_map.add_item(*std::get<2>(*state_error_broker));
+			t->client_broker_map.add_item(*status->broker);
 
-		delete state_error_broker;
+		delete status;
 	}
 
 	if (t->state == WFT_STATE_SUCCESS)

src/protocol/kafka_parser.c

@@ -23,9 +23,6 @@
 #include <ctype.h>
 #include "kafka_parser.h"
 
-
-#define MIN(a, b)	((x) <= (y) ? (x) : (y))
-
 static kafka_api_version_t kafka_api_version_queryable[] = {
 	{ Kafka_ApiVersions, 0, 0 }
 };
@@ -207,8 +204,8 @@ static int kafka_get_legacy_api_version(const char *broker_version,
 		{ "", kafka_api_version_queryable, 1 },
 		{ NULL, NULL, 0 }
 	};
-
 	int i;
+
 	for (i = 0 ; vermap[i].pfx ; i++)
 	{
 		if (!strncmp(vermap[i].pfx, broker_version, strlen(vermap[i].pfx)))
@@ -263,7 +260,7 @@ unsigned kafka_get_features(kafka_api_version_t *api, size_t api_cnt)
 	int i, fails, r;
 	const kafka_api_version_t *match;
 
-	for (i = 0 ; kafka_feature_map[i].feature != 0 ; i++)
+	for (i = 0; kafka_feature_map[i].feature != 0; i++)
 	{
 		fails = 0;
 		for (match = &kafka_feature_map[i].depends[0];
@@ -574,6 +571,7 @@ void kafka_block_deinit(kafka_block_t *block)
 int kafka_parser_append_message(const void *buf, size_t *size,
 								kafka_parser_t *parser)
 {
+	size_t s = *size;
 	int totaln;
 
 	if (parser->complete)
@@ -582,9 +580,7 @@ int kafka_parser_append_message(const void *buf, size_t *size,
 		return 1;
 	}
 
-	size_t s = *size;
-
-	if (parser->hsize + *size < 4)
+	if (parser->hsize + s < 4)
 	{
 		memcpy(parser->headbuf + parser->hsize, buf, s);
 		parser->hsize += s;
@@ -672,15 +668,12 @@ int kafka_record_header_set_kv(const void *key, size_t key_len,
 							   kafka_record_header_t *header)
 {
 	void *k = malloc(key_len);
-
-	if (!k)
-		return -1;
-
 	void *v = malloc(val_len);
 
-	if (!v)
+	if (!k || !v)
 	{
 		free(k);
+		free(v);
 		return -1;
 	}
 
@@ -728,11 +721,12 @@ int kafka_sasl_plain_client_new(void *p)
 	size_t ulen = strlen(conf->sasl.username);
 	size_t plen = strlen(conf->sasl.passwd);
 	size_t blen = ulen + plen + 3;
-	char *buf = malloc(blen);
+	size_t off = 0;
+	char *buf = (char *)malloc(blen);
+
 	if (!buf)
 		return -1;
 
-	size_t off = 0;
 	buf[off++] = '\0';
 
 	memcpy(buf + off, conf->sasl.username, ulen);
@@ -783,3 +777,4 @@ int kafka_sasl_set_passwd(const char *passwd, kafka_config_t *conf)
 	conf->sasl.passwd = t;
 	return 0;
 }
+