FS-9775: Encapsulated message details into ks_dht2_message_t

libs/libks/Makefile.am

@@ -13,7 +13,7 @@ libks_la_SOURCES += src/ks_time.c src/ks_printf.c src/ks_hash.c src/ks_q.c src/k
 libks_la_SOURCES += src/ks_ssl.c src/kws.c src/ks_rng.c
 libks_la_SOURCES += src/utp/utp_api.cpp src/utp/utp_callbacks.cpp src/utp/utp_hash.cpp src/utp/utp_internal.cpp
 libks_la_SOURCES += src/utp/utp_packedsockaddr.cpp src/utp/utp_utils.cpp src/ks_bencode.c
-libks_la_SOURCES += src/dht/ks_dht.c src/dht/ks_dht_endpoint.c src/dht/ks_dht_nodeid.c
+libks_la_SOURCES += src/dht/ks_dht.c src/dht/ks_dht_endpoint.c src/dht/ks_dht_nodeid.c src/dht/ks_dht_message.c
 libks_la_SOURCES += crypt/aeskey.c crypt/aestab.c crypt/sha2.c crypt/twofish.c crypt/aes_modes.c crypt/aescrypt.c crypt/twofish_cfb.c 
 #aes.h aescpp.h brg_endian.h aesopt.h aestab.h brg_types.h sha2.h twofish.h
 
@@ -29,7 +29,8 @@ library_include_HEADERS += src/include/ks_dso.h src/include/ks_dht.h src/include
 library_include_HEADERS += src/include/ks_printf.h src/include/ks_hash.h src/include/ks_ssl.h src/include/kws.h
 library_include_HEADERS += src/utp/utp_internal.h src/utp/utp.h src/utp/utp_types.h src/utp/utp_callbacks.h src/utp/utp_templates.h
 library_include_HEADERS += src/utp/utp_hash.h src/utp/utp_packedsockaddr.h src/utp/utp_utils.h src/include/ks_utp.h
-library_include_HEADERS += src/dht/ks_dht.h src/dht/ks_dht-int.h src/dht/ks_dht_endpoint.h src/dht/ks_dht_endpoint-int.h src/dht/ks_dht_nodeid.h
+library_include_HEADERS += src/dht/ks_dht.h src/dht/ks_dht-int.h src/dht/ks_dht_endpoint.h src/dht/ks_dht_endpoint-int.h
+library_include_HEADERS += src/dht/ks_dht_nodeid.h src/dht/ks_dht_message.h
 
 tests: libks.la
 	$(MAKE) -C test tests

libs/libks/src/dht/ks_dht-int.h

@@ -8,11 +8,7 @@ KS_BEGIN_EXTERN_C
 
 KS_DECLARE(ks_status_t) ks_dht2_idle(ks_dht2_t *dht);
 KS_DECLARE(ks_status_t) ks_dht2_process(ks_dht2_t *dht, ks_sockaddr_t *raddr);
-KS_DECLARE(ks_status_t) ks_dht2_parse(ks_dht2_t *dht,
-									  struct bencode **message,
-									  uint8_t *transactionid,
-									  ks_size_t *transactionid_len,
-									  char *messagetype);
+
 						
 KS_END_EXTERN_C
 

libs/libks/src/dht/ks_dht.c

@@ -44,6 +44,7 @@ KS_DECLARE(ks_status_t) ks_dht2_free(ks_dht2_t *dht)
 	ks_pool_t *pool = dht->pool;
 	ks_bool_t pool_alloc = dht->pool_alloc;
 
+	ks_dht2_deinit(dht);
 	ks_pool_free(pool, dht);
 	if (pool_alloc) {
 		ks_pool_close(&pool);
@@ -99,6 +100,7 @@ KS_DECLARE(ks_status_t) ks_dht2_deinit(ks_dht2_t *dht)
 		ks_dht2_endpoint_deinit(ep);
 		ks_dht2_endpoint_free(ep);
 	}
+	dht->endpoints_size = 0;
 	if (dht->endpoints) {
 		ks_pool_free(dht->pool, dht->endpoints);
 		dht->endpoints = NULL;
@@ -107,11 +109,17 @@ KS_DECLARE(ks_status_t) ks_dht2_deinit(ks_dht2_t *dht)
 		ks_pool_free(dht->pool, dht->endpoints_poll);
 		dht->endpoints_poll = NULL;
 	}
