To fix memory leak in upstream event message flow

Backoff

CHANGELOG.md

@@ -5,8 +5,19 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/)
 and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html).
 
 ## [Unreleased]
-- Refactor of connection code (simplification).
+- Refactored connection.c and updated corresponding unit tests
 - Additional `/cloud-status` and `/cloud-disconnect` fields.
+- Switched from nanomsg (Release 1.1.2) to NNG (Release v1.0.1)
+- Memory leak fixes
+- Changed connection logic (connection.c) for retries, and added unit test
+- Partner-id comparison made case insensitive
+- Reverted from NNG to nanomag (v1.1.2)
+- reverted temporary CMake reference to https://github.com/bill1600/seshat
+- Added log for time difference of parodus connect time and boot time
+- added NULL check for device mac id in upstream retrieve message handling
+- backoff retry to include find_servers in loop (connection.c)
+- backoff max is max count not max delay
+- Fixed memory leak in upstream event message flow
 
 ## [1.0.1] - 2018-07-18
 ### Added
@@ -20,7 +31,6 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 - Configurable security flag and port.
 - Default HTTP port to 80 and HTTPS port to 443.
 - Updates to how `nopoll` is called have been implemented.
-- Refactored connection.c and updated corresponding unit tests
 
 ### Fixed
 - 64 bit pointer fixes (#144, #145).

CMakeLists.txt

@@ -23,6 +23,7 @@ set(INSTALL_DIR ${CMAKE_CURRENT_BINARY_DIR}/_install)
 set(PREFIX_DIR ${CMAKE_CURRENT_BINARY_DIR}/_prefix)
 set(INCLUDE_DIR ${INSTALL_DIR}/include)
 set(INCLUDE_UCRESOLV ${PREFIX_DIR}/ucresolv/src/ucresolv/include)
+set(PATCHES_DIR ${CMAKE_CURRENT_SOURCE_DIR}/patches)
 set(LIBRARY_DIR ${INSTALL_DIR}/lib)
 set(LIBRARY_DIR64 ${INSTALL_DIR}/lib64)
 set(COMMON_LIBRARY_DIR ${INSTALL_DIR}/lib/${CMAKE_LIBRARY_ARCHITECTURE})
@@ -71,7 +72,7 @@ if (NOT BUILD_YOCTO)
 ExternalProject_Add(trower-base64
     PREFIX ${CMAKE_CURRENT_BINARY_DIR}/_prefix/trower-base64
     GIT_REPOSITORY https://github.com/Comcast/trower-base64.git
-    GIT_TAG "fbb9440ae2bc1118866baefcea7ff814f16613dd"
+    GIT_TAG "v1.1.1"
     CMAKE_ARGS += -DCMAKE_INSTALL_PREFIX=${INSTALL_DIR} -DBUILD_TESTING=OFF
 )
 add_library(libtrower-base64 STATIC SHARED IMPORTED)
@@ -83,7 +84,7 @@ add_dependencies(libtrower-base64 trower-base64)
 ExternalProject_Add(nopoll
     PREFIX ${PREFIX_DIR}/nopoll
     GIT_REPOSITORY https://github.com/Comcast/nopoll.git
-    GIT_TAG "739b740683b0615e5999bdeea756cbd593b7f1cc"
+    GIT_TAG "1.0.1"
     CONFIGURE_COMMAND COMMAND <SOURCE_DIR>/autogen.sh --prefix=${PREFIX}
                                                      --includedir=${INCLUDE_DIR}
                                                      --libdir=${LIBRARY_DIR}
@@ -99,7 +100,7 @@ add_dependencies(libnopoll nopoll)
 ExternalProject_Add(nanomsg
     PREFIX ${CMAKE_CURRENT_BINARY_DIR}/_prefix/nanomsg
     GIT_REPOSITORY https://github.com/nanomsg/nanomsg.git
-    GIT_TAG "0c1aa2b288f6b167dbafe7e29c20e6fc7e71c000"
+    GIT_TAG "1.1.2"
     CMAKE_ARGS += -DCMAKE_INSTALL_PREFIX=${INSTALL_DIR}
 )
 add_library(libnanomsg STATIC SHARED IMPORTED)
@@ -112,7 +113,7 @@ ExternalProject_Add(cJSON
     #PREFIX ${PREFIX_DIR}/cJSON
     PREFIX ${CMAKE_CURRENT_BINARY_DIR}/_prefix/cJSON
     GIT_REPOSITORY https://github.com/DaveGamble/cJSON.git
-    GIT_TAG "aafb64a1c549b7b927e339df6d35b1d5059dc235"
+    GIT_TAG "v1.7.8"
     CMAKE_ARGS += -DCMAKE_INSTALL_PREFIX=${INSTALL_DIR} -DBUILD_TESTING=OFF
 )
 add_library(libcJSON STATIC SHARED IMPORTED)
@@ -124,7 +125,7 @@ add_dependencies(libcJSON cJSON)
 ExternalProject_Add(msgpack
     PREFIX ${CMAKE_CURRENT_BINARY_DIR}/_prefix/msgpack
     GIT_REPOSITORY https://github.com/msgpack/msgpack-c.git
-    GIT_TAG "7a98138f27f27290e680bf8fbf1f8d1b089bf138"
+    GIT_TAG "cpp-3.1.1"
     CMAKE_ARGS += -DCMAKE_INSTALL_PREFIX=${INSTALL_DIR}
                   -DMSGPACK_ENABLE_CXX=OFF
                   -DMSGPACK_BUILD_EXAMPLES=OFF
@@ -138,7 +139,7 @@ add_dependencies(libmsgpack msgpack)
 ExternalProject_Add(cimplog
     PREFIX ${CMAKE_CURRENT_BINARY_DIR}/_prefix/cimplog
     GIT_REPOSITORY https://github.com/Comcast/cimplog.git
-    GIT_TAG "8a5fb3c2f182241d17f5342bea5b7688c28cd1fd"
+    GIT_TAG "1.0.1"
     CMAKE_ARGS += -DCMAKE_INSTALL_PREFIX=${INSTALL_DIR} -DBUILD_TESTING=OFF
 )
 add_library(libcimplog STATIC SHARED IMPORTED)
@@ -151,7 +152,7 @@ ExternalProject_Add(wrp-c
 	DEPENDS trower-base64 msgpack cimplog
     PREFIX ${CMAKE_CURRENT_BINARY_DIR}/_prefix/wrp-c
     GIT_REPOSITORY https://github.com/Comcast/wrp-c.git
-    GIT_TAG "c2bc9be9dad6b0e4f7f9138f3d1074702ace7976"
+    GIT_TAG "1.0.1"
     CMAKE_ARGS += -DCMAKE_INSTALL_PREFIX=${INSTALL_DIR}
                   -DMSGPACK_ENABLE_CXX=OFF
                   -DMSGPACK_BUILD_EXAMPLES=OFF
@@ -163,22 +164,6 @@ ExternalProject_Add(wrp-c
 add_library(libwrp-c STATIC SHARED IMPORTED)
 add_dependencies(libwrp-c wrp-c)
 
-# libparodus external dependency
-#-------------------------------------------------------------------------------
-ExternalProject_Add(libparodus
-    DEPENDS trower-base64 msgpack nanomsg wrp-c
-    PREFIX ${CMAKE_CURRENT_BINARY_DIR}/_prefix/libparodus
-    GIT_REPOSITORY https://github.com/Comcast/libparodus.git
-    GIT_TAG "a7615d69bd5859e337f467fb7e33182da970a5de"
-    CMAKE_ARGS += -DCMAKE_INSTALL_PREFIX=${INSTALL_DIR} -DBUILD_TESTING=OFF
-    		  -DMAIN_PROJ_BUILD=ON
-    		  -DMAIN_PROJ_LIB_PATH=${LIBRARY_DIR}
-    		  -DMAIN_PROJ_LIB64_PATH=${LIBRARY_DIR64}
-    		  -DMAIN_PROJ_COMMON_PATH=${COMMON_LIBRARY_DIR}
-    		  -DMAIN_PROJ_INCLUDE_PATH=${INCLUDE_DIR}
-)
-add_library(liblibparodus STATIC SHARED IMPORTED)
-add_dependencies(liblibparodus libparodus)
 
 if (ENABLE_SESHAT)
 # libseshat external dependency
@@ -186,8 +171,8 @@ if (ENABLE_SESHAT)
 ExternalProject_Add(libseshat
     DEPENDS cJSON trower-base64 msgpack nanomsg wrp-c
     PREFIX ${CMAKE_CURRENT_BINARY_DIR}/_prefix/libseshat
-    GIT_REPOSITORY https://github.com/comcast/seshat.git
-    GIT_TAG "470f8d5e9457755028aae6da65d3df9b62a53942"
+    GIT_REPOSITORY https://github.com/Comcast/seshat.git
+    GIT_TAG "1.0.1"
     CMAKE_ARGS += -DCMAKE_INSTALL_PREFIX=${INSTALL_DIR} -DBUILD_TESTING=OFF
     	          -DMAIN_PROJ_BUILD=ON
     	          -DMAIN_PROJ_LIB_PATH=${LIBRARY_DIR}
@@ -206,7 +191,7 @@ endif (ENABLE_SESHAT)
 ExternalProject_Add(cjwt
     PREFIX ${CMAKE_CURRENT_BINARY_DIR}/_prefix/cjwt
     GIT_REPOSITORY https://github.com/Comcast/cjwt.git
-    GIT_TAG "ddd077f0f00407f9657934492b252ee5cfcde535"
+    GIT_TAG "1.0.1"
     CMAKE_ARGS += -DCMAKE_INSTALL_PREFIX=${INSTALL_DIR} -DBUILD_TESTING=OFF
 )
 add_library(libcjwt STATIC SHARED IMPORTED)
@@ -218,7 +203,7 @@ if (FEATURE_DNS_QUERY)
 ExternalProject_Add(ucresolv
     PREFIX ${CMAKE_CURRENT_BINARY_DIR}/_prefix/ucresolv
     GIT_REPOSITORY https://github.com/Comcast/libucresolv.git
-    GIT_TAG "b58d49e165208791f84b44e3c079b1b4ef6d5c9d"
+    GIT_TAG "1.0.0"
     CMAKE_ARGS += -DCMAKE_INSTALL_PREFIX=${INSTALL_DIR}
 )
 add_library(libucresolv STATIC SHARED IMPORTED)

src/ParodusInternal.h

@@ -106,7 +106,8 @@ typedef struct {
 
 //--- Used in connection.c for backoff delay timer
 typedef struct {
-  int max_delay;
+  int count;
+  int max_count;
   int delay;
 } backoff_timer_t;
 

src/config.c

@@ -693,8 +693,8 @@ void setDefaultValuesToCfg(ParodusCfg *cfg)
     
     parStrncpy(cfg->webpa_uuid, "1234567-345456546",sizeof(cfg->webpa_uuid));
     ParodusPrint("cfg->webpa_uuid is :%s\n", cfg->webpa_uuid);
-    cfg->crud_config_file = strdup("parodus_cfg.json");
-	ParodusPrint("Default crud_config_file is %s\n", cfg->crud_config_file);
+    cfg->crud_config_file = NULL;
+	
 	cfg->cloud_status = CLOUD_STATUS_OFFLINE;
 	ParodusInfo("Default cloud_status is %s\n", cfg->cloud_status);
 }

src/conn_interface.c

@@ -51,6 +51,7 @@
 /*----------------------------------------------------------------------------*/
 /*                            File Scoped Variables                           */
 /*----------------------------------------------------------------------------*/
+bool g_shutdown  = false;
 
 /*----------------------------------------------------------------------------*/
 /*                             Function Prototypes                            */
@@ -168,10 +169,14 @@ void createSocketConnection(void (* initKeypress)())
             }
             createNopollConnection(ctx);
         }
-       } while(!get_close_retry());
+       } while(!get_close_retry() && !g_shutdown);
 
     close_and_unref_connection(get_global_conn());
     nopoll_ctx_unref(ctx);
     nopoll_cleanup_library();
 }
 
+void shutdownSocketConnection(void) {
+   g_shutdown = true;
+}
+

src/conn_interface.h

@@ -45,6 +45,7 @@ extern UpStreamMsg *UpStreamMsgQ;
  *  and creates the intial connection and manages the connection wait, close mechanisms.
  */
 void createSocketConnection(void (* initKeypress)());
+void shutdownSocketConnection(void);
    
 #ifdef __cplusplus
 }

