added wan and cpe-service state in metadata

src/ParodusInternal.h

@@ -50,6 +50,8 @@
 /*                                   Macros                                   */
 /*----------------------------------------------------------------------------*/
 
+#define PARODUS_FREE(__x__) if(__x__ != NULL) { free((void*)(__x__)); __x__ = NULL;} else {printf("Trying to free null pointer\n");}
+
 #define UNUSED(x) (void )(x)
 #define NANO_SOCKET_SEND_TIMEOUT_MS                     2000
 #define NANO_SOCKET_RCV_TIMEOUT_MS			500

src/config.c

@@ -40,6 +40,8 @@ pthread_mutex_t cloud_status_mut=PTHREAD_MUTEX_INITIALIZER;
 pthread_cond_t cloud_status_cond=PTHREAD_COND_INITIALIZER;
 
 char webpa_interface[64]={'\0'};
+char wan_state_cache[64]="Unknown";
+char cpe_service_state_cache[64]="unknown";
 
 static ParodusCfg parodusCfg;
 static unsigned int rsa_algorithms = 
@@ -103,6 +105,54 @@ char *get_cloud_status(void)
     return status;    
 }
 
+int read_persisted_cpe_service_state(char *buf, size_t buf_size)
+{
+    FILE *fp = fopen(CPE_SERVICE_STATE_FILE, "r");
+    if (fp == NULL)
+        return -1;
+
+    if (fgets(buf, (int)buf_size, fp) == NULL) {
+        ParodusError("File %s is empty, unable to get cpe_service_state\n", CPE_SERVICE_STATE_FILE);
+        fclose(fp);
+        return -1;
+    }
+    fclose(fp);
+
+    /* Strip trailing newline */
+    size_t len = strlen(buf);
+    if (len > 0 && buf[len-1] == '\n')
+        buf[len-1] = '\0';
+
+    if (strcmp(buf, "fully-manageable") == 0 ||
+        strcmp(buf, "operational") == 0 ||
+        strcmp(buf, "non-operational") == 0)
+    {
+        return 0;
+    }
+    else
+    {
+        ParodusError("File %s is having invalid data: %s\n", CPE_SERVICE_STATE_FILE, buf);
+        return -1;
+    }
+}
+
+void write_cpe_service_state_to_file(const char *state)
+{
+    FILE *fp = fopen(CPE_SERVICE_STATE_FILE, "w");
+    if (fp == NULL) {
+        ParodusError("Failed to write cpe_service_state to %s\n", CPE_SERVICE_STATE_FILE);
+        return;
+    }
+    if(state == NULL)
+    {
+        ParodusError("cpe_service_state is NULL, cannot write to %s\n", CPE_SERVICE_STATE_FILE);
+        fclose(fp);
+        return;
+    }
+    fprintf(fp, "%s\n", state);
+    fclose(fp);
+}
+
 const char *get_tok (const char *src, int delim, char *result, int resultsize)
 {
 	int i;
@@ -869,6 +919,18 @@ void setDefaultValuesToCfg(ParodusCfg *cfg)
 
 	cfg->cloud_status = CLOUD_STATUS_OFFLINE;
 	ParodusInfo("Default cloud_status is %s\n", cfg->cloud_status);
+
+	parStrncpy(cfg->wan_state, "Unknown", sizeof(cfg->wan_state));
+	parStrncpy(cfg->cpe_service_state, "unknown", sizeof(cfg->cpe_service_state));
+
+	/* Restore persisted cpe-service-state from /tmp file to handle parodus restart/crash scenarios */
+	{
+		char persisted[64] = {0};
+		if (read_persisted_cpe_service_state(persisted, sizeof(persisted)) == 0) {
+			parStrncpy(cfg->cpe_service_state, persisted, sizeof(cfg->cpe_service_state));
+			ParodusInfo("Restored persisted cpe_service_state: %s\n", cfg->cpe_service_state);
+		}
+	}    
 }
 
 void loadParodusCfg(ParodusCfg * config,ParodusCfg *cfg)
@@ -1139,3 +1201,38 @@ char *getWebpaInterface(void)
 		return webpa_interface;
 }
 
+void setWanState(const char *value)
+{
+    pthread_mutex_lock(&config_mut);
+	parStrncpy(get_parodus_cfg()->wan_state, (value != NULL && strlen(value) != 0) ? value : "Unknown", sizeof(get_parodus_cfg()->wan_state));    
+    pthread_mutex_unlock(&config_mut);
+}
+
+void setCpeServiceState(const char *value)
+{
+    pthread_mutex_lock(&config_mut);
+    parStrncpy(get_parodus_cfg()->cpe_service_state, (value != NULL && strlen(value) != 0) ? value : "unknown", sizeof(get_parodus_cfg()->cpe_service_state));
+    pthread_mutex_unlock(&config_mut);
+}
+
+const char *getWanState(void)
+{
+	#ifdef ENABLE_WEBCFGBIN	
+		pthread_mutex_lock (&config_mut);	
+		parStrncpy(wan_state_cache, get_parodus_cfg()->wan_state, sizeof(wan_state_cache));
+		pthread_mutex_unlock (&config_mut);
+	#endif
+		ParodusPrint("wan_state:%s\n", wan_state_cache);
+    return wan_state_cache;
+}
+
+const char *getCpeServiceState(void)
+{
+	#ifdef ENABLE_WEBCFGBIN	
+		pthread_mutex_lock (&config_mut);	
+		parStrncpy(cpe_service_state_cache, get_parodus_cfg()->cpe_service_state, sizeof(cpe_service_state_cache));
+		pthread_mutex_unlock (&config_mut);
+	#endif
+		ParodusPrint("cpe_service_state:%s\n", cpe_service_state_cache);
+    return cpe_service_state_cache;
+}

