Deepstream 5.0 Subscription Implementation

aws_protocol_adaptor/device_client/aws_iot_config.h

@@ -36,6 +36,8 @@
 
 #define AWS_IOT_MAX_SEND_INTERVAL_SEC 120 ///< Have to yield every this amount of time before time-out disconnect
 #define MAX_SUBSCRIPTIONS 50
+#define NVDS_MSGAPI_VERSION 2.0
+#define NVDS_MSGAPI_PROTOCOL "AWS"
 
 // Thing Shadow specific configs
 #define SHADOW_MAX_SIZE_OF_RX_BUFFER (AWS_IOT_MQTT_RX_BUF_LEN + 1)                                       ///< Maximum size of the SHADOW buffer to store the received Shadow message, including terminating NULL byte.

aws_protocol_adaptor/device_client/aws_nvmsgbroker.c

@@ -33,10 +33,12 @@
 #include "aws_config_parser.h"
 #include "nvds_msgapi.h"
 #include "aws_nvmsgbroker.h"
+#include <openssl/sha.h>
 
 NvDsMsgApiHandle (*nvds_msgapi_connect_ptr)(char *connection_str, nvds_msgapi_connect_cb_t connect_cb, char *config_path);
 NvDsMsgApiErrorType (*nvds_msgapi_send_ptr)(NvDsMsgApiHandle conn, char *topic, const uint8_t *payload, size_t nbuf);
 NvDsMsgApiErrorType (*nvds_msgapi_disconnect_ptr)(NvDsMsgApiHandle h_ptr);
+NvDsMsgApiErrorType (*nvds_msgapi_connection_signature_ptr)(char *connection_str, char *config_path, char *output_str, int max_len);
 static GMutex thread_mutex;
 static GQueue *work_queue;
 static struct timespec last_send_time_stamp; // this is to make sure we send or yield frequent enough so we do not get disconnected.
@@ -353,3 +355,80 @@ void nvds_msgapi_do_work(NvDsMsgApiHandle h_ptr)
 	last_send_time_stamp = current_time_stamp;
 	return;
 }
+
+char *nvds_msgapi_getversion()
+{
+	return (char *)NVDS_MSGAPI_VERSION;
+}
+
+char *nvds_msgapi_get_protocol_name()
+{
+	return (char *)NVDS_MSGAPI_PROTOCOL;
+}
+
+bool is_valid_connection_str(char *connection_str, char *&burl, char *&bport)
+{
+	if (connection_str == NULL)
+	{
+		IOT_ERROR("connection string cant be NULL");
+		return false;
+	}
+
+	char conn_str[] = connection_str;
+	int i = 0;
+
+	char *token = strtok(conn_str, ";");
+	char *data[2];
+
+	while (token)
+	{
+		data[i++] = token;
+		token = strtok(NULL, ";");
+	}
+	for (i = 0; i < 2; i++)
+	{
+		printf("%s\n", data[i]);
+	}
+	char *burl = data[0];
+	char *bport = data[1];
+
+	if (burl == "" || bport == "")
+	{
+		IOT_ERROR("connection string is invalid. hostname or port is empty\n");
+		return false;
+	}
+	return true;
+}
+
+char *generate_sha256_hash(char *str)
+{
+	unsigned char hashval[SHA256_DIGEST_LENGTH];
+	int len = SHA256_DIGEST_LENGTH * 2 + 1;
+	char res[len];
+	SHA256_CTX sha256;
+	SHA256_Init(&sha256);
+	SHA256_Update(&sha256, str, strlen(str));
+	SHA256_Final(hashval, &sha256);
+	for (int i = 0; i < SHA256_DIGEST_LENGTH; i++)
+	{
+		res + (i * 2), "%02x", hashval[i];
+	}
+	return res;
+}
+
+NvDsMsgApiErrorType nvds_msgapi_connection_signature(char *broker_str, char *cfg, char *output_str, int max_len)
+{
+	strcpy(output_str, "");
+	if (broker_str == NULL || cfg == NULL)
+	{
+		IOT_ERROR("nvds_msgapi_connection_signature: broker_str or cfg path cant be NULL\n");
+		return NVDS_MSGAPI_ERR;
+	}
+	char *burl = "", *bport = "";
+	if (!is_valid_connection_str(broker_str, burl, bport))
+		return NVDS_MSGAPI_ERR;
+
+	char *aws_connection_signature = generate_sha256_hash(burl + bport);
+	strcpy(output_str, aws_connection_signature);
+	return NVDS_MSGAPI_OK;
+}

aws_protocol_adaptor/device_client/aws_nvmsgbroker.h

@@ -43,3 +43,4 @@ NvDsMsgApiErrorType nvds_msgapi_send(NvDsMsgApiHandle conn, char *topic, const u
 NvDsMsgApiErrorType nvds_msgapi_send_async(NvDsMsgApiHandle h_ptr, char *topic, const uint8_t *payload, size_t nbuf, nvds_msgapi_send_cb_t send_callback, void *user_ptr);
 NvDsMsgApiErrorType nvds_msgapi_subscribe(NvDsMsgApiHandle h_ptr, char **topics, int num_topics, nvds_msgapi_subscribe_request_cb_t cb, void *user_ctx);
 void nvds_msgapi_do_work(NvDsMsgApiHandle h_ptr);
+NvDsMsgApiErrorType nvds_msgapi_connection_signature(char *broker_str, char *cfg, char *output_str, int max_len);