src/connection.c

@@ -192,19 +192,20 @@ int check_timer_expired (expire_timer_t *timer, long timeout_ms)
 }
 
 //--------------------------------------------------------------------
-void init_backoff_timer (backoff_timer_t *timer, int max_delay)
+void init_backoff_timer (backoff_timer_t *timer, int max_count)
 {
-  timer->max_delay = max_delay;
+  timer->count = 1;
+  timer->max_count = max_count;
   timer->delay = 1;
 }
 
 int update_backoff_delay (backoff_timer_t *timer)
 {
-  if (timer->delay < timer->max_delay)
+  if (timer->count < timer->max_count) {
+    timer->count += 1;
     timer->delay = timer->delay + timer->delay + 1;
     // 3,7,15,31 ..
-  if (timer->delay > timer->max_delay)
-    timer->delay = timer->max_delay;
+  }
   return timer->delay;
 }  
 
@@ -503,14 +504,10 @@ int connect_and_wait (create_connection_ctx_t *ctx)
 // a) success, or
 // b) need to requery dns
 int keep_trying_to_connect (create_connection_ctx_t *ctx, 
-	int max_retry_sleep,
-	int query_dns_status)
+	backoff_timer_t *backoff_timer)
 {
-    backoff_timer_t backoff_timer;
     int rtn;
     
-    init_backoff_timer (&backoff_timer, max_retry_sleep);
-
     while (true)
     {
       rtn = connect_and_wait (ctx);
@@ -518,10 +515,9 @@ int keep_trying_to_connect (create_connection_ctx_t *ctx,
         return true;
       if (rtn == CONN_WAIT_ACTION_RETRY) // if redirected or build_headers
         continue;
-      backoff_delay (&backoff_timer); // 3,7,15,31 ..
+      backoff_delay (backoff_timer); // 3,7,15,31 ..
       if (rtn == CONN_WAIT_RETRY_DNS)
-        if (query_dns_status < 0)
-          return false;  //find_server again
+        return false;  //find_server again
       // else retry
     }
 }
@@ -536,8 +532,11 @@ int keep_trying_to_connect (create_connection_ctx_t *ctx,
 int createNopollConnection(noPollCtx *ctx)
 {
   create_connection_ctx_t conn_ctx;
-  int max_retry_sleep;
+  int max_retry_count;
   int query_dns_status;
+  struct timespec connect_time,*connectTimePtr;
+  connectTimePtr = &connect_time;
+  backoff_timer_t backoff_timer;
   
   if(ctx == NULL) {
         return nopoll_false;
@@ -547,14 +546,15 @@ int createNopollConnection(noPollCtx *ctx)
 	ParodusInfo("Received reboot_reason as:%s\n", get_parodus_cfg()->hw_last_reboot_reason);
 	ParodusInfo("Received reconnect_reason as:%s\n", reconnect_reason);
 	
-	max_retry_sleep = (int) get_parodus_cfg()->webpa_backoff_max;
-	ParodusPrint("max_retry_sleep is %d\n", max_retry_sleep );
+	max_retry_count = (int) get_parodus_cfg()->webpa_backoff_max;
+	ParodusPrint("max_retry_count is %d\n", max_retry_count );
   
 	conn_ctx.nopoll_ctx = ctx;
 	init_expire_timer (&conn_ctx.connect_timer);
 	init_header_info (&conn_ctx.header_info);
 	set_extra_headers (&conn_ctx, false);
         set_server_list_null (&conn_ctx.server_list);
+        init_backoff_timer (&backoff_timer, max_retry_count);
   
 	while (true)
 	{
@@ -562,7 +562,7 @@ int createNopollConnection(noPollCtx *ctx)
 	  if (query_dns_status == FIND_INVALID_DEFAULT)
 		return nopoll_false;
 	  set_current_server (&conn_ctx);
-	  if (keep_trying_to_connect (&conn_ctx, max_retry_sleep, query_dns_status))
+	  if (keep_trying_to_connect (&conn_ctx, &backoff_timer))
 		break;
 	  // retry dns query
 	}
@@ -579,6 +579,11 @@ int createNopollConnection(noPollCtx *ctx)
 	get_parodus_cfg()->cloud_status = CLOUD_STATUS_ONLINE;
 	ParodusInfo("cloud_status set as %s after successful connection\n", get_parodus_cfg()->cloud_status);
 
+	if(get_parodus_cfg()->boot_time != 0) {
+		getCurrentTime(connectTimePtr);
+		ParodusInfo("connect_time-diff-boot_time=%d\n", connectTimePtr->tv_sec - get_parodus_cfg()->boot_time);
+	}
+
 	free_extra_headers (&conn_ctx);
         free_header_info (&conn_ctx.header_info);
         free_server_list (&conn_ctx.server_list);

src/downstream.c

@@ -75,6 +75,7 @@ void listenerOnMessage(void * msg, size_t msgSize)
                 case WRP_MSG_TYPE__AUTH:
                 {
                     ParodusInfo("Authorization Status received with Status code :%d\n", message->u.auth.status);
+		    wrp_free_struct(message);
                     break;
                 }
 

src/main.c

@@ -103,7 +103,7 @@ static void sig_handler(int sig)
 	{
 		signal(SIGINT, sig_handler); /* reset it to this function */
 		ParodusInfo("SIGINT received!\n");
-		exit(0);
+		shutdownSocketConnection();
 	}
 	else if ( sig == SIGUSR1 ) 
 	{
@@ -132,7 +132,7 @@ static void sig_handler(int sig)
 	else 
 	{
 		ParodusInfo("Signal %d received!\n", sig);
-		exit(0);
+		shutdownSocketConnection();
 	}
 	
 }

src/partners_check.c

@@ -93,13 +93,18 @@ int validate_partner_id(wrp_msg_t *msg, partners_t **partnerIds)
                 {
                     for(j = 0; j<partnersList->count; j++)
                     {
-                        ParodusPrint("partnersList->partner_ids[%lu] = %s\n",j, partnersList->partner_ids[j]);
-                        if(strcmp(partnersList->partner_ids[j], msg->u.event.partner_ids->partner_ids[i]) == 0)
-                        {
-                            ParodusInfo("partner_id match found\n");
-                            matchFlag = 1;
-                            break;
-                        }
+			if(NULL != partnersList->partner_ids[j]) {
+		                ParodusPrint("partnersList->partner_ids[%lu] = %s\n",j, partnersList->partner_ids[j]);
+				ParodusPrint("msg->u.event.partner_ids->partner_ids[%lu] = %s\n",i, msg->u.event.partner_ids->partner_ids[i]);
+		                if(strcasecmp(partnersList->partner_ids[j], msg->u.event.partner_ids->partner_ids[i]) == 0)
+		                {
+		                    ParodusInfo("partner_id match found\n");
+		                    matchFlag = 1;
+		                    break;
+		                }
+			}
+			else
+				ParodusError("partner Id in partnersList is NULL but count is not 0");
                     }
                     /* Commandline input partner_ids matched with partner_ids from request */
                     if(matchFlag == 1)
@@ -153,13 +158,19 @@ int validate_partner_id(wrp_msg_t *msg, partners_t **partnerIds)
                 {
                     for(j = 0; j<partnersList->count; j++)
                     {
-                        ParodusPrint("partnersList->partner_ids[%lu] = %s\n",j, partnersList->partner_ids[j]);
-                        if(strcmp(partnersList->partner_ids[j], msg->u.req.partner_ids->partner_ids[i]) == 0)
-                        {
-                            ParodusInfo("partner_id match found\n");
-                            matchFlag = 1;
-                            break;
-                        }
+			if(NULL != partnersList->partner_ids[j]) 
+			{
+		                ParodusPrint("partnersList->partner_ids[%lu] = %s\n",j, partnersList->partner_ids[j]);
+				ParodusPrint("msg->u.req.partner_ids->partner_ids[%lu] = %s\n",i, msg->u.req.partner_ids->partner_ids[i]);
+		                if(strcasecmp(partnersList->partner_ids[j], msg->u.req.partner_ids->partner_ids[i]) == 0)
+		                {
+		                    ParodusInfo("partner_id match found\n");
+		                    matchFlag = 1;
+		                    break;
+		                }
+			}
+			else
+				ParodusError("partner Id in partnersList is NULL but count is not 0");
                     }
                 }
 
@@ -171,7 +182,7 @@ int validate_partner_id(wrp_msg_t *msg, partners_t **partnerIds)
                     {
                         for(j=0; j<partnersList->count; j++)
                         {
-                            if(partnersList->partner_ids[j] != NULL)
+                            if(NULL != partnersList->partner_ids[j])
                             {
                                 free(partnersList->partner_ids[j]);
                             }
@@ -191,7 +202,7 @@ int validate_partner_id(wrp_msg_t *msg, partners_t **partnerIds)
         {
             for(j=0; j<partnersList->count; j++)
             {
-                if(partnersList->partner_ids[j] != NULL)
+                if(NULL != partnersList->partner_ids[j])
                 {
                     free(partnersList->partner_ids[j]);
                 }

src/upstream.c

@@ -198,6 +198,7 @@ void *processUpstreamMessage()
     int matchFlag = 0;
     int status = -1;
     char *serviceName = NULL;
+    char *macId = NULL;
     char *destService, *destApplication =NULL;
     char *sourceService, *sourceApplication =NULL;
     int sendStatus =-1;
@@ -293,6 +294,7 @@ void *processUpstreamMessage()
                 {
                     ParodusInfo(" Received upstream event data: dest '%s'\n", msg->u.event.dest);
                     partners_t *partnersList = NULL;
+		    int j = 0;
 
                     int ret = validate_partner_id(msg, &partnersList);
                     if(ret == 1)
@@ -321,6 +323,18 @@ void *processUpstreamMessage()
                     {
                         sendUpstreamMsgToServer(&message->msg, message->len);
                     }
+		    if(partnersList != NULL)
+                    {
+                        for(j=0; j<(int)partnersList->count; j++)
+                        {
+                            if(NULL != partnersList->partner_ids[j])
+                            {
+                                free(partnersList->partner_ids[j]);
+                            }
+                        }
+                        free(partnersList);
+                    }
+                    partnersList = NULL;
                 }
                 else
                 {
@@ -335,6 +349,7 @@ void *processUpstreamMessage()
 						ParodusInfo(" Received upstream data with MsgType: %d dest: '%s' transaction_uuid: %s status: %d\n",msgType, msg->u.crud.dest, msg->u.crud.transaction_uuid, msg->u.crud.status );
 						if(WRP_MSG_TYPE__RETREIVE == msgType && msg->u.crud.dest !=NULL && msg->u.crud.source != NULL)
 						{
+                                                        macId = wrp_get_msg_element(WRP_ID_ELEMENT__ID, msg, DEST);
 							destService = wrp_get_msg_element(WRP_ID_ELEMENT__SERVICE, msg, DEST);
 							destApplication = wrp_get_msg_element(WRP_ID_ELEMENT__APPLICATION, msg, DEST);
 							sourceService = wrp_get_msg_element(WRP_ID_ELEMENT__SERVICE, msg, SOURCE);
@@ -343,48 +358,55 @@ void *processUpstreamMessage()
 								Expecting dest format as mac:xxxxxxxxxxxx/parodus/cloud-status
 								Parse dest field and check destService is "parodus" and destApplication is "cloud-status"
 							*/
-							if(destService != NULL && destApplication != NULL && strcmp(destService,"parodus")== 0 && strcmp(destApplication,"cloud-status")== 0)
-							{
-								retrieve_msg = ( wrp_msg_t *)malloc( sizeof( wrp_msg_t ) );
-								memset(retrieve_msg, 0, sizeof(wrp_msg_t));
-								retrieve_msg->msg_type = msg->msg_type;
-								retrieve_msg->u.crud.transaction_uuid = strdup(msg->u.crud.transaction_uuid);
-								retrieve_msg->u.crud.source = strdup(msg->u.crud.source);
-								retrieve_msg->u.crud.dest = strdup(msg->u.crud.dest);
-								addCRUDmsgToQueue(retrieve_msg);
-							}
-							else if(sourceService != NULL && sourceApplication != NULL && strcmp(sourceService,"parodus")== 0 && strcmp(sourceApplication,"cloud-status")== 0 && strncmp(msg->u.crud.dest,"mac:", 4)==0)
-							{
-								/*  Handle cloud-status retrieve response here to send it to registered client
-									Expecting src format as mac:xxxxxxxxxxxx/parodus/cloud-status and dest as mac:
-									Parse src field and check sourceService is "parodus" and sourceApplication is "cloud-status"
-								*/
-								serviceName = wrp_get_msg_element(WRP_ID_ELEMENT__SERVICE, msg, DEST);
-								if ( serviceName != NULL)
+							if(macId != NULL)
+						    	{
+								if(destService != NULL && destApplication != NULL && strcmp(destService,"parodus")== 0 && strcmp(destApplication,"cloud-status")== 0)
 								{
-									//Send Client cloud-status response back to registered client
-									ParodusInfo("Sending cloud-status response to %s client\n",serviceName);
-									sendStatus=sendMsgtoRegisteredClients(serviceName,(const char **)&message->msg,message->len);
-									if(sendStatus ==1)
+									retrieve_msg = ( wrp_msg_t *)malloc( sizeof( wrp_msg_t ) );
+									memset(retrieve_msg, 0, sizeof(wrp_msg_t));
+									retrieve_msg->msg_type = msg->msg_type;
+									retrieve_msg->u.crud.transaction_uuid = strdup(msg->u.crud.transaction_uuid);
+									retrieve_msg->u.crud.source = strdup(msg->u.crud.source);
+									retrieve_msg->u.crud.dest = strdup(msg->u.crud.dest);
+									addCRUDmsgToQueue(retrieve_msg);
+								}
+								else if(sourceService != NULL && sourceApplication != NULL && strcmp(sourceService,"parodus")== 0 && strcmp(sourceApplication,"cloud-status")== 0 && strncmp(msg->u.crud.dest,"mac:", 4)==0)
+								{
+									/*  Handle cloud-status retrieve response here to send it to registered client
+										Expecting src format as mac:xxxxxxxxxxxx/parodus/cloud-status and dest as mac:
+										Parse src field and check sourceService is "parodus" and sourceApplication is "cloud-status"
+									*/
+									serviceName = wrp_get_msg_element(WRP_ID_ELEMENT__SERVICE, msg, DEST);
+									if ( serviceName != NULL)
 									{
-										ParodusInfo("Send upstreamMsg successfully to registered client %s\n", serviceName);
+										//Send Client cloud-status response back to registered client
+										ParodusInfo("Sending cloud-status response to %s client\n",serviceName);
+										sendStatus=sendMsgtoRegisteredClients(serviceName,(const char **)&message->msg,message->len);
+										if(sendStatus ==1)
+										{
+											ParodusInfo("Send upstreamMsg successfully to registered client %s\n", serviceName);
+										}
+										else
+										{
+											ParodusError("Failed to send upstreamMsg to registered client %s\n", serviceName);
+										}
+										free(serviceName);
+										serviceName = NULL;
 									}
 									else
 									{
-										ParodusError("Failed to send upstreamMsg to registered client %s\n", serviceName);
+										ParodusError("serviceName is NULL,not sending cloud-status response to client\n");
 									}
-									free(serviceName);
-									serviceName = NULL;
 								}
 								else
 								{
-									ParodusError("serviceName is NULL,not sending cloud-status response to client\n");
+									ParodusInfo("sendUpstreamMsgToServer \n");
+									sendUpstreamMsgToServer(&message->msg, message->len);
 								}
 							}
 							else
 							{
-								ParodusInfo("sendUpstreamMsgToServer \n");
-								sendUpstreamMsgToServer(&message->msg, message->len);
+								ParodusError("MAC is null, not handling retrieve wrp message \n");
 							}
 							if(sourceService !=NULL)
 							{

tests/CMakeLists.txt

@@ -18,7 +18,7 @@ set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wall -W  -g -fprofile-arcs -ftest-cove
 set (CMAKE_EXE_LINKER_FLAGS "${CMAKE_EXE_LINKER_FLAGS} -fprofile-arcs -ftest-coverage -O0")
 set (PARODUS_COMMON_SRC ../src/string_helpers.c ../src/mutex.c ../src/time.c ../src/config.c ../src/spin_thread.c ../src/token.c)
 set (PARODUS_COMMON_LIBS gcov -lcunit -lcimplog -lwrp-c
- -luuid -lpthread -lmsgpackc -lnopoll -lnanomsg
+ -luuid -lmsgpackc -lnopoll -lnanomsg -lpthread
  -Wl,--no-as-needed -lcjson -lcjwt -ltrower-base64
  -lssl -lcrypto -lrt -lm)
 
@@ -71,26 +71,6 @@ add_test(NAME test_nopoll_helpers COMMAND ${MEMORY_CHECK} ./test_nopoll_helpers)
 add_executable(test_nopoll_helpers test_nopoll_helpers.c ../src/nopoll_helpers.c)
 target_link_libraries (test_nopoll_helpers -Wl,--no-as-needed -lrt -lcmocka -lcimplog -lnopoll)
 
-#-------------------------------------------------------------------------------
-#   libpd_test
-#-------------------------------------------------------------------------------
-add_executable (libpd_test libpd_test.c ../src/string_helpers.c) 
-
-target_link_libraries (libpd_test
-                       cunit
-                       -llibparodus
-                       -lwrp-c
-		       -luuid
-                       -lmsgpackc
-                       -ltrower-base64
-                       -lnanomsg
-                       -lcimplog 
-                       -lm
-                       -lpthread
-                       -lrt)
-if (ENABLE_SESHAT)
-target_link_libraries (libpd_test -llibseshat)
-endif (ENABLE_SESHAT)
 #-------------------------------------------------------------------------------
 #   test_time
 #-------------------------------------------------------------------------------
@@ -230,14 +210,20 @@ target_link_libraries (test_crud_internal -lcmocka ${PARODUS_COMMON_LIBS} )
 #-------------------------------------------------------------------------------
 add_test(NAME test_upstream COMMAND ${MEMORY_CHECK} ./test_upstream)
 add_executable(test_upstream test_upstream.c ../src/upstream.c ../src/close_retry.c ../src/string_helpers.c)
-target_link_libraries (test_upstream -lcmocka ${PARODUS_COMMON_LIBS} )
+target_link_libraries (test_upstream -lcmocka gcov -lcunit -lcimplog
+ -lwrp-c -luuid -lpthread -lmsgpackc -lnopoll
+ -Wl,--no-as-needed -lcjson -lcjwt -ltrower-base64
+ -lssl -lcrypto -lrt -lm)
 
 #-------------------------------------------------------------------------------
 #   test_downstream
 #-------------------------------------------------------------------------------
 add_test(NAME test_downstream COMMAND ${MEMORY_CHECK} ./test_downstream)
 add_executable(test_downstream test_downstream.c ../src/downstream.c ../src/string_helpers.c)
-target_link_libraries (test_downstream -lcmocka ${PARODUS_COMMON_LIBS} )
+target_link_libraries (test_downstream -lcmocka gcov -lcunit -lcimplog
+ -lwrp-c -luuid -lpthread -lmsgpackc -lnopoll
+ -Wl,--no-as-needed -lcjson -lcjwt -ltrower-base64
+ -lssl -lcrypto -lrt -lm)
 
 #-------------------------------------------------------------------------------
 #   test_downstream_more
@@ -362,5 +348,5 @@ set(SIMPLE_SRC ${SIMPLE_SRC} ../src/seshat_interface_stub.c)
 endif (ENABLE_SESHAT)
 add_executable(simple ${SIMPLE_SRC}) 
 
-target_link_libraries (simple ${PARODUS_CONN_LIBS} ${PARODUS_COMMON_LIBS}  gcov -lnopoll -lnanomsg )
+target_link_libraries (simple ${PARODUS_CONN_LIBS} ${PARODUS_COMMON_LIBS}  gcov -lnopoll )
 endif (INTEGRATION_TESTING)

tests/libpd_test.c

@@ -1,409 +0,0 @@
-/**
- *  Copyright 2010-2016 Comcast Cable Communications Management, LLC
- *
- *  Licensed under the Apache License, Version 2.0 (the "License");
- *  you may not use this file except in compliance with the License.
- *  You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- *  Unless required by applicable law or agreed to in writing, software
- *  distributed under the License is distributed on an "AS IS" BASIS,
- *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- *  See the License for the specific language governing permissions and
- *  limitations under the License.
- */
-#include <sys/types.h>
-#include <sys/stat.h>
-#include <sys/wait.h>
-#include <stdarg.h>
-#include <stdlib.h>
-#include <stdio.h>
-#include <string.h>
-#include <ctype.h>
-#include <fcntl.h>
-#include <unistd.h>
-#include <CUnit/Basic.h>
-#include <stdbool.h>
-
-#include <libparodus.h>
-#include <pthread.h>
-#include "../src/ParodusInternal.h"
-
-#define SEND_EVENT_MSGS 1
-
-//#define TCP_URL(ip) "tcp://" ip
-
-#define GOOD_CLIENT_URL "tcp://127.0.0.1:6667"
-//#define PARODUS_URL "ipc:///tmp/parodus_server.ipc"
-#define GOOD_PARODUS_URL "tcp://127.0.0.1:6666"
-//#define CLIENT_URL "ipc:///tmp/parodus_client.ipc"
-
-
-static void initEndKeypressHandler();
-static void *endKeypressHandlerTask();
-
-static pthread_t endKeypressThreadId;
-
-static const char *service_name = "iot";
-//static const char *service_name = "config";
-
-static bool no_mock_send_only_test = false;
-
-static libpd_instance_t test_instance;
-
-// libparodus functions to be tested
-extern int flush_wrp_queue (uint32_t delay_ms);
-extern int connect_receiver (const char *rcv_url);
-extern int connect_sender (const char *send_url);
-extern void shutdown_socket (int *sock);
-
-extern bool is_auth_received (void);
-extern int libparodus_receive__ (wrp_msg_t **msg, uint32_t ms);
-
-// libparodus_log functions to be tested
-extern int get_valid_file_num (const char *file_name, const char *date);
-extern int get_last_file_num_in_dir (const char *date, const char *log_dir);
-
-extern const char *wrp_queue_name;
-extern const char *parodus_url;
-extern const char *client_url;
-extern volatile int keep_alive_count;
-extern volatile int reconnect_count;
-
-void show_src_dest_payload (char *src, char *dest, void *payload, size_t payload_size)
-{
-	size_t i;
-	char *payload_str = (char *) payload;
-	printf (" SOURCE:  %s\n", src);
-	printf (" DEST  :  %s\n", dest);
-	printf (" PAYLOAD: ");
-	for (i=0; i<payload_size; i++)
-		putchar (payload_str[i]);
-	putchar ('\n');
-}
-
-void show_wrp_req_msg (struct wrp_req_msg *msg)
-{
-	show_src_dest_payload (msg->source, msg->dest, msg->payload, msg->payload_size);
-}
-
-void show_wrp_event_msg (struct wrp_event_msg *msg)
-{
-	show_src_dest_payload (msg->source, msg->dest, msg->payload, msg->payload_size);
-}
-
-void show_wrp_msg (wrp_msg_t *wrp_msg)
-{
-	printf ("Received WRP Msg type %d\n", wrp_msg->msg_type);
-	if (wrp_msg->msg_type == WRP_MSG_TYPE__REQ) {
-		show_wrp_req_msg (&wrp_msg->u.req);
-		return;
-	}
-	if (wrp_msg->msg_type == WRP_MSG_TYPE__EVENT) {
-		show_wrp_event_msg (&wrp_msg->u.event);
-		return;
-	}
-	return;
-}
-
-void send_reply (wrp_msg_t *wrp_msg)
-{
-	size_t i;
-	size_t payload_size = wrp_msg->u.req.payload_size;
-	char *payload = (char *) wrp_msg->u.req.payload;
-	char *temp;
-	// swap source and dest
-	temp = wrp_msg->u.req.source;
-	wrp_msg->u.req.source = wrp_msg->u.req.dest;
-	wrp_msg->u.req.dest = temp;
-	// Alter the payload
-	for (i=0; i<payload_size; i++)
-		payload[i] = tolower (payload[i]);
-	libparodus_send (test_instance, wrp_msg);
-}
-
-char *new_str (const char *str)
-{
-	char *buf = malloc (strlen(str) + 1);
-	if (NULL == buf)
-		return NULL;
-	parStrncpy(buf, str, (strlen(str)+1));
-	return buf;
-}
-
-void insert_number_into_buf (char *buf, unsigned num)
-{
-	char *pos = strrchr (buf, '#');
-	if (NULL == pos)
-		return;
-	while (true) {
-		*pos = (num%10) + '0';
-		num /= 10;
-		if (pos <= buf)
-			break;
-		pos--;
-		if (*pos != '#')
-			break;
-	}
-}
-
-int send_event_msg (const char *src, const char *dest, 
-	const char *payload, unsigned event_num)
-{
-	int rtn = 0;
-	char *payload_buf;
-	wrp_msg_t *new_msg;
-
-#ifndef SEND_EVENT_MSGS
-	return 0;
-#endif
-	new_msg = malloc (sizeof (wrp_msg_t));
-	if (NULL == new_msg)
-		return -1;
-	printf ("Making event msg\n");
-	new_msg->msg_type = WRP_MSG_TYPE__EVENT;
-	new_msg->u.event.source = new_str (src);
-	new_msg->u.event.dest = new_str (dest);
-	new_msg->u.event.headers = NULL;
-	new_msg->u.event.metadata = NULL;
-	payload_buf = new_str (payload);
-	insert_number_into_buf (payload_buf, event_num);
-	new_msg->u.event.payload = (void*) payload_buf;
-	new_msg->u.event.payload_size = strlen (payload) + 1;
-	printf ("Sending event msg %u\n", event_num);
-	rtn = libparodus_send (test_instance, new_msg);
-	//printf ("Freeing event msg\n");
-	wrp_free_struct (new_msg);
-	//printf ("Freed event msg\n");
-	return rtn;
-}
-
-int send_event_msgs (unsigned *msg_num, unsigned *event_num, int count)
-{
-	int i;
-	unsigned msg_num_mod;
-
-#ifndef SEND_EVENT_MSGS
-	return 0;
-#endif
-	if (NULL != msg_num) {
-		(*msg_num)++;
-		msg_num_mod = (*msg_num) % 3;
-		if (msg_num_mod != 0)
-			return 0;
-	}
-	for (i=0; i<count; i++) {
-		(*event_num)++;
-		if (send_event_msg ("---LIBPARODUS---", "---ParodusService---",
-			"---EventMessagePayload####", *event_num) != 0)
-			return -1;
-	}
-	return 0;
-}
-
-int get_msg_num (const char *msg)
-{
-	int num = -1;
-	bool found_pound = false;
-	int i;
-	char c;
-
-	for (i=0; (c=msg[i]) != 0; i++)
-	{
-		if (!found_pound) {
-			if (c == '#')
-				found_pound = true;
-			continue;
-		}
-		if ((c>='0') && (c<='9')) {
-			if (num == -1)
-				num = c - '0';
-			else
-				num = 10*num + (c - '0');
-		}
-	}
-	return num;
-}
-
-static int flush_queue_count = 0;
-
-void qfree (void * msg)
-{
-	flush_queue_count++;
-	free (msg);
-}
-
-void delay_ms(unsigned int secs, unsigned int msecs)
-{
-  struct timespec ts;
-  ts.tv_sec = (time_t) secs;
-  ts.tv_nsec = (long) msecs * 1000000L;
-  nanosleep (&ts, NULL);
-}
-
-void dbg_log_err (const char *fmt, ...)
-{
-    char errbuf[100];
-
-    va_list arg_ptr;
-
-    va_start(arg_ptr, fmt);
-    vprintf(fmt, arg_ptr);
-    va_end(arg_ptr);
-
-    if( 0 == strerror_r (errno, errbuf, 100) ) {   
-        printf("LIBPD_TEST: %s\n", errbuf);
-    } else {   
-        printf("LIBPD_TEST: strerror_r returned failure!\n");
-    }
-}
-
-void wait_auth_received (void)
-{
-	if (!is_auth_received ()) {
-		printf ("Waiting for auth received\n");
-		sleep(1);
-	}
-	if (!is_auth_received ()) {
-		printf ("Waiting for auth received\n");
-		sleep(1);
-	}
-	CU_ASSERT (is_auth_received ());
-}
-
-void test_send_only (void)
-{
-	unsigned event_num = 0;
-
-        libpd_cfg_t cfg = {.service_name = service_name,
-                .receive = false, .keepalive_timeout_secs = 0};
-        CU_ASSERT (libparodus_init (&test_instance, &cfg) == 0);
-        CU_ASSERT (send_event_msgs (NULL, &event_num, 10) == 0);
-	CU_ASSERT (libparodus_shutdown (&test_instance) == 0);
-}
-
- void test_1(void)
-{
-	unsigned msgs_received_count = 0;
-	int rtn;
-	wrp_msg_t *wrp_msg;
-	unsigned event_num = 0;
-	unsigned msg_num = 0;
-        libpd_cfg_t cfg = {.service_name = service_name,
-                .receive = true, .keepalive_timeout_secs = 0};
-
-	if (no_mock_send_only_test) {
-		test_send_only ();
-		return;
-	}
-
-        cfg.parodus_url = GOOD_PARODUS_URL;
-        cfg.client_url = GOOD_CLIENT_URL;
-        CU_ASSERT (libparodus_init (&test_instance, &cfg) == 0);
-	printf ("LIBPD_TEST: libparodus_init successful\n");
-	initEndKeypressHandler ();
-
-	wait_auth_received ();
-
-	printf ("LIBPD_TEST: starting msg receive loop\n");
-	while (true) {
-		rtn = libparodus_receive (test_instance, &wrp_msg, 2000);
-		if (rtn == 1) {
-			printf ("LIBPD_TEST: Timed out waiting for msg\n");
-			if (msgs_received_count > 0)
-				if (send_event_msgs (&msg_num, &event_num, 5) != 0)
-					break;
-			continue;
-		}
-		if (rtn != 0)
-			break;
-		show_wrp_msg (wrp_msg);
-		msgs_received_count++;
-		if (wrp_msg->msg_type == WRP_MSG_TYPE__REQ)
-			send_reply (wrp_msg);
-		wrp_free_struct (wrp_msg);
-		if (send_event_msgs (&msg_num, &event_num, 5) != 0)
-			break;
-	}
-	printf ("Messages received %u\n", msgs_received_count);
-	CU_ASSERT (libparodus_shutdown (&test_instance) == 0);
-}
-
-/*
- * @brief To initiate end keypress handler
- */
-static void initEndKeypressHandler()
-{
-	int err = 0;
-	err = pthread_create(&endKeypressThreadId, NULL, endKeypressHandlerTask, NULL);
-	if (err != 0) 
-	{
-		libpd_log (LEVEL_ERROR, "Error creating End Keypress Handler thread\n");
-	}
-	else 
-	{
-		printf ("End Keypress handler Thread created successfully\n");
-		printf ("\n--->> Press <Enter> to shutdown the test. ---\n");
-	}
-}
-
-/*
- * @brief To handle End Keypress
- */
-static void *endKeypressHandlerTask()
-{
-	char inbuf[10];
-	memset(inbuf, 0, 10);
-	while (true) {
-		fgets (inbuf, 10, stdin);
-		if ((inbuf[0] != '\n') && (inbuf[0] != '\0')) {
-            printf ("endKeyPressHandler exiting\n");
-			break;
-		}
-	}
-	libparodus_close_receiver (test_instance);
-	return NULL;
-}
-
-
-void add_suites( CU_pSuite *suite )
-{
-    *suite = CU_add_suite( "libparodus tests", NULL, NULL );
-    CU_add_test( *suite, "Test 1", test_1 );
-}
-
-/*----------------------------------------------------------------------------*/
-/*                             External Functions                             */
-/*----------------------------------------------------------------------------*/
-int main( int argc, char **argv __attribute__((unused)) )
-{
-    unsigned rv = 1;
-    CU_pSuite suite = NULL;
-
-		if (argc > 1) {
-			const char *arg = argv[1];
-			if ((arg[0] == 's') || (arg[0] == 'S'))
-				no_mock_send_only_test = true;
-		}
-
-    if( CUE_SUCCESS == CU_initialize_registry() ) {
-        add_suites( &suite );
-
-        if( NULL != suite ) {
-            CU_basic_set_mode( CU_BRM_VERBOSE );
-            CU_basic_run_tests();
-            printf( "\n" );
-            CU_basic_show_failures( CU_get_failure_list() );
-            printf( "\n\n" );
-            rv = CU_get_number_of_tests_failed();
-        }
-
-        CU_cleanup_registry();
-    }
-
-    if( 0 != rv ) {
-        return 1;
-    }
-    return 0;
-}

tests/test_close_retry.c

@@ -71,8 +71,8 @@ void test_mutex_close_retry() {
 	ParodusInfo("Threads execution is completed, close_retry is: %d\n", close_retry);
 	assert_int_equal(close_retry, 1);
 
-	pthread_create(&thread[3], NULL, test_mutex_reset_close_retry, NULL);
-	pthread_join(thread[3], NULL);
+	pthread_create(&thread[2], NULL, test_mutex_reset_close_retry, NULL);
+	pthread_join(thread[2], NULL);
 
 	close_retry = get_close_retry();
 	ParodusInfo("close_retry reset to: %d\n", close_retry);

tests/test_connection.c

@@ -36,7 +36,7 @@ extern server_t *get_current_server (server_list_t *server_list);
 extern int parse_server_url (const char *full_url, server_t *server);
 extern void init_expire_timer (expire_timer_t *timer);
 extern int check_timer_expired (expire_timer_t *timer, long timeout_ms);
-extern void init_backoff_timer (backoff_timer_t *timer, int max_delay);
+extern void init_backoff_timer (backoff_timer_t *timer, int max_count);
 extern int update_backoff_delay (backoff_timer_t *timer);
 extern int init_header_info (header_info_t *header_info);
 extern void free_header_info (header_info_t *header_info);
@@ -51,8 +51,7 @@ extern int nopoll_connect (create_connection_ctx_t *ctx, int is_ipv6);
 extern int wait_connection_ready (create_connection_ctx_t *ctx);
 extern int connect_and_wait (create_connection_ctx_t *ctx);
 extern int keep_trying_to_connect (create_connection_ctx_t *ctx, 
-	int max_retry_sleep,
-	int query_dns_status);
+	backoff_timer_t *backoff_timer);
 
 
 /*----------------------------------------------------------------------------*/
@@ -117,10 +116,15 @@ nopoll_bool  nopoll_conn_wait_for_status_until_connection_ready (noPollConn * co
 	int timeout, int *status, char ** message)
 {
     UNUSED(conn); UNUSED(timeout);
-    *status = mock_wait_status;
+    if (mock_wait_status >= 1000) {
+      *status = mock_wait_status / 1000;
+      mock_wait_status = mock_wait_status % 1000;
+    } else {
+      *status = mock_wait_status;
+    }
     if ((NULL != mock_redirect) &&
-        ((mock_wait_status == 307) || (mock_wait_status == 302) || 
-         (mock_wait_status == 303)) )
+        ((*status == 307) || (*status == 302) || 
+         (*status == 303)) )
     {
       *message = malloc (strlen(mock_redirect) + 10);
       sprintf (*message, "Redirect:%s", mock_redirect);
@@ -297,12 +301,12 @@ void test_expire_timer()
 void test_backoff_delay_timer()
 {
   backoff_timer_t btimer;
-  init_backoff_timer (&btimer, 30);
+  init_backoff_timer (&btimer, 5);
   assert_int_equal (3, update_backoff_delay (&btimer));
   assert_int_equal (7, update_backoff_delay (&btimer));
   assert_int_equal (15, update_backoff_delay (&btimer));
-  assert_int_equal (30, update_backoff_delay (&btimer));
-  assert_int_equal (30, update_backoff_delay (&btimer));
+  assert_int_equal (31, update_backoff_delay (&btimer));
+  assert_int_equal (31, update_backoff_delay (&btimer));
 }
 
 
@@ -580,6 +584,11 @@ void test_wait_connection_ready ()
   expect_function_call (nopoll_conn_wait_for_status_until_connection_ready);
   assert_int_equal (wait_connection_ready (&ctx), WAIT_FAIL);
   
+  mock_wait_status = 503;
+  will_return (nopoll_conn_wait_for_status_until_connection_ready, nopoll_false);
+  expect_function_call (nopoll_conn_wait_for_status_until_connection_ready);
+  assert_int_equal (wait_connection_ready (&ctx), WAIT_FAIL);
+
   mock_wait_status = 307;
   mock_redirect = "mydns.mycom.net";
   will_return (nopoll_conn_wait_for_status_until_connection_ready, nopoll_false);
@@ -668,7 +677,18 @@ void test_connect_and_wait ()
   will_return (checkHostIp, 0);
   expect_function_call (checkHostIp);
   assert_int_equal (connect_and_wait (&ctx), CONN_WAIT_RETRY_DNS);
-  
+
+  mock_wait_status = 503;  
+  will_return (nopoll_conn_new_opts, &connection1);
+  expect_function_call (nopoll_conn_new_opts);
+  will_return (nopoll_conn_is_ok, nopoll_true);
+  expect_function_call (nopoll_conn_is_ok);
+  will_return (nopoll_conn_wait_for_status_until_connection_ready, nopoll_false);
+  expect_function_call (nopoll_conn_wait_for_status_until_connection_ready);
+  will_return (nopoll_conn_ref_count, 0);
+  expect_function_call (nopoll_conn_ref_count);
+  assert_int_equal (connect_and_wait (&ctx), CONN_WAIT_RETRY_DNS);
+
   Cfg.flags = 0;
   set_parodus_cfg(&Cfg);
 
@@ -771,6 +791,7 @@ void test_keep_trying ()
   create_connection_ctx_t ctx;
   noPollCtx test_nopoll_ctx;
   server_t test_server;
+  backoff_timer_t backoff_timer;
   ParodusCfg Cfg;
   char *test_extra_headers =
       "\r\nAuthorization: Bearer SER_MAC Fer23u948590 123567892366"
@@ -802,7 +823,8 @@ void test_keep_trying ()
   expect_function_call (nopoll_conn_is_ok);
   will_return (nopoll_conn_wait_for_status_until_connection_ready, nopoll_true);
   expect_function_call (nopoll_conn_wait_for_status_until_connection_ready);
-  rtn = keep_trying_to_connect (&ctx, 30, -1);
+  init_backoff_timer (&backoff_timer, 5);
+  rtn = keep_trying_to_connect (&ctx, &backoff_timer);
   assert_int_equal (rtn, true);
 
   test_server.allow_insecure = 0;
@@ -823,24 +845,42 @@ void test_keep_trying ()
   expect_function_call (nopoll_conn_tls_new);
   will_return (nopoll_conn_is_ok, nopoll_true);
   expect_function_call (nopoll_conn_is_ok);
-  mock_wait_status = 0;
   will_return (nopoll_conn_wait_for_status_until_connection_ready, nopoll_true);
   expect_function_call (nopoll_conn_wait_for_status_until_connection_ready);
-  rtn = keep_trying_to_connect (&ctx, 30, 0);
+  init_backoff_timer (&backoff_timer, 5);
+  rtn = keep_trying_to_connect (&ctx, &backoff_timer);
   assert_int_equal (rtn, true);
 
-  will_return (nopoll_conn_tls_new, NULL);
-  expect_function_call (nopoll_conn_tls_new);
-  will_return (checkHostIp, 0);
-  expect_function_call (checkHostIp);
   will_return (nopoll_conn_tls_new, &connection1);
   expect_function_call (nopoll_conn_tls_new);
   will_return (nopoll_conn_is_ok, nopoll_true);
   expect_function_call (nopoll_conn_is_ok);
-  will_return (nopoll_conn_wait_for_status_until_connection_ready, nopoll_true);
+  mock_wait_status = 302503;
+  mock_redirect = "https://mydns.mycom.net";
+  will_return (nopoll_conn_wait_for_status_until_connection_ready, nopoll_false);
   expect_function_call (nopoll_conn_wait_for_status_until_connection_ready);
-  rtn = keep_trying_to_connect (&ctx, 30, 0);
-  assert_int_equal (rtn, true);
+  will_return (nopoll_conn_ref_count, 0);
+  expect_function_call (nopoll_conn_ref_count);
+  will_return (nopoll_conn_tls_new, &connection1);
+  expect_function_call (nopoll_conn_tls_new);
+  will_return (nopoll_conn_is_ok, nopoll_true);
+  expect_function_call (nopoll_conn_is_ok);
+  will_return (nopoll_conn_wait_for_status_until_connection_ready, nopoll_false);
+  expect_function_call (nopoll_conn_wait_for_status_until_connection_ready);
+  will_return (nopoll_conn_ref_count, 0);
+  expect_function_call (nopoll_conn_ref_count);
+  init_backoff_timer (&backoff_timer, 5);
+  rtn = keep_trying_to_connect (&ctx, &backoff_timer);
+  assert_int_equal (rtn, false);
+
+  mock_wait_status = 0;
+  will_return (nopoll_conn_tls_new, NULL);
+  expect_function_call (nopoll_conn_tls_new);
+  will_return (checkHostIp, 0);
+  expect_function_call (checkHostIp);
+  init_backoff_timer (&backoff_timer, 5);
+  rtn = keep_trying_to_connect (&ctx, &backoff_timer);
+  assert_int_equal (rtn, false);
 
   mock_wait_status = 302;
   mock_redirect = "mydns.mycom.net";
@@ -853,7 +893,8 @@ void test_keep_trying ()
   expect_function_call (nopoll_conn_wait_for_status_until_connection_ready);
   will_return (nopoll_conn_ref_count, 0);
   expect_function_call (nopoll_conn_ref_count);
-  rtn = keep_trying_to_connect (&ctx, 30, -1);
+  init_backoff_timer (&backoff_timer, 5);
+  rtn = keep_trying_to_connect (&ctx, &backoff_timer);
   assert_int_equal (rtn, false);
 }
 

tests/test_heartBeatTimer.c

@@ -72,8 +72,8 @@ void test_mutexHeartBeatTimer() {
 	ParodusInfo("Threads execution is completed, heartBeatTimer is: %d\n", heartBeatTimer);
 	assert_int_equal(heartBeatTimer, 10);
 
-	pthread_create(&thread[3], NULL, test_mutexResetTimer, NULL);
-	pthread_join(thread[3], NULL);
+	pthread_create(&thread[2], NULL, test_mutexResetTimer, NULL);
+	pthread_join(thread[2], NULL);
 
 	heartBeatTimer = get_heartBeatTimer();
 	ParodusInfo("heartBeatTimer reset to: %d\n", heartBeatTimer);

tests/test_partners_check.c

@@ -60,6 +60,25 @@ void test_validate_partner_id_for_req()
     free(msg);
 }
 
+void test_validate_partner_id_for_req_case_insensitive()
+{
+    static partners_t partner_ids = {2,{"Shaw","Comcast"}};
+    wrp_msg_t *msg = (wrp_msg_t*) malloc(sizeof(wrp_msg_t));
+    memset(msg, 0, sizeof(wrp_msg_t));
+    msg->msg_type = WRP_MSG_TYPE__REQ;
+    msg->u.req.partner_ids = &partner_ids;
+    
+    ParodusCfg cfg;
+    memset(&cfg, 0, sizeof(ParodusCfg));
+    parStrncpy(cfg.partner_id, "comcast", sizeof(cfg.partner_id));
+    
+    will_return(get_parodus_cfg, (intptr_t)&cfg);
+    expect_function_call(get_parodus_cfg);
+    int ret = validate_partner_id(msg, NULL); 
+    assert_int_equal(ret, 1);
+    free(msg);
+}
+
 void test_validate_partner_id_for_req_listNULL()
 {
     wrp_msg_t *msg = (wrp_msg_t*) malloc(sizeof(wrp_msg_t));
@@ -134,6 +153,28 @@ void test_validate_partner_id_for_event()
     free(msg);
 }
 
+void test_validate_partner_id_for_event_case_insensitive()
+{
+    static partners_t partner_ids = {1,{"Comcast"}};
+    wrp_msg_t *msg = (wrp_msg_t*) malloc(sizeof(wrp_msg_t));
+    memset(msg, 0, sizeof(wrp_msg_t));
+    msg->msg_type = WRP_MSG_TYPE__EVENT;
+    msg->u.event.partner_ids = &partner_ids;
+    
+    ParodusCfg cfg;
+    memset(&cfg, 0, sizeof(ParodusCfg));
+    parStrncpy(cfg.partner_id, "*,comcast", sizeof(cfg.partner_id));
+    
+    will_return(get_parodus_cfg, (intptr_t)&cfg);
+    expect_function_call(get_parodus_cfg);
+    
+    partners_t *list = NULL;
+    int ret = validate_partner_id(msg, &list); 
+    assert_int_equal(ret, 1);
+    free(list);  
+    free(msg);
+}
+
 void test_validate_partner_id_for_event_listNULL()
 {
     wrp_msg_t *msg = (wrp_msg_t*) malloc(sizeof(wrp_msg_t));
@@ -218,10 +259,12 @@ int main(void)
 {
     const struct CMUnitTest tests[] = {
         cmocka_unit_test(test_validate_partner_id_for_req),
+	cmocka_unit_test(test_validate_partner_id_for_req_case_insensitive),
         cmocka_unit_test(test_validate_partner_id_for_req_listNULL),
         cmocka_unit_test(test_validate_partner_id_for_req_withoutId),
         cmocka_unit_test(err_validate_partner_id_for_req),
         cmocka_unit_test(test_validate_partner_id_for_event),
+	cmocka_unit_test(test_validate_partner_id_for_event_case_insensitive),
         cmocka_unit_test(test_validate_partner_id_for_event_listNULL),
         cmocka_unit_test(test_validate_partner_id_for_event_withoutId),
         cmocka_unit_test(err_validate_partner_id_for_event),

tests/test_upstream.c

@@ -762,6 +762,81 @@ void test_processUpstreamMsg_sendToClient()
     free(UpStreamMsgQ);
     UpStreamMsgQ = NULL;
 }
+void test_processUpstreamMessageNullCheck()
+{
+    numLoops = 1;
+    metaPackSize = 20;
+    UpStreamMsgQ = (UpStreamMsg *) malloc(sizeof(UpStreamMsg));
+    UpStreamMsgQ->msg = strdup("First Message");
+    UpStreamMsgQ->len = 13;
+    UpStreamMsgQ->next= NULL;
+    temp = (wrp_msg_t *) malloc(sizeof(wrp_msg_t));
+    memset(temp,0,sizeof(wrp_msg_t));
+    temp->msg_type = WRP_MSG_TYPE__RETREIVE;
+    temp->u.crud.dest = strdup("mac:14cfe2142xxx/parodus/cloud-status");
+    temp->u.crud.source = strdup("mac:14cfe2142xxx/config");
+    temp->u.crud.transaction_uuid = strdup("123");
+    will_return(wrp_to_struct, 12);
+    expect_function_call(wrp_to_struct);
+    expect_function_call(addCRUDmsgToQueue);
+    will_return(nn_freemsg, 0);
+    expect_function_call(nn_freemsg);
+    expect_function_call(wrp_free_struct);
+    processUpstreamMessage();
+    free(temp);
+    free(UpStreamMsgQ);
+    UpStreamMsgQ = NULL;
+}
+void err_processUpstreamMessageNullCheck()
+{
+    numLoops = 1;
+    metaPackSize = 20;
+    UpStreamMsgQ = (UpStreamMsg *) malloc(sizeof(UpStreamMsg));
+    UpStreamMsgQ->msg = strdup("First Message");
+    UpStreamMsgQ->len = 13;
+    UpStreamMsgQ->next= NULL;
+    temp = (wrp_msg_t *) malloc(sizeof(wrp_msg_t));
+    memset(temp,0,sizeof(wrp_msg_t));
+    temp->msg_type = WRP_MSG_TYPE__RETREIVE;
+    temp->u.crud.dest = strdup("mac:/parodus/cloud-status");
+    temp->u.crud.source = strdup("mac:14cfe2142xxx/config");
+    temp->u.crud.transaction_uuid = strdup("123");
+    will_return(wrp_to_struct, 12);
+    expect_function_call(wrp_to_struct);
+    will_return(nn_freemsg, 0);
+    expect_function_call(nn_freemsg);
+    expect_function_call(wrp_free_struct);
+	
+    processUpstreamMessage();
+    free(temp);
+    free(UpStreamMsgQ);
+	UpStreamMsgQ = NULL;
+}
+void err_processUpstreamMessageWithoutMac()
+{
+    numLoops = 1;
+    metaPackSize = 20;
+    UpStreamMsgQ = (UpStreamMsg *) malloc(sizeof(UpStreamMsg));
+    UpStreamMsgQ->msg = strdup("First Message");
+    UpStreamMsgQ->len = 13;
+    UpStreamMsgQ->next= NULL;
+    temp = (wrp_msg_t *) malloc(sizeof(wrp_msg_t));
+    memset(temp,0,sizeof(wrp_msg_t));
+    temp->msg_type = WRP_MSG_TYPE__RETREIVE;
+    temp->u.crud.dest = strdup("/parodus/cloud-status");
+    temp->u.crud.source = strdup("mac:14cfe2142xxx/config");
+    temp->u.crud.transaction_uuid = strdup("123");
+    will_return(wrp_to_struct, 12);
+    expect_function_call(wrp_to_struct);
+    will_return(nn_freemsg, 0);
+    expect_function_call(nn_freemsg);
+    expect_function_call(wrp_free_struct);
+	
+    processUpstreamMessage();
+    free(temp);
+    free(UpStreamMsgQ);
+    UpStreamMsgQ = NULL;
+}
 
 /*----------------------------------------------------------------------------*/
 /*                             External Functions                             */
@@ -795,6 +870,9 @@ int main(void)
         cmocka_unit_test(test_processUpstreamMsgCrud_nnfree),
         cmocka_unit_test(test_processUpstreamMsg_cloud_status),
         cmocka_unit_test(test_processUpstreamMsg_sendToClient),
+	cmocka_unit_test(test_processUpstreamMessageNullCheck),
+	cmocka_unit_test(err_processUpstreamMessageNullCheck),
+	cmocka_unit_test(err_processUpstreamMessageWithoutMac),
     };
 
     return cmocka_run_group_tests(tests, NULL, NULL);