diff --git a/Makefile.am b/Makefile.am index e92b5703..3d530de1 100644 --- a/Makefile.am +++ b/Makefile.am @@ -80,4 +80,4 @@ endif libdiagnostic_la_CPPFLAGS = -I$(top_srcdir)/source/diagnostic/include -I$(top_srcdir)/source/diagnostic/BbhmDiagIpPing -I$(top_srcdir)/source/dmltad -I$(top_srcdir)/source/TandDSsp libdiagnostic_la_DEPENDENCIES = $(EXTRA_DEPENDENCIES) libdiagnostic_la_LIBADD = $(libdiagnostic_la_DEPENDENCIES) -libdiagnostic_la_LDFLAGS = -lccsp_common +libdiagnostic_la_LDFLAGS = -lccsp_common -lmosquitto diff --git a/source/LatencyMeasurement/xNetDP/Makefile.am b/source/LatencyMeasurement/xNetDP/Makefile.am index b0897d6a..fb471554 100644 --- a/source/LatencyMeasurement/xNetDP/Makefile.am +++ b/source/LatencyMeasurement/xNetDP/Makefile.am @@ -21,4 +21,4 @@ bin_PROGRAMS = xNetDP xNetDP_CPPFLAGS = -I${PKG_CONFIG_SYSROOT_DIR}$(includedir)/rbus xNetDP_SOURCES = xNetDP.c -xNetDP_LDFLAGS = -lpcap -lrbus -lsyscfg -lm -lpthread +xNetDP_LDFLAGS = -lpcap -lrbus -lsyscfg -lm -lpthread -lmosquitto diff --git a/source/LatencyMeasurement/xNetDP/xNetDP.c b/source/LatencyMeasurement/xNetDP/xNetDP.c index c2fe3fe1..d32848af 100644 --- a/source/LatencyMeasurement/xNetDP/xNetDP.c +++ b/source/LatencyMeasurement/xNetDP/xNetDP.c @@ -42,6 +42,7 @@ #include #include #include +#include #include "syscfg/syscfg.h" #define ADD_MAX_SAMPLE 10 #define MAX_SAMPLE 50 @@ -74,6 +75,11 @@ pthread_mutex_t latency_report_lock = PTHREAD_MUTEX_INITIALIZER; #define TRUE 1 #define FALSE 0 #define BUF_SIZE 200 +#define MQTT_LOCAL_MQTT_BROKER_IP_ADDR "192.168.245.254" +#define MQTT_LOCAL_MQTT_BROKER_PORT_VAL 1883 +#define TCP_LAN_latency_TOPIC "local/tcplatency" +#define MQTT_KEEPALIVE_TIME 60 + enum ip_family { IPV4=0, @@ -1049,6 +1055,31 @@ void* LatencyReportThread(void* arg) #if 1 + +// Function that takes MAC + LAN latency (microseconds) +int send_latency_message(struct mosquitto *mosq, const char *mac, long long lan_latency_sec, long long lan_latency_usec) { + char payload[256]; + int rc =0; + + // JSON payload: you can format however you need + snprintf(payload, sizeof(payload), + "{\"mac\":\"%s\", \"lan_latency_sec\":%lld.%06lld}", mac, lan_latency_sec, lan_latency_usec); + + + // Publish message + rc = mosquitto_publish(mosq, NULL, TCP_LAN_latency_TOPIC, + strlen(payload), payload, + 1, false); + + if (rc == MOSQ_ERR_SUCCESS) { + dbg_log("Published: %s", payload); + } else { + dbg_log("Failed to publish message: %s", mosquitto_strerror(rc)); + } + + return rc; +} + void* LatencyReportThreadPerSession(void* arg) { // detach the current thread @@ -1058,6 +1089,8 @@ void* LatencyReportThreadPerSession(void* arg) int count = 0; char *str = NULL; char str1[1024]; + struct mosquitto *mosq; + int rc =0; str = (char*) malloc (MAX_REPORT_SIZE); if (str == NULL ) return NULL; @@ -1068,6 +1101,26 @@ void* LatencyReportThreadPerSession(void* arg) FILE *fp; //fp = fopen("LatencyReport.txt", "w+"); + mosquitto_lib_init(); + mosq = mosquitto_new("LatencyPublisher", true, NULL); + if (!mosq) { + dbg_log("Failed to create Mosquitto client"); + mosquitto_lib_cleanup(); + return NULL; + } + + rc = mosquitto_connect(mosq, + MQTT_LOCAL_MQTT_BROKER_IP_ADDR, + MQTT_LOCAL_MQTT_BROKER_PORT_VAL, + MQTT_KEEPALIVE_TIME); + + if (rc != MOSQ_ERR_SUCCESS) { + dbg_log("Failed to connect to MQTT broker: %s", mosquitto_strerror(rc)); + + mosquitto_destroy(mosq); + mosquitto_lib_cleanup(); + return NULL; + } while(1) { sleep(5); @@ -1081,6 +1134,15 @@ void* LatencyReportThreadPerSession(void* arg) if(hashArray[i].bComputed == TRUE) { tempCount = snprintf(str1,sizeof(str1),"%s,%u,%lld.%lld,%lld.%06lld|",hashArray[i].mac,hashArray[i].TcpInfo[INDEX_SYN].th_seq,hashArray[i].latency_sec,hashArray[i].latency_usec,hashArray[i].Lan_latency_sec,hashArray[i].Lan_latency_usec); + //Send LAN side Latency Notification to HCM module + rc = send_latency_message(mosq, hashArray[i].mac, + hashArray[i].Lan_latency_sec, hashArray[i].Lan_latency_usec); + + if (rc != MOSQ_ERR_SUCCESS) { + dbg_log("MQTT publish failed for MAC %s: %s", + hashArray[i].mac, mosquitto_strerror(rc)); + } + if(tempCount) { if((byteCount+tempCount) < MAX_REPORT_SIZE) @@ -1142,6 +1204,9 @@ void* LatencyReportThreadPerSession(void* arg) free(str); str=NULL; } + mosquitto_disconnect(mosq); + mosquitto_destroy(mosq); + mosquitto_lib_cleanup(); // exit the current thread //pthread_exit(NULL); }