src/config.h

@@ -58,6 +58,9 @@ extern "C" {
 #define BOOT_RETRY_WAIT         "boot-time-retry-wait"
 #define WEBPA_INTERFACE_LABEL   "webpa-interface-label"
 #define WAN_IPV4_ADDRESS        "wan-ipv4-address"
+#define WAN_STATE               "wan-state"
+#define CPE_SERVICE_STATE       "cpe-service-state"
+#define CPE_SERVICE_STATE_FILE  "/tmp/parodus_cpe_service_state"
 
 #define PROTOCOL_VALUE 					"PARODUS-2.0"
 #define WEBPA_PATH_URL                  "/api/v2/device"
@@ -120,6 +123,8 @@ typedef struct
 	unsigned int boot_retry_wait;
     char webpa_interface_label[64];
     char wan_ipv4_address[64];
+    char wan_state[64];
+    char cpe_service_state[64];
 #ifdef FEATURE_DNS_QUERY
     char *record_jwt_file;
 #endif
@@ -154,8 +159,14 @@ char *get_token_application(void) ;
 void set_cloud_disconnect_reason(ParodusCfg *cfg, char *disconn_reason);
 void reset_cloud_disconnect_reason(ParodusCfg *cfg);
 char *getWebpaInterface(void);
+const char *getCpeServiceState(void);
+const char *getWanState(void);
+void setWanState(const char *state);
+void setCpeServiceState(const char *state);
 void set_cloud_status(char *status);
 char *get_cloud_status(void);
+int read_persisted_cpe_service_state(char *buf, size_t buf_size);
+void write_cpe_service_state_to_file(const char *state);
 int get_parodus_init();
 /**
  * parse a webpa url. Extract the server address, the port

src/connection.c

@@ -834,7 +834,9 @@ int createNopollConnection(noPollCtx *ctx, server_list_t *server_list)
 	ParodusPrint("LastReasonStatus reset after successful connection\n");
 	setMessageHandlers();
     stop_conn_in_progress ();
+        lock_metadata_mutex();
         packMetaData();
+        unlock_metadata_mutex();
 	ParodusPrint("set cloud_status\n");
 	set_cloud_status(CLOUD_STATUS_ONLINE);
 	ParodusInfo("cloud_status set as %s after successful connection\n", get_cloud_status());

src/main.c

@@ -18,6 +18,7 @@
 #include <string.h>
 #include "stdlib.h"
 #include "config.h"
+#include "upstream.h"
 #include "auth_token.h"
 #include "connection.h"
 #include "conn_interface.h"
@@ -125,6 +126,9 @@ int main( int argc, char **argv)
     regXmidtSendDataMethod();
     #endif
     setDefaultValuesToCfg(cfg);
+	#ifdef ENABLE_WEBCFGBIN
+	subscribeWanStateEvent();
+    #endif	
     if (0 != parseCommandLine(argc,argv,cfg)) {
 		abort();
 	}

src/upstream.c

@@ -33,7 +33,7 @@
 /*----------------------------------------------------------------------------*/
 /*                                   Macros                                   */
 /*----------------------------------------------------------------------------*/
-#define METADATA_COUNT 					12
+#define METADATA_COUNT 					14
 #define PARODUS_SERVICE_NAME			"parodus"
 /*----------------------------------------------------------------------------*/
 /*                            File Scoped Variables                           */
@@ -49,6 +49,18 @@ pthread_mutex_t nano_mut=PTHREAD_MUTEX_INITIALIZER;
 
 pthread_cond_t nano_con=PTHREAD_COND_INITIALIZER;
 
+static pthread_mutex_t metadata_mut=PTHREAD_MUTEX_INITIALIZER;
+
+void lock_metadata_mutex(void)
+{
+    pthread_mutex_lock(&metadata_mut);
+}
+
+void unlock_metadata_mutex(void)
+{
+    pthread_mutex_unlock(&metadata_mut);
+}
+
 UpStreamMsg * get_global_UpStreamMsgQ(void)
 {
     return UpStreamMsgQ;
@@ -95,7 +107,9 @@ void packMetaData()
             {WEBPA_PROTOCOL, get_parodus_cfg()->webpa_protocol},
             {WEBPA_UUID,get_parodus_cfg()->webpa_uuid},
             {WEBPA_INTERFACE, getWebpaInterface()},
-            {PARTNER_ID, get_parodus_cfg()->partner_id}
+            {PARTNER_ID, get_parodus_cfg()->partner_id},
+            {WAN_STATE, getWanState()},
+            {CPE_SERVICE_STATE, getCpeServiceState()}
         };
     const data_t metapack = {METADATA_COUNT, meta_pack};
 
@@ -213,6 +227,65 @@ void *handle_upstream()
     return 0;
 }
 
+void extractAndSetCpeServiceState(const char *dest)
+{
+    if (dest == NULL) 
+        return;
+    const char *prefix = "event:device-status/";
+    const char *p = strstr(dest, prefix);
+    if (!p) 
+        return;
+
+    p += strlen(prefix);
+
+    // Find first and second '/'
+    const char *first = strchr(p, '/');
+    if (!first) 
+        return;
+
+    const char *second = strchr(first + 1, '/');
+    if (!second) 
+        return;
+
+    char state[64] = {0};
+
+    // Extract substring between slashes
+    size_t len = second - first - 1;
+    if (len == 0 || len >= sizeof(state)) 
+        return;
+
+    memcpy(state, first+1, len);
+    state[len] = '\0';
+
+    // Validate state
+    const char *new_state = "unknown";
+    if (strcmp(state, "fully-manageable") == 0 ||
+        strcmp(state, "operational") == 0 ||
+        strcmp(state, "non-operational") == 0)
+    {
+        new_state = state;
+    }
+    else
+    {
+        ParodusError("Invalid CPE service state received: %s\n", state);
+        return;
+    }
+
+
+    if (strcmp(getCpeServiceState(), new_state) != 0)
+    {
+        setCpeServiceState(new_state);
+        write_cpe_service_state_to_file(new_state);
+        lock_metadata_mutex();
+        packMetaData();
+        unlock_metadata_mutex();        
+        ParodusInfo("metadata cpe_service_state set to : %s\n", getCpeServiceState());        
+    }
+    else
+    {
+        ParodusPrint("CPE service state is already %s, no update needed\n", new_state);
+    }
+}
 
 void *processUpstreamMessage()
 {		
@@ -329,6 +402,7 @@ void *processUpstreamMessage()
 		    if(msg->u.event.transaction_uuid != NULL) {
 			    ParodusInfo("transaction_uuid in event: %s\n", msg->u.event.transaction_uuid);
 		    }	    
+                    extractAndSetCpeServiceState(msg->u.event.dest);
                     partners_t *partnersList = NULL;
                     int j = 0;
 
@@ -619,9 +693,11 @@ int sendUpstreamMsgToServer(void **resp_bytes, size_t resp_size)
 	bool close_retry = false;
 	int sendRetStatus = 1;
 	//appending response with metadata 			
+	lock_metadata_mutex();
 	if(metaPackSize > 0)
 	{
 	   	encodedSize = appendEncodedData( &appendData, *resp_bytes, resp_size, metadataPack, metaPackSize );
+	   	unlock_metadata_mutex();
 	   	ParodusPrint("metadata appended upstream response %s\n", (char *)appendData);
 	   	ParodusPrint("encodedSize after appending :%zu\n", encodedSize);
 
@@ -652,7 +728,8 @@ int sendUpstreamMsgToServer(void **resp_bytes, size_t resp_size)
 		appendData =NULL;
 	}
 	else
-	{		
+	{
+		unlock_metadata_mutex();
 		ParodusError("Failed to send upstream as metadata packing is not successful\n");
 		sendRetStatus = 1;
 	}

src/upstream.h

@@ -26,6 +26,9 @@
 
 #include <pthread.h>
 #include <wrp-c.h>
+#ifdef ENABLE_WEBCFGBIN
+#include <rbus.h>
+#endif
 #ifdef __cplusplus
 extern "C" {
 #endif
@@ -45,6 +48,8 @@ typedef struct UpStreamMsg__
 /*----------------------------------------------------------------------------*/
 
 void packMetaData();
+void lock_metadata_mutex(void);
+void unlock_metadata_mutex(void);
 void *handle_upstream();
 void *processUpstreamMessage();
 void registerRBUSlistener();
@@ -56,11 +61,15 @@ void set_global_UpStreamMsgQ(UpStreamMsg * UpStreamQ);
 #ifdef WAN_FAILOVER_SUPPORTED
 int subscribeCurrentActiveInterfaceEvent();
 #endif
+#ifdef ENABLE_WEBCFGBIN
+int subscribeWanStateEvent();
+void wanStateEventHandler(rbusHandle_t handle, rbusEvent_t const* event, rbusEventSubscription_t* subscription);
+#endif
 UpStreamMsg * get_global_UpStreamMsgQ(void);
 pthread_cond_t *get_global_nano_con(void);
 pthread_mutex_t *get_global_nano_mut(void);
 void clear_metadata();
-
+void extractAndSetCpeServiceState(const char *dest);
 #ifdef __cplusplus
 }
 #endif