+	if (dht->endpoints_hash) {
 		ks_hash_destroy(&dht->endpoints_hash);
+		dht->endpoints_hash = NULL;
+	}
 	dht->bind_ipv4 = KS_FALSE;
 	dht->bind_ipv6 = KS_FALSE;
 
+	if (dht->registry_y) {
 		ks_hash_destroy(&dht->registry_y);
+		dht->registry_y = NULL;
+	}
 
 	ks_dht2_nodeid_deinit(&dht->nodeid);
 	
@@ -250,10 +258,7 @@ KS_DECLARE(ks_status_t) ks_dht2_idle(ks_dht2_t *dht)
  */
 KS_DECLARE(ks_status_t) ks_dht2_process(ks_dht2_t *dht, ks_sockaddr_t *raddr)
 {
-	struct bencode *message = NULL;
-	uint8_t transactionid[KS_DHT_TRANSACTIONID_MAX_SIZE];
-	ks_size_t transactionid_len;
-	char messagetype[KS_DHT_MESSAGETYPE_MAX_SIZE];
+	ks_dht2_message_t message;
 	ks_dht2_registry_callback_t callback;
 	ks_status_t ret = KS_STATUS_FAIL;
 
@@ -268,97 +273,23 @@ KS_DECLARE(ks_status_t) ks_dht2_process(ks_dht2_t *dht, ks_sockaddr_t *raddr)
 
 	// @todo blacklist check for bad actor nodes
 	
-	if (ks_dht2_parse(dht, &message, transactionid, &transactionid_len, messagetype) != KS_STATUS_SUCCESS) {
+	if (ks_dht2_message_prealloc(&message, dht->pool) != KS_STATUS_SUCCESS) {
 		return KS_STATUS_FAIL;
 	}
 
-	if (!(callback = (ks_dht2_registry_callback_t)(intptr_t)ks_hash_search(dht->registry_y, messagetype, KS_UNLOCKED))) {
-		ks_log(KS_LOG_DEBUG, "Message type '%s' is not registered\n", messagetype);
-	} else {
-		ret = callback(dht, raddr, transactionid, transactionid_len, message);
-	}
-
-	ben_free(message);
-	return ret;
-}
-
-/**
- *
- */
-KS_DECLARE(ks_status_t) ks_dht2_parse(ks_dht2_t *dht,
-									  struct bencode **message,
-									  uint8_t *transactionid,
-									  ks_size_t *transactionid_len,
-									  char *messagetype)
-{
-	struct bencode *msg = NULL;
-	struct bencode *t;
-	struct bencode *y;
-	const char *tv;
-	const char *yv;
-	ks_size_t tv_len;
-	ks_size_t yv_len;
-
-	ks_assert(dht);
-	ks_assert(message);
-	ks_assert(transactionid);
-	ks_assert(messagetype);
-
-	msg = ben_decode((const void *)dht->recv_buffer, dht->recv_buffer_length);
-	if (!msg) {
-		ks_log(KS_LOG_DEBUG, "Message cannot be decoded\n");
-	    goto failure;
-	}
-
-	ks_log(KS_LOG_DEBUG, "Message decoded\n");
-	ks_log(KS_LOG_DEBUG, "%s\n", ben_print(msg));
-
-	t = ben_dict_get_by_str(msg, "t");
-	if (!t) {
-		ks_log(KS_LOG_DEBUG, "Message missing required key 't'\n");
-		goto failure;
-	}
-
-	tv = ben_str_val(t);
-	tv_len = ben_str_len(t);
-	if (tv_len > KS_DHT_TRANSACTIONID_MAX_SIZE) {
-		ks_log(KS_LOG_DEBUG, "Message 't' value has an unexpectedly large size of %d\n", tv_len);
-		goto failure;
-	}
-
-	memcpy(transactionid, tv, tv_len);
-	*transactionid_len = tv_len;
-	// @todo hex output of transactionid
-	//ks_log(KS_LOG_DEBUG, "Message transaction id is %d\n", *transactionid);
-
-	y = ben_dict_get_by_str(msg, "y");
-	if (!y) {
-		ks_log(KS_LOG_DEBUG, "Message missing required key 'y'\n");
-		goto failure;
+	if (ks_dht2_message_init(&message, dht->recv_buffer, dht->recv_buffer_length) != KS_STATUS_SUCCESS) {
+		return KS_STATUS_FAIL;
 	}
 	
-	yv = ben_str_val(y);
-	yv_len = ben_str_len(y);
-	if (yv_len >= KS_DHT_MESSAGETYPE_MAX_SIZE) {
-		ks_log(KS_LOG_DEBUG, "Message 'y' value has an unexpectedly large size of %d\n", yv_len);
-		goto failure;
+	if (!(callback = (ks_dht2_registry_callback_t)(intptr_t)ks_hash_search(dht->registry_y, message.type, KS_UNLOCKED))) {
+		ks_log(KS_LOG_DEBUG, "Message type '%s' is not registered\n", message.type);
+	} else {
+		ret = callback(dht, raddr, &message);
 	}
 
-	memcpy(messagetype, yv, yv_len);
-	messagetype[yv_len] = '\0';
-	ks_log(KS_LOG_DEBUG, "Message type is '%s'\n", messagetype);
-	
-	*message = msg;
-	return KS_STATUS_SUCCESS;
+	ks_dht2_message_deinit(&message);
 	
- failure:
-	if (msg) {
-		ben_free(msg);
-	}
-	*message = NULL;
-	*transactionid_len = 0;
-	messagetype[0] = '\0';
-	return KS_STATUS_FAIL;
+	return ret;
 }
 
 /* For Emacs:

libs/libks/src/dht/ks_dht.h

@@ -5,6 +5,7 @@
 #include "ks_bencode.h"
 
 #include "ks_dht_endpoint.h"
+#include "ks_dht_message.h"
 #include "ks_dht_nodeid.h"
 
 KS_BEGIN_EXTERN_C
@@ -12,8 +13,6 @@ KS_BEGIN_EXTERN_C
 
 #define KS_DHT_DEFAULT_PORT 5309
 #define KS_DHT_RECV_BUFFER_SIZE 0xFFFF
-#define KS_DHT_TRANSACTIONID_MAX_SIZE 20
-#define KS_DHT_MESSAGETYPE_MAX_SIZE 20
 
 typedef struct ks_dht2_s ks_dht2_t;
 struct ks_dht2_s {
@@ -36,11 +35,7 @@ struct ks_dht2_s {
 	ks_size_t recv_buffer_length;
 };
 
-typedef ks_status_t (*ks_dht2_registry_callback_t)(ks_dht2_t *dht,
-												   ks_sockaddr_t *raddr,
-												   uint8_t *transactionid,
-												   ks_size_t transactionid_len,
-												   struct bencode *message);
+typedef ks_status_t (*ks_dht2_registry_callback_t)(ks_dht2_t *dht, ks_sockaddr_t *raddr, ks_dht2_message_t *message);
 
 
 KS_DECLARE(ks_status_t) ks_dht2_alloc(ks_dht2_t **dht, ks_pool_t *pool);

libs/libks/src/dht/ks_dht_endpoint.c

@@ -13,6 +13,7 @@ KS_DECLARE(ks_status_t) ks_dht2_endpoint_alloc(ks_dht2_endpoint_t **endpoint, ks
 	
 	*endpoint = ep = ks_pool_alloc(pool, sizeof(ks_dht2_endpoint_t));
 	ep->pool = pool;
+	ep->sock = KS_SOCK_INVALID;
 
 	return KS_STATUS_SUCCESS;
 }
@@ -26,6 +27,7 @@ KS_DECLARE(ks_status_t) ks_dht2_endpoint_prealloc(ks_dht2_endpoint_t *endpoint,
 	ks_assert(pool);
 
 	endpoint->pool = pool;
+	endpoint->sock = KS_SOCK_INVALID;
 
 	return KS_STATUS_SUCCESS;
 }
@@ -37,6 +39,7 @@ KS_DECLARE(ks_status_t) ks_dht2_endpoint_free(ks_dht2_endpoint_t *endpoint)
 {
 	ks_assert(endpoint);
 
+	ks_dht2_endpoint_deinit(endpoint);
 	ks_pool_free(endpoint->pool, endpoint);
 
 	return KS_STATUS_SUCCESS;
@@ -66,7 +69,10 @@ KS_DECLARE(ks_status_t) ks_dht2_endpoint_deinit(ks_dht2_endpoint_t *endpoint)
 {
 	ks_assert(endpoint);
 
+	if (endpoint->sock != KS_SOCK_INVALID) {
 		ks_socket_close(&endpoint->sock);
+		endpoint->sock = KS_SOCK_INVALID;
+	}
 
 	return KS_STATUS_SUCCESS;
 }

libs/libks/src/dht/ks_dht_message.c

+#include "ks_dht.h"
+
+/**
+ *
+ */
+KS_DECLARE(ks_status_t) ks_dht2_message_alloc(ks_dht2_message_t **message, ks_pool_t *pool)
+{
+	ks_dht2_message_t *msg;
+
+	ks_assert(message);
+	ks_assert(pool);
+	
+	*message = msg = ks_pool_alloc(pool, sizeof(ks_dht2_message_t));
+	msg->pool = pool;
+
+	return KS_STATUS_SUCCESS;
+}
+
+/**
+ *
+ */
+KS_DECLARE(ks_status_t) ks_dht2_message_prealloc(ks_dht2_message_t *message, ks_pool_t *pool)
+{
+	ks_assert(message);
+	ks_assert(pool);
+	
+	message->pool = pool;
+
+	return KS_STATUS_SUCCESS;
+}
+
+/**
+ *
+ */
+KS_DECLARE(ks_status_t) ks_dht2_message_free(ks_dht2_message_t *message)
+{
+	ks_assert(message);
+
+	ks_dht2_message_deinit(message);
+	ks_pool_free(message->pool, message);
+
+	return KS_STATUS_SUCCESS;
+}
+												
+
+/**
+ *
+ */
+KS_DECLARE(ks_status_t) ks_dht2_message_init(ks_dht2_message_t *message, const uint8_t *buffer, ks_size_t buffer_length)
+{
+	struct bencode *t;
+	struct bencode *y;
+	const char *tv;
+	const char *yv;
+	ks_size_t tv_len;
+	ks_size_t yv_len;
+
+	ks_assert(message);
+	ks_assert(message->pool);
+	ks_assert(buffer);
+
+    message->data = ben_decode((const void *)buffer, buffer_length);
+	if (!message->data) {
+		ks_log(KS_LOG_DEBUG, "Message cannot be decoded\n");
+		goto failure;
+	}
+
+	ks_log(KS_LOG_DEBUG, "Message decoded\n");
+	ks_log(KS_LOG_DEBUG, "%s\n", ben_print(message->data));
+	
+    t = ben_dict_get_by_str(message->data, "t");
+	if (!t) {
+		ks_log(KS_LOG_DEBUG, "Message missing required key 't'\n");
+		goto failure;
+	}
+
+	tv = ben_str_val(t);
+	tv_len = ben_str_len(t);
+	if (tv_len > KS_DHT_MESSAGE_TRANSACTIONID_MAX_SIZE) {
+		ks_log(KS_LOG_DEBUG, "Message 't' value has an unexpectedly large size of %d\n", tv_len);
+		goto failure;
+	}
+
+	memcpy(message->transactionid, tv, tv_len);
+	message->transactionid_length = tv_len;
+	// @todo hex output of transactionid
+	//ks_log(KS_LOG_DEBUG, "Message transaction id is %d\n", *transactionid);
+
+    y = ben_dict_get_by_str(message->data, "y");
+	if (!y) {
+		ks_log(KS_LOG_DEBUG, "Message missing required key 'y'\n");
+		goto failure;
+	}
+
+	yv = ben_str_val(y);
+	yv_len = ben_str_len(y);
+	if (yv_len >= KS_DHT_MESSAGE_TYPE_MAX_SIZE) {
+		ks_log(KS_LOG_DEBUG, "Message 'y' value has an unexpectedly large size of %d\n", yv_len);
+		goto failure;
+	}
+
+	memcpy(message->type, yv, yv_len);
+	message->type[yv_len] = '\0';
+	ks_log(KS_LOG_DEBUG, "Message type is '%s'\n", message->type);
+
+	return KS_STATUS_SUCCESS;
+
+ failure:
+	ks_dht2_message_deinit(message);
+	return KS_STATUS_FAIL;
+}
+
+/**
+ *
+ */
+KS_DECLARE(ks_status_t) ks_dht2_message_deinit(ks_dht2_message_t *message)
+{
+	ks_assert(message);
+
+	message->type[0] = '\0';
+	message->transactionid_length = 0;
+	if (message->data) {
+		ben_free(message->data);
+		message->data = NULL;
+	}
+
+	return KS_STATUS_SUCCESS;
+}
+
+
+/* For Emacs:
+ * Local Variables:
+ * mode:c
+ * indent-tabs-mode:t
+ * tab-width:4
+ * c-basic-offset:4
+ * End:
+ * For VIM:
+ * vim:set softtabstop=4 shiftwidth=4 tabstop=4 noet:
+ */

libs/libks/src/dht/ks_dht_message.h

+#ifndef KS_DHT_MESSAGE_H
+#define KS_DHT_MESSAGE_H
+
+#include "ks.h"
+
+KS_BEGIN_EXTERN_C
+
+#define KS_DHT_MESSAGE_TRANSACTIONID_MAX_SIZE 20
+#define KS_DHT_MESSAGE_TYPE_MAX_SIZE 20
+
+typedef struct ks_dht2_message_s ks_dht2_message_t;
+struct ks_dht2_message_s {
+	ks_pool_t *pool;
+	struct bencode *data;
+    uint8_t transactionid[KS_DHT_MESSAGE_TRANSACTIONID_MAX_SIZE];
+	ks_size_t transactionid_length;
+	char type[KS_DHT_MESSAGE_TYPE_MAX_SIZE];
+};
+
+KS_DECLARE(ks_status_t) ks_dht2_message_alloc(ks_dht2_message_t **message, ks_pool_t *pool);
+KS_DECLARE(ks_status_t) ks_dht2_message_prealloc(ks_dht2_message_t *message, ks_pool_t *pool);
+KS_DECLARE(ks_status_t) ks_dht2_message_free(ks_dht2_message_t *message);
+
+KS_DECLARE(ks_status_t) ks_dht2_message_init(ks_dht2_message_t *message, const uint8_t *buffer, ks_size_t buffer_length);
+KS_DECLARE(ks_status_t) ks_dht2_message_deinit(ks_dht2_message_t *message);
+
+KS_END_EXTERN_C
+
+#endif /* KS_DHT_MESSAGE_H */
+
+/* For Emacs:
+ * Local Variables:
+ * mode:c
+ * indent-tabs-mode:t
+ * tab-width:4
+ * c-basic-offset:4
+ * End:
+ * For VIM:
+ * vim:set softtabstop=4 shiftwidth=4 tabstop=4 noet:
+ */
+

libs/libks/src/dht/ks_dht_nodeid.c

@@ -37,6 +37,7 @@ KS_DECLARE(ks_status_t) ks_dht2_nodeid_free(ks_dht2_nodeid_t *nodeid)
 {
 	ks_assert(nodeid);
 
+	ks_dht2_nodeid_deinit(nodeid);
 	ks_pool_free(nodeid->pool, nodeid);
 
 	return KS_STATUS_SUCCESS;

libs/libks/test/testdht2.c

@@ -7,10 +7,10 @@
 #define TEST_DHT1_REGISTER_Y_BUFFER "d1:ad2:id20:12345678901234567890e1:q4:ping1:t2:421:y1:ze"
 #define TEST_DHT1_PROCESS_BUFFER "d1:ad2:id20:12345678901234567890e1:q4:ping1:t2:421:y1:qe"
 
-ks_status_t dht_z_callback(ks_dht2_t *dht, ks_sockaddr_t *raddr, uint8_t *transactionid, ks_size_t transactionid_len, struct bencode *message)
+ks_status_t dht_z_callback(ks_dht2_t *dht, ks_sockaddr_t *raddr, ks_dht2_message_t *message)
 {
 	diag("dht_z_callback\n");
-	ok(transactionid[0] == '4' && transactionid[1] == '2');
+	ok(message->transactionid[0] == '4' && message->transactionid[1] == '2');
 	return KS_STATUS_SUCCESS;
 }