Merge pull request #281 from sadhyama/parodus-interface

To correct the spelling 'webpa-inteface-used' in convey header

.travis.yml

@@ -26,7 +26,7 @@ install:
 script:
     - mkdir build
     - cd build
-    - cmake .. -DINTEGRATION_TESTING:BOOL=false -DDISABLE_VALGRIND:BOOL=true -DENABLE_SESHAT:BOOL=true -DFEATURE_DNS_QUERY:BOOL=true
+    - cmake .. -DINTEGRATION_TESTING:BOOL=false -DDISABLE_VALGRIND:BOOL=false -DENABLE_SESHAT:BOOL=true -DFEATURE_DNS_QUERY:BOOL=true
     - make
     - export ARGS="-V"
     - make test

CHANGELOG.md

@@ -17,6 +17,18 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 - 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
+- used mutex protection to make client list and nn_sends thread safe
+- put mutex lock into get_global_node
+- change svc alive from a thread to a function called every 30 sec from main
+- shut down tasks properly
+- fix memory leaks
+- Fixed memory leak in upstream event message flow
+- Fixed crash in CRUD request processing
+- Fixed issue on RETRIEVE respone processing
+- Enabled valgrind
+- Fixed main loop to keep calling svc_alive_task during a cloud disconnect and retry
+- change svc alive back to a separate thread.  Shut it down with pthread_cond_timedwait
+- Refactored Upsteam RETRIEVE flow
 
 ## [1.0.1] - 2018-07-18
 ### Added

CMakeLists.txt

@@ -191,7 +191,8 @@ endif (ENABLE_SESHAT)
 ExternalProject_Add(cjwt
     PREFIX ${CMAKE_CURRENT_BINARY_DIR}/_prefix/cjwt
     GIT_REPOSITORY https://github.com/Comcast/cjwt.git
-    GIT_TAG "1.0.1"
+    GIT_TAG "1b023c41bb2d6dbbf493c202ed81f06c84d5b51b"
+    #GIT_TAG "1.0.1"
     CMAKE_ARGS += -DCMAKE_INSTALL_PREFIX=${INSTALL_DIR} -DBUILD_TESTING=OFF
 )
 add_library(libcjwt STATIC SHARED IMPORTED)

src/ParodusInternal.h

@@ -50,7 +50,8 @@
 /*                                   Macros                                   */
 /*----------------------------------------------------------------------------*/
 #define UNUSED(x) (void )(x)
-#define NANOMSG_SOCKET_TIMEOUT_MSEC                      2000
+#define NANO_SOCKET_SEND_TIMEOUT_MS                     2000
+#define NANO_SOCKET_RCV_TIMEOUT_MS			500
 
 #ifndef TEST
 #define FOREVER()   1
@@ -133,6 +134,7 @@ typedef struct {
 /*----------------------------------------------------------------------------*/
 /*                            File Scoped Variables                           */
 /*----------------------------------------------------------------------------*/
+extern bool g_shutdown;
 extern ParodusMsg *ParodusMsgQ;
 int numLoops;
 /*----------------------------------------------------------------------------*/

src/client_list.c

@@ -30,6 +30,7 @@
 /*----------------------------------------------------------------------------*/
 static int numOfClients = 0;
 static reg_list_item_t * g_head = NULL;
+pthread_mutex_t client_mut=PTHREAD_MUTEX_INITIALIZER;
 
 /*----------------------------------------------------------------------------*/
 /*                             External functions                             */
@@ -37,9 +38,16 @@ static reg_list_item_t * g_head = NULL;
 
 reg_list_item_t * get_global_node(void)
 {
+    pthread_mutex_lock (&client_mut);
     return g_head;
 }
 
+void release_global_node (void)
+{
+    pthread_mutex_unlock (&client_mut);
+}
+
+
 int get_numOfClients()
 {
     return numOfClients;
@@ -52,11 +60,12 @@ int addToList( wrp_msg_t **msg)
     int rc = -1;
     int sock;
     int retStatus = -1;
+
     sock = nn_socket( AF_SP, NN_PUSH );
     ParodusPrint("sock created for adding entries to list: %d\n", sock);
     if(sock >= 0)
     {
-            int t = NANOMSG_SOCKET_TIMEOUT_MSEC;
+            int t = NANO_SOCKET_SEND_TIMEOUT_MS;
             rc = nn_setsockopt(sock, NN_SOL_SOCKET, NN_SNDTIMEO, &t, sizeof(t));
             if(rc < 0)
             {
@@ -214,7 +223,7 @@ int deleteFromList(char* service_name)
 			curr_node = NULL;
 			ParodusInfo("Deleted successfully and returning..\n");
 			numOfClients =numOfClients - 1;
-			ParodusPrint("numOfClients after delte is %d\n", numOfClients);
+			ParodusPrint("numOfClients after delete is %d\n", numOfClients);
 			return 0;
 		}
 		
@@ -225,6 +234,22 @@ int deleteFromList(char* service_name)
 	return -1;
 }
 
+void deleteAllClients (void)
+{
+  reg_list_item_t *next_node = NULL;
+
+  while (NULL != g_head)
+  {
+    next_node = g_head->next;
+    free (g_head);
+    g_head = next_node;
+  }
+  if (numOfClients > 0) {
+    ParodusInfo ("Deleted %d clients\n", numOfClients);
+    numOfClients = 0;
+  }
+}
+
 /*
 *@dest : Client destination to send message
 *@Msg:	Msg to send it to client (No free done here), user responsibilites to free the msg
@@ -243,6 +268,7 @@ int sendMsgtoRegisteredClients(char *dest,const char **Msg,size_t msgSize)
 		if( strcmp(dest, temp->service_name) == 0)
 		{
 			bytes = nn_send(temp->sock, *Msg, msgSize, 0);
+			release_global_node ();
 			ParodusInfo("sent downstream message to reg_client '%s'\n", temp->url);
 			ParodusPrint("downstream bytes sent:%d\n", bytes);
 			return 1;
@@ -250,5 +276,6 @@ int sendMsgtoRegisteredClients(char *dest,const char **Msg,size_t msgSize)
 		ParodusPrint("checking the next item in the list\n");
 		temp= temp->next;
 	}
+	release_global_node ();
 	return 0;
 }

src/client_list.h

@@ -50,9 +50,11 @@ int sendAuthStatus(reg_list_item_t *new_node);
 
 int deleteFromList(char* service_name);
 int get_numOfClients();
+void deleteAllClients (void);
 int sendMsgtoRegisteredClients(char *dest,const char **Msg,size_t msgSize);
 
 reg_list_item_t * get_global_node(void);
+void release_global_node (void);
 
 #ifdef __cplusplus
 }

src/config.h

@@ -42,7 +42,7 @@ extern "C" {
 #define BOOT_TIME               "boot-time"
 #define LAST_RECONNECT_REASON   "webpa-last-reconnect-reason"
 #define WEBPA_PROTOCOL          "webpa-protocol"
-#define WEBPA_INTERFACE         "webpa-inteface-used"
+#define WEBPA_INTERFACE         "webpa-interface-used"
 #define WEBPA_UUID              "webpa-uuid"
 #define WEBPA_URL               "webpa-url"
 #define WEBPA_PING_TIMEOUT      "webpa-ping-timeout"

src/conn_interface.c

@@ -52,12 +52,18 @@
 /*                            File Scoped Variables                           */
 /*----------------------------------------------------------------------------*/
 bool g_shutdown  = false;
+pthread_t upstream_tid;
+pthread_t upstream_msg_tid;
+pthread_t downstream_tid;
+pthread_t svc_alive_tid;
+pthread_t crud_tid;
 
 /*----------------------------------------------------------------------------*/
 /*                             Function Prototypes                            */
 /*----------------------------------------------------------------------------*/
 void timespec_diff(struct timespec *start, struct timespec *stop,
                    struct timespec *result);
+
 /*----------------------------------------------------------------------------*/
 /*                             External Functions                             */
 /*----------------------------------------------------------------------------*/
@@ -69,6 +75,7 @@ void createSocketConnection(void (* initKeypress)())
     bool seshat_registered = false;
     unsigned int webpa_ping_timeout_ms = 1000 * get_parodus_cfg()->webpa_ping_timeout;
     unsigned int heartBeatTimer = 0;
+    struct timespec start_svc_alive_timer;
     
     //loadParodusCfg(tmpCfg,get_parodus_cfg());
 #ifdef FEATURE_DNS_QUERY
@@ -94,12 +101,12 @@ void createSocketConnection(void (* initKeypress)())
     packMetaData();
     
     UpStreamMsgQ = NULL;
-    StartThread(handle_upstream);
-    StartThread(processUpstreamMessage);
+    StartThread(handle_upstream, &upstream_tid);
+    StartThread(processUpstreamMessage, &upstream_msg_tid);
     ParodusMsgQ = NULL;
-    StartThread(messageHandlerTask);
-    StartThread(serviceAliveTask);
-	StartThread(CRUDHandlerTask);
+    StartThread(messageHandlerTask, &downstream_tid);
+    StartThread(serviceAliveTask, &svc_alive_tid);
+    StartThread(CRUDHandlerTask, &crud_tid);
 
     if (NULL != initKeypress) 
     {
@@ -108,6 +115,8 @@ void createSocketConnection(void (* initKeypress)())
 
     seshat_registered = __registerWithSeshat();
     
+    clock_gettime(CLOCK_REALTIME, &start_svc_alive_timer);
+
     do
     {
         struct timespec start, stop, diff;
@@ -158,19 +167,41 @@ void createSocketConnection(void (* initKeypress)())
             ParodusInfo("cloud_status set as %s after connection close\n", get_parodus_cfg()->cloud_status);
             if(get_parodus_cfg()->cloud_disconnect !=NULL)
             {
-				ParodusPrint("get_parodus_cfg()->cloud_disconnect is %s\n", get_parodus_cfg()->cloud_disconnect);
-				set_cloud_disconnect_time(CLOUD_RECONNECT_TIME);
-				ParodusInfo("Waiting for %d minutes for reconnecting .. \n", get_cloud_disconnect_time());
+                ParodusPrint("get_parodus_cfg()->cloud_disconnect is %s\n", get_parodus_cfg()->cloud_disconnect);
+		set_cloud_disconnect_time(CLOUD_RECONNECT_TIME);
+		ParodusInfo("Waiting for %d minutes for reconnecting .. \n", get_cloud_disconnect_time());
 
-				sleep( get_cloud_disconnect_time() * 60 );
-				ParodusInfo("cloud-disconnect reason reset after %d minutes\n", get_cloud_disconnect_time());
-				free(get_parodus_cfg()->cloud_disconnect);
-				reset_cloud_disconnect_reason(get_parodus_cfg());
+		sleep (get_cloud_disconnect_time() * 60);
+		ParodusInfo("cloud-disconnect reason reset after %d minutes\n", get_cloud_disconnect_time());
+		free(get_parodus_cfg()->cloud_disconnect);
+		reset_cloud_disconnect_reason(get_parodus_cfg());
             }
             createNopollConnection(ctx);
         }
        } while(!get_close_retry() && !g_shutdown);
 
+    pthread_mutex_lock (get_global_svc_mut());
+    pthread_cond_signal (get_global_svc_con());
+    pthread_mutex_unlock (get_global_svc_mut());
+    pthread_mutex_lock (get_global_crud_mut());
+    pthread_cond_signal (get_global_crud_con());
+    pthread_mutex_unlock (get_global_crud_mut());
+    pthread_mutex_lock (&g_mutex);
+    pthread_cond_signal (&g_cond);
+    pthread_mutex_unlock (&g_mutex);
+    pthread_mutex_lock (get_global_nano_mut ());
+    pthread_cond_signal (get_global_nano_con());
+    pthread_mutex_unlock (get_global_nano_mut());
+
+    ParodusInfo ("joining threads\n");
+    JoinThread (svc_alive_tid);
+    JoinThread (upstream_tid);
+    JoinThread (downstream_tid);
+    JoinThread (upstream_msg_tid);
+    JoinThread (crud_tid);
+
+    deleteAllClients ();
+
     close_and_unref_connection(get_global_conn());
     nopoll_ctx_unref(ctx);
     nopoll_cleanup_library();

src/connection.c

@@ -46,6 +46,7 @@ static char *reconnect_reason = "webpa_process_starts";
 static int cloud_disconnect_max_time = 5;
 static noPollConn *g_conn = NULL;
 static bool LastReasonStatus = false;
+static int init = 1;
 static noPollConnOpts * createConnOpts (char * extra_headers, bool secure);
 static char* build_extra_headers( const char *auth, const char *device_id,
                                   const char *user_agent, const char *convey );
@@ -384,7 +385,7 @@ int nopoll_connect (create_connection_ctx_t *ctx, int is_ipv6)
            NULL, default_url,NULL,NULL);
       }      
    }
-   if (NULL == connection) {
+   if ((NULL == connection) && (!is_ipv6)) {
      if((checkHostIp(server->server_addr) == -2)) {
        if (check_timer_expired (&ctx->connect_timer, 15*60*1000)) {
   	 ParodusError("WebPA unable to connect due to DNS resolving to 10.0.0.1 for over 15 minutes; crashing service.\n");
@@ -406,6 +407,8 @@ int nopoll_connect (create_connection_ctx_t *ctx, int is_ipv6)
 #define WAIT_ACTION_RETRY	1	// if wait_status is 307, 302, 303 or 403
 #define WAIT_FAIL 	2
 
+#define FREE_NON_NULL_PTR(ptr) if (NULL != ptr) free(ptr)
+
 int wait_connection_ready (create_connection_ctx_t *ctx)
 {
   int wait_status;
@@ -413,7 +416,10 @@ int wait_connection_ready (create_connection_ctx_t *ctx)
 
   if(nopoll_conn_wait_for_status_until_connection_ready(get_global_conn(), 10, 
 	&wait_status, &redirectURL)) 
+  { 
+     FREE_NON_NULL_PTR (redirectURL);
      return WAIT_SUCCESS;
+  }
   if(wait_status == 307 || wait_status == 302 || wait_status == 303)    // only when there is a http redirect
   {
 	char *redirect_ptr = redirectURL;
@@ -431,9 +437,7 @@ int wait_connection_ready (create_connection_ctx_t *ctx)
 	set_current_server (ctx); // set current server to redirect server
 	return WAIT_ACTION_RETRY;
   }
-  if (NULL != redirectURL) {
-    free (redirectURL);
-  }
+  FREE_NON_NULL_PTR (redirectURL);
   if(wait_status == 403) 
   {
 	ParodusError("Received Unauthorized response with status: %d\n", wait_status);
@@ -556,7 +560,7 @@ int createNopollConnection(noPollCtx *ctx)
         set_server_list_null (&conn_ctx.server_list);
         init_backoff_timer (&backoff_timer, max_retry_count);
   
-	while (true)
+	while (!g_shutdown)
 	{
 	  query_dns_status = find_servers (&conn_ctx.server_list);
 	  if (query_dns_status == FIND_INVALID_DEFAULT)
@@ -579,9 +583,10 @@ 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) {
+	if((get_parodus_cfg()->boot_time != 0) && init) {
 		getCurrentTime(connectTimePtr);
 		ParodusInfo("connect_time-diff-boot_time=%d\n", connectTimePtr->tv_sec - get_parodus_cfg()->boot_time);
+		init = 0; //set init to 0 so that this is logged only during process start up and not during reconnect
 	}
 
 	free_extra_headers (&conn_ctx);

src/crud_interface.c

@@ -44,6 +44,16 @@ CrudMsg *crudMsgQ = NULL;
 /*                             External functions                             */
 /*----------------------------------------------------------------------------*/
 
+pthread_cond_t *get_global_crud_con(void)
+{
+    return &crud_con;
+}
+
+pthread_mutex_t *get_global_crud_mut(void)
+{
+    return &crud_mut;
+}
+
 void addCRUDmsgToQueue(wrp_msg_t *crudMsg)
 {
 	CrudMsg * crudMessage;
@@ -122,6 +132,10 @@ void *CRUDHandlerTask()
 		}
 		else
 		{
+			if (g_shutdown) {
+			  pthread_mutex_unlock (&crud_mut);
+			  break;
+			}
 			pthread_cond_wait(&crud_con, &crud_mut);
 			pthread_mutex_unlock (&crud_mut);
 		}

src/crud_interface.h

@@ -24,6 +24,8 @@
 #ifndef _CRUD_INTERFACE_H_
 #define _CRUD_INTERFACE_H_
 
+#include <pthread.h>
+
 #ifdef __cplusplus
 extern "C" {
 #endif
@@ -43,6 +45,8 @@ typedef struct CrudMsg__
 /*----------------------------------------------------------------------------*/
 
 void addCRUDresponseToUpstreamQ(void *response_bytes, ssize_t response_size);
+pthread_cond_t *get_global_crud_con(void);
+pthread_mutex_t *get_global_crud_mut(void);
 
 #ifdef __cplusplus
 }

src/downstream.c

@@ -27,7 +27,10 @@
 #include "partners_check.h"
 #include "ParodusInternal.h"
 #include "crud_interface.h"
-
+/*----------------------------------------------------------------------------*/
+/*                             Function Prototypes                            */
+/*----------------------------------------------------------------------------*/
+static void createNewMsgForCRUD(wrp_msg_t *message, wrp_msg_t **crudMessage );
 /*----------------------------------------------------------------------------*/
 /*                             External Functions                             */
 /*----------------------------------------------------------------------------*/
@@ -42,6 +45,7 @@ void listenerOnMessage(void * msg, size_t msgSize)
 {
     int rv =0;
     wrp_msg_t *message;
+    wrp_msg_t *crudMessage= NULL;
     char* destVal = NULL;
     char dest[32] = {'\0'};
     int msgType;
@@ -75,7 +79,6 @@ 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;
                 }
 
@@ -120,7 +123,8 @@ void listenerOnMessage(void * msg, size_t msgSize)
                             ((WRP_MSG_TYPE__EVENT == msgType) ? "NA" : message->u.crud.transaction_uuid)));
                         
                         free(destVal);
-						temp = get_global_node();
+
+                        temp = get_global_node();
                         //Checking for individual clients & Sending to each client
 
                         while (NULL != temp)
@@ -139,10 +143,11 @@ void listenerOnMessage(void * msg, size_t msgSize)
                             ParodusPrint("checking the next item in the list\n");
                             temp= temp->next;
                         }
+                        release_global_node ();
 
-						/* check Downstream dest for CRUD requests */
-						if(destFlag ==0 && strcmp("parodus", dest)==0)
-						{
+			/* check Downstream dest for CRUD requests */
+			if(destFlag ==0 && strcmp("parodus", dest)==0)
+			{
 							ParodusPrint("Received CRUD request : dest : %s\n", dest);
 							if ((message->u.crud.source == NULL) || (message->u.crud.transaction_uuid == NULL))
 							{
@@ -154,11 +159,12 @@ void listenerOnMessage(void * msg, size_t msgSize)
 							}
 							else
 							{
-								addCRUDmsgToQueue(message);
+							    createNewMsgForCRUD(message, &crudMessage);
+							    addCRUDmsgToQueue(crudMessage);
 							}
 							destFlag =1;
-						}
-						//if any unknown dest received sending error response to server
+			}
+			//if any unknown dest received sending error response to server
                         if(destFlag ==0)
                         {
                             ParodusError("Unknown dest:%s\n", dest);
@@ -184,14 +190,14 @@ void listenerOnMessage(void * msg, size_t msgSize)
                         else
                         {
                             resp_msg ->u.crud.source = message->u.crud.dest;
-							if(message->u.crud.source !=NULL)
-							{
-								resp_msg ->u.crud.dest = message->u.crud.source;
-							}
-							else
-							{
-								resp_msg ->u.crud.dest = "unknown";
-							}
+				if(message->u.crud.source !=NULL)
+				{
+					resp_msg ->u.crud.dest = message->u.crud.source;
+				}
+				else
+				{
+					resp_msg ->u.crud.dest = "unknown";
+				}
                             resp_msg ->u.crud.transaction_uuid = message->u.crud.transaction_uuid;
                             resp_msg ->u.crud.path =             message->u.crud.path;
                         }
@@ -225,8 +231,6 @@ void listenerOnMessage(void * msg, size_t msgSize)
                             resp_bytes = NULL;
                         }
                         free(resp_msg);
-                        ParodusPrint("free for downstream decoded msg\n");
-                        wrp_free_struct(message);
                     }
                     break;
                 }
@@ -235,9 +239,11 @@ void listenerOnMessage(void * msg, size_t msgSize)
                 case WRP_MSG_TYPE__SVC_ALIVE:
                 case WRP_MSG_TYPE__UNKNOWN:
                 default:
-                        wrp_free_struct(message);
                     break;
             }
+            ParodusPrint("free for downstream decoded msg\n");
+            wrp_free_struct(message);
+            message = NULL;
         }
         else
         {
@@ -245,3 +251,116 @@ void listenerOnMessage(void * msg, size_t msgSize)
         }
     }
 }
+/*----------------------------------------------------------------------------*/
+/*                             Internal Functions                             */
+/*----------------------------------------------------------------------------*/
+/**
+ * @brief createNewMsgForCRUD function to create new message for processing CRUD requests
+ *
+ * @param[in] message The message received from server
+ * @param[out] crudMessage New message for processing CRUD requests
+ */
+static void createNewMsgForCRUD(wrp_msg_t *message, wrp_msg_t **crudMessage )
+{
+    wrp_msg_t *msg;
+    msg = ( wrp_msg_t * ) malloc( sizeof( wrp_msg_t ) );
+    size_t i;
+    if(msg != NULL)
+    {
+        memset( msg, 0, sizeof( wrp_msg_t ) );
+        msg->msg_type = message->msg_type;
+        if(message->u.crud.source != NULL)
+        {
+            ParodusPrint("message->u.crud.source = %s\n",message->u.crud.source);
+            msg->u.crud.source = strdup(message->u.crud.source);
+        }
+
+        if(message->u.crud.dest!= NULL)
+        {
+            ParodusPrint("message->u.crud.dest = %s\n",message->u.crud.dest);
+            msg->u.crud.dest = strdup(message->u.crud.dest);
+        }
+
+        if(message->u.crud.transaction_uuid != NULL)
+        {
+            ParodusPrint("message->u.crud.transaction_uuid = %s\n",message->u.crud.transaction_uuid);
+            msg->u.crud.transaction_uuid = strdup(message->u.crud.transaction_uuid);
+        }
+
+        if(message->u.crud.partner_ids!= NULL && message->u.crud.partner_ids->count >0)
+        {
+            msg->u.crud.partner_ids = ( partners_t * ) malloc( sizeof( partners_t ) +
+                                                              sizeof( char * ) * message->u.crud.partner_ids->count );
+            if(msg->u.crud.partner_ids != NULL)
+            {
+                msg->u.crud.partner_ids->count = message->u.crud.partner_ids->count;
+                for(i = 0; i<message->u.crud.partner_ids->count; i++)
+                {
+                    ParodusPrint("message->u.crud.partner_ids->partner_ids[%d] = %s\n",i,message->u.crud.partner_ids->partner_ids[i]);
+                    msg->u.crud.partner_ids->partner_ids[i] = strdup(message->u.crud.partner_ids->partner_ids[i]);
+                }
+            }
+        }
+
+        if(message->u.crud.headers!= NULL && message->u.crud.headers->count >0)
+        {
+            msg->u.crud.headers = ( headers_t * ) malloc( sizeof( headers_t ) +
+                                                                sizeof( char * ) * message->u.crud.headers->count );
+            if(msg->u.crud.headers != NULL)
+            {
+                msg->u.crud.headers->count = message->u.crud.headers->count;
+                for(i = 0; i<message->u.crud.headers->count; i++)
+                {
+                    ParodusPrint("message->u.crud.headers->headers[%d] = %s\n",i,message->u.crud.headers->headers[i]);
+                    msg->u.crud.headers->headers[i] = strdup(message->u.crud.headers->headers[i]);
+                }
+            }
+        }
+
+        if(message->u.crud.metadata != NULL && message->u.crud.metadata->count > 0)
+        {
+            msg->u.crud.metadata = (data_t *) malloc( sizeof( data_t ) );
+            if(msg->u.crud.metadata != NULL)
+            {
+                memset( msg->u.crud.metadata, 0, sizeof( data_t ) );
+                msg->u.crud.metadata->count = message->u.crud.metadata->count;
+                msg->u.crud.metadata->data_items = ( struct data* )malloc( sizeof( struct data ) * ( message->u.crud.metadata->count ) );
+                for(i=0; i<message->u.crud.metadata->count; i++)
+                {
+                    if(message->u.crud.metadata->data_items[i].name != NULL)
+                    {
+                        ParodusPrint("message->u.crud.metadata->data_items[%d].name : %s\n",i,message->u.crud.metadata->data_items[i].name);
+                        msg->u.crud.metadata->data_items[i].name = strdup(message->u.crud.metadata->data_items[i].name);
+                    }
+                    if(message->u.crud.metadata->data_items[i].value != NULL)
+                    {
+                        ParodusPrint("message->u.crud.metadata->data_items[%d].value : %s\n",i,message->u.crud.metadata->data_items[i].value);
+                        msg->u.crud.metadata->data_items[i].value = strdup(message->u.crud.metadata->data_items[i].value);
+                    }
+                }
+            }
+        }
+        msg->u.crud.include_spans = message->u.crud.include_spans;
+        if(message->u.crud.content_type != NULL)
+        {
+            ParodusPrint("message->u.crud.content_type : %s\n",message->u.crud.content_type);
+            msg->u.crud.content_type = strdup(message->u.crud.content_type);
+        }
+        msg->u.crud.spans.spans = NULL;   /* not supported */
+        msg->u.crud.spans.count = 0;     /* not supported */
+        msg->u.crud.status = message->u.crud.status;
+        msg->u.crud.rdr = message->u.crud.rdr;
+        if(message->u.crud.payload != NULL)
+        {
+            ParodusPrint("message->u.crud.payload = %s\n", (char *)message->u.crud.payload);
+            msg->u.crud.payload = strdup((char *)message->u.crud.payload);
+        }
+        msg->u.crud.payload_size = message->u.crud.payload_size;
+        if(message->u.crud.path != NULL)
+        {
+            ParodusPrint("message->u.crud.path = %s\n", message->u.crud.path);
+            msg->u.crud.path = strdup(message->u.crud.path);
+        }
+        *crudMessage = msg;
+    }
+}

src/service_alive.c

@@ -31,9 +31,41 @@
 /*----------------------------------------------------------------------------*/
 #define KEEPALIVE_INTERVAL_SEC                         	30
 
+pthread_mutex_t svc_mut=PTHREAD_MUTEX_INITIALIZER;
+pthread_cond_t svc_con=PTHREAD_COND_INITIALIZER;
+
+/*----------------------------------------------------------------------------*/
+/*                             Utiliy Functions                               */
+/*----------------------------------------------------------------------------*/
+static int wait__ (unsigned int secs)
+{
+  int shutdown_flag;
+  struct timespec svc_alive_timer;
+
+  clock_gettime(CLOCK_REALTIME, &svc_alive_timer);
+  svc_alive_timer.tv_sec += secs;
+  pthread_mutex_lock(&svc_mut);
+  pthread_cond_timedwait (&svc_con, &svc_mut, &svc_alive_timer);
+  shutdown_flag = g_shutdown;
+  pthread_mutex_unlock (&svc_mut);
+  return shutdown_flag;
+}
+
 /*----------------------------------------------------------------------------*/
 /*                             External Functions                             */
 /*----------------------------------------------------------------------------*/
+
+pthread_cond_t *get_global_svc_con(void)
+{
+    return &svc_con;
+}
+
+pthread_mutex_t *get_global_svc_mut(void)
+{
+    return &svc_mut;
+}
+
+
 /*
  * @brief To handle registered services to indicate that the service is still alive.
  */
@@ -58,10 +90,10 @@ void *serviceAliveTask()
 	        while(1)
 	        {
 		        ParodusPrint("serviceAliveTask: numOfClients registered is %d\n", get_numOfClients());
+		        temp = get_global_node();
 		        if(get_numOfClients() > 0)
 		        {
 			        //sending svc msg to all the clients every 30s
-			        temp = get_global_node();
 			        size = (size_t) nbytes;
 			        while(NULL != temp)
 			        {
@@ -91,15 +123,20 @@ void *serviceAliveTask()
 					        temp= temp->next;
 				        }
 			        }
+				release_global_node ();
 		         	ParodusPrint("Waiting for 30s to send keep alive msg \n");
-		         	sleep(KEEPALIVE_INTERVAL_SEC);
+		         	if (wait__ (KEEPALIVE_INTERVAL_SEC))
+				  break;
 	            	}
 	            	else
 	            	{
+				release_global_node ();
 	            		ParodusInfo("No clients are registered, waiting ..\n");
-	            		sleep(50);
+	            		if (wait__ (50))
+				  break;
 	            	}
 	        }
+		free (svc_bytes);
 	}
 	return 0;
 }

src/service_alive.h

@@ -24,6 +24,8 @@
 #ifndef _SERVICE_ALIVE_H_
 #define _SERVICE_ALIVE_H_
 
+#include <pthread.h>
+
 /*----------------------------------------------------------------------------*/
 /*                             Function Prototypes                            */
 /*----------------------------------------------------------------------------*/
@@ -33,7 +35,8 @@ extern "C" {
 #endif
   
 void *serviceAliveTask();
-
+pthread_cond_t *get_global_svc_con(void);
+pthread_mutex_t *get_global_svc_mut(void);
 
 #ifdef __cplusplus
 }

src/spin_thread.c

@@ -23,7 +23,6 @@
  
 #include <stdlib.h>
 #include <string.h>
-#include <pthread.h>
 
 #include "spin_thread.h"
 #include "parodus_log.h"
@@ -31,12 +30,12 @@
 /*----------------------------------------------------------------------------*/
 /*                             External Functions                             */
 /*----------------------------------------------------------------------------*/
-void StartThread(void *(*start_routine) (void *))
+void StartThread(void *(*start_routine) (void *), pthread_t *threadId)
 {
     int err = 0;
-	pthread_t threadId;
+	pthread_t __threadId;
 
-	err = pthread_create(&threadId, NULL, start_routine, NULL);
+	err = pthread_create(&__threadId, NULL, start_routine, NULL);
 	if (err != 0) 
 	{
 		ParodusError("Error creating thread :[%s]\n", strerror(err));
@@ -44,8 +43,13 @@ void StartThread(void *(*start_routine) (void *))
 	}
 	else
 	{
-		ParodusPrint("Thread created Successfully %lu\n", (unsigned long) threadId);
+		*threadId = __threadId;
+		ParodusPrint("Thread created Successfully %lu\n", (unsigned long) __threadId);
 	}    
 }
 
-         
+
+void JoinThread (pthread_t threadId)
+{
+  pthread_join (threadId, NULL);
+}

src/spin_thread.h

@@ -24,6 +24,8 @@
 #ifndef _SPIN_THREAD_H_
 #define _SPIN_THREAD_H_
 
+#include <pthread.h>
+
 #ifdef __cplusplus
 extern "C" {
 #endif
@@ -32,8 +34,8 @@ extern "C" {
 /*                             Function Prototypes                            */
 /*----------------------------------------------------------------------------*/
 
-void StartThread(void *(*start_routine) (void *));
-
+void StartThread(void *(*start_routine) (void *), pthread_t *threadId);
+void JoinThread (pthread_t threadId);
 
 #ifdef __cplusplus
 }

src/thread_tasks.c

@@ -43,6 +43,10 @@ void *messageHandlerTask()
         }
         else
         {
+            if (g_shutdown) {
+                pthread_mutex_unlock (&g_mutex);
+                break;
+            }
             ParodusPrint("Before pthread cond wait in consumer thread\n");
             pthread_cond_wait(&g_cond, &g_mutex);
             pthread_mutex_unlock (&g_mutex);

src/token.c

@@ -527,7 +527,9 @@ int allow_insecure_conn(char **server_addr, unsigned int *port)
 	}
 
 	if (insecure >= 0) {
-		ParodusInfo ("JWT claims: %s\n", cJSON_Print (jwt->private_claims));
+		char *claim_str = cJSON_Print (jwt->private_claims);
+		ParodusInfo ("JWT claims: %s\n", claim_str);
+		free (claim_str);
 	}
 	cjwt_destroy(&jwt);
 	

src/upstream.c

@@ -34,7 +34,7 @@
 /*                                   Macros                                   */
 /*----------------------------------------------------------------------------*/
 #define METADATA_COUNT 					12
-#define CLOUD_STATUS_FORMAT				"parodus/cloud-status"
+#define PARODUS_SERVICE_NAME			"parodus"
 /*----------------------------------------------------------------------------*/
 /*                            File Scoped Variables                           */
 /*----------------------------------------------------------------------------*/
@@ -129,6 +129,12 @@ void *handle_upstream()
     sock = nn_socket( AF_SP, NN_PULL );
     if(sock >= 0)
     {
+        int t = NANO_SOCKET_RCV_TIMEOUT_MS;
+        int rc = nn_setsockopt(sock, NN_SOL_SOCKET, NN_RCVTIMEO, &t, sizeof(t));
+        if (rc < 0)
+        {
+            ParodusError ("Unable to set socket receive timeout (errno=%d, %s)\n",errno, strerror(errno));
+        }
         ParodusPrint("Nanomsg bind with get_parodus_cfg()->local_url  %s\n", get_parodus_cfg()->local_url);
         bind = nn_bind(sock, get_parodus_cfg()->local_url);
         if(bind < 0)
@@ -137,11 +143,18 @@ void *handle_upstream()
         }
         else
         {
+            ParodusInfo("nanomsg server gone into the listening mode...\n");
             while( FOREVER() ) 
             {
                 buf = NULL;
-                ParodusInfo("nanomsg server gone into the listening mode...\n");
                 bytes = nn_recv (sock, &buf, NN_MSG, 0);
+                if (g_shutdown)
+                  break;
+                if (bytes < 0) {
+                  if ((errno != EAGAIN) && (errno != ETIMEDOUT))
+                    ParodusInfo ("Error (%d) receiving message from nanomsg client\n", errno);
+                  continue;
+                }
                 ParodusInfo ("Upstream message received from nanomsg client\n");
                 message = (UpStreamMsg *)malloc(sizeof(UpStreamMsg));
 
@@ -197,11 +210,10 @@ void *processUpstreamMessage()
     reg_list_item_t *temp = NULL;
     int matchFlag = 0;
     int status = -1;
-    char *serviceName = NULL;
-    char *macId = NULL;
-    char *destService, *destApplication =NULL;
-    char *sourceService, *sourceApplication =NULL;
-    int sendStatus =-1;
+	char *device_id = NULL;
+	size_t device_id_len = 0;
+	size_t parodus_len;
+	int ret = -1;
 
     while(FOREVER())
     {
@@ -226,11 +238,11 @@ void *processUpstreamMessage()
                 {
                     ParodusInfo("\n Nanomsg client Registration for Upstream\n");
                     //Extract serviceName and url & store it in a linked list for reg_clients
+                    temp = get_global_node();
                     if(get_numOfClients() !=0)
                     {
                         matchFlag = 0;
                         ParodusPrint("matchFlag reset to %d\n", matchFlag);
-                        temp = get_global_node();
                         while(temp!=NULL)
                         {
                             if(strcmp(temp->service_name, msg->u.reg.service_name)==0)
@@ -245,11 +257,11 @@ void *processUpstreamMessage()
                                 temp->sock = nn_socket(AF_SP,NN_PUSH );
                                 if(temp->sock >= 0)
                                 {					
-                                    int t = NANOMSG_SOCKET_TIMEOUT_MSEC;
+                                    int t = NANO_SOCKET_SEND_TIMEOUT_MS;
                                     rc = nn_setsockopt(temp->sock, NN_SOL_SOCKET, NN_SNDTIMEO, &t, sizeof(t));
                                     if(rc < 0)
                                     {
-                                        ParodusError ("Unable to set socket timeout (errno=%d, %s)\n",errno, strerror(errno));
+                                        ParodusError ("Unable to set socket send timeout (errno=%d, %s)\n",errno, strerror(errno));
                                     }
                                     rc = nn_connect(temp->sock, msg->u.reg.url); 
                                     if(rc < 0)
@@ -289,11 +301,13 @@ void *processUpstreamMessage()
                             ParodusPrint("sent auth status to reg client\n");
                         }
                     }
+                    release_global_node ();
                 }
                 else if(msgType == WRP_MSG_TYPE__EVENT)
                 {
                     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)
@@ -322,6 +336,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
                 {
@@ -334,87 +360,46 @@ void *processUpstreamMessage()
 					else
 					{
 						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)
+						if(WRP_MSG_TYPE__RETREIVE == msgType)
 						{
-                                                        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);
-							sourceApplication = wrp_get_msg_element(WRP_ID_ELEMENT__APPLICATION, msg, SOURCE);
-							/*  Handle cloud-status retrieve request here
-								Expecting dest format as mac:xxxxxxxxxxxx/parodus/cloud-status
-								Parse dest field and check destService is "parodus" and destApplication is "cloud-status"
-							*/
-							if(macId != NULL)
-						    	{
-								if(destService != NULL && destApplication != NULL && strcmp(destService,"parodus")== 0 && strcmp(destApplication,"cloud-status")== 0)
+								ret = getDeviceId(&device_id, &device_id_len);
+								if(ret == 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)
+									ParodusPrint("device_id %s device_id_len %lu\n", device_id, device_id_len);
+									/* Match dest based on device_id. Check dest start with: "mac:112233445xxx" ? */
+									if( 0 == strncasecmp(device_id, msg->u.crud.dest, device_id_len-1) )
 									{
-										//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)
+										 /* For this device. */
+										parodus_len = strlen( PARODUS_SERVICE_NAME );
+										if( 0 == strncmp(PARODUS_SERVICE_NAME, &msg->u.crud.dest[device_id_len], parodus_len-1) ) 
 										{
-											ParodusInfo("Send upstreamMsg successfully to registered client %s\n", serviceName);
+											/* For Parodus CRUD queue. */
+											ParodusInfo("Create RetrieveMsg and add to parodus CRUD queue\n");
+											createUpstreamRetrieveMsg(msg, &retrieve_msg);
+											addCRUDmsgToQueue(retrieve_msg);
 										}
 										else
 										{
-											ParodusError("Failed to send upstreamMsg to registered client %s\n", serviceName);
+											/* For nanomsg clients. */
+											getServiceNameAndSendResponse(msg, &message->msg, message->len);
 										}
-										free(serviceName);
-										serviceName = NULL;
 									}
 									else
 									{
-										ParodusError("serviceName is NULL,not sending cloud-status response to client\n");
+										/* Not for this device. Send upstream */
+										ParodusInfo("sendUpstreamMsgToServer \n");
+										sendUpstreamMsgToServer(&message->msg, message->len);
+									}
+									if(device_id != NULL)
+									{
+										free(device_id);
+										device_id = NULL;
 									}
 								}
 								else
 								{
-									ParodusInfo("sendUpstreamMsgToServer \n");
-									sendUpstreamMsgToServer(&message->msg, message->len);
+									ParodusError("Failed to get device_id\n");
 								}
-							}
-							else
-							{
-								ParodusError("MAC is null, not handling retrieve wrp message \n");
-							}
-							if(sourceService !=NULL)
-							{
-								free(sourceService);
-								sourceService = NULL;
-							}
-							if(sourceApplication !=NULL)
-							{
-								free(sourceApplication);
-								sourceApplication = NULL;
-							}
-							if(destService !=NULL)
-							{
-								free(destService);
-								destService = NULL;
-							}
-							if(destApplication !=NULL)
-							{
-								free(destApplication);
-								destApplication = NULL;
-							}
 						}
 						else
 						{
@@ -450,6 +435,10 @@ void *processUpstreamMessage()
         }
         else
         {
+            if (g_shutdown) {
+                pthread_mutex_unlock (&nano_mut);
+                break;
+            }
             ParodusPrint("Before pthread cond wait in consumer thread\n");   
             pthread_cond_wait(&nano_con, &nano_mut);
             pthread_mutex_unlock (&nano_mut);
@@ -459,6 +448,107 @@ void *processUpstreamMessage()
     return NULL;
 }
 
+/**
+ * @brief getDeviceId function to create deviceId in the format "mac:112233445xxx"
+ *
+ * @param[out] device_id in the format "mac:112233445xxx"
+ * @param[out] total size of device_id
+ */
+int getDeviceId(char **device_id, size_t *device_id_len)
+{
+	char *deviceID = NULL;
+	size_t len;
+
+	if((get_parodus_cfg()->hw_mac !=NULL) && (strlen(get_parodus_cfg()->hw_mac)!=0))
+	{
+		len = strlen(get_parodus_cfg()->hw_mac) + 5;
+
+		deviceID = (char *) malloc(sizeof(char)*64);
+		if(deviceID != NULL)
+		{
+			snprintf(deviceID, len, "mac:%s", get_parodus_cfg()->hw_mac);
+			*device_id = deviceID;
+			*device_id_len = len;
+		}
+		else
+		{
+			ParodusError("device_id allocation failed\n");
+			return -1;
+		}
+	}
+	else
+	{
+		ParodusError("device mac is NULL\n");
+		return -1;
+	}
+	return 0;
+}
+
+
+/**
+ * @brief createUpstreamRetrieveMsg function to create new message for processing Retrieve requests
+ *
+ * @param[in] message The upstream message received from cloud or internal clients
+ * @param[out] retrieve_msg New message for processing Retrieve requests
+ */
+void createUpstreamRetrieveMsg(wrp_msg_t *message, wrp_msg_t **retrieve_msg)
+{
+	wrp_msg_t *msg;
+	msg = ( wrp_msg_t * ) malloc( sizeof( wrp_msg_t ) );
+	if(msg != NULL)
+    {
+		memset(msg, 0, sizeof(wrp_msg_t));
+
+		msg->msg_type = message->msg_type;
+		if(message->u.crud.transaction_uuid != NULL)
+		{
+			msg->u.crud.transaction_uuid = strdup(message->u.crud.transaction_uuid);
+		}
+		if(message->u.crud.source !=NULL)
+		{
+			msg->u.crud.source = strdup(message->u.crud.source);
+		}
+		if(message->u.crud.dest != NULL)
+		{
+			msg->u.crud.dest = strdup(message->u.crud.dest);
+		}
+		*retrieve_msg = msg;
+	}
+}
+
+/**
+ * @brief getServiceNameAndSendResponse function to fetch client service name and to send msg to it.
+ *
+ * @param[in] msg The decoded message to fetch client service name from its dest field
+ * @param[in] msg_bytes The encoded upstream msg to be sent to client
+ * @param[in] msg_size Total size of the msg to send to client
+ */
+void getServiceNameAndSendResponse(wrp_msg_t *msg, void **msg_bytes, size_t msg_size)
+{
+	char *serviceName = NULL;
+	int sendStatus =-1;
+
+	serviceName = wrp_get_msg_element(WRP_ID_ELEMENT__SERVICE, msg, DEST);
+	if ( serviceName != NULL)
+	{
+		sendStatus=sendMsgtoRegisteredClients(serviceName,(const char **)msg_bytes, msg_size);
+		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("serviceName is NULL,not sending retrieve response to client\n");
+	}
+}
+
 void sendUpstreamMsgToServer(void **resp_bytes, size_t resp_size)
 {
 	void *appendData;

src/upstream.h

@@ -47,8 +47,10 @@ typedef struct UpStreamMsg__
 void packMetaData();
 void *handle_upstream();
 void *processUpstreamMessage();
-
+int getDeviceId(char **device_id, size_t *device_id_len);
 void sendUpstreamMsgToServer(void **resp_bytes, size_t resp_size);
+void getServiceNameAndSendResponse(wrp_msg_t *msg, void **msg_bytes, size_t msg_size);
+void createUpstreamRetrieveMsg(wrp_msg_t *message, wrp_msg_t **retrieve_msg);
 void set_global_UpStreamMsgQ(UpStreamMsg * UpStreamQ);
 UpStreamMsg * get_global_UpStreamMsgQ(void);
 pthread_cond_t *get_global_nano_con(void);

tests/test_client_list.c

@@ -26,6 +26,9 @@
 #define TEST_CLIENT1_URL "tcp://127.0.0.1:6677"
 #define TEST_CLIENT2_URL "tcp://127.0.0.1:6655"
 
+pthread_t test_tid;
+pthread_t test_tid2;
+
 static void *client_rcv_task();
 static void *client2_rcv_task();
 
@@ -58,7 +61,7 @@ void test_client_addtolist()
 	ParodusPrint("decoded service_name:%s\n", message->u.reg.service_name);
 	ParodusPrint("decoded dest:%s\n", message->u.reg.url);
 	
-	StartThread(client_rcv_task);
+	StartThread(client_rcv_task, &test_tid);
 	
 	status = addToList(&message);
 	ParodusPrint("addToList status is %d\n", status);
@@ -73,7 +76,7 @@ void test_client_addtolist()
 		CU_ASSERT_STRING_EQUAL( temp->service_name, message->u.reg.service_name );
 		CU_ASSERT_STRING_EQUAL( temp->url, message->u.reg.url );
 	}
-
+	release_global_node ();
 	wrp_free_struct(message);
 	free(bytes);
 	ParodusInfo("test_client_addtolist done..\n");	
@@ -189,7 +192,7 @@ void test_addtolist_multiple_clients()
 	ParodusPrint("decoded service_name:%s\n", message->u.reg.service_name);
 	ParodusPrint("decoded dest:%s\n", message->u.reg.url);
 	
-	StartThread(client2_rcv_task);
+	StartThread(client2_rcv_task, &test_tid2);
 	
 	status = addToList(&message);
 	ParodusPrint("addToList status is %d\n", status);
@@ -206,7 +209,7 @@ void test_addtolist_multiple_clients()
 		CU_ASSERT_STRING_EQUAL( temp->url, message->u.reg.url );
 		
 	}
-
+	release_global_node ();
 	wrp_free_struct(message);
 	free(bytes);
 	ParodusInfo("test_addtolist_multiple_clients done..\n");	

tests/test_conn_interface.c

@@ -35,8 +35,12 @@
 /*----------------------------------------------------------------------------*/
 UpStreamMsg *UpStreamMsgQ;
 ParodusMsg *ParodusMsgQ;
-pthread_mutex_t nano_mut;
-pthread_cond_t nano_con;
+pthread_mutex_t g_mutex=PTHREAD_MUTEX_INITIALIZER;
+pthread_cond_t g_cond=PTHREAD_COND_INITIALIZER;
+pthread_mutex_t nano_mut=PTHREAD_MUTEX_INITIALIZER;
+pthread_cond_t nano_con=PTHREAD_COND_INITIALIZER;
+pthread_mutex_t svc_mut=PTHREAD_MUTEX_INITIALIZER;
+pthread_cond_t svc_con=PTHREAD_COND_INITIALIZER;
 
  
 /*----------------------------------------------------------------------------*/
@@ -103,9 +107,9 @@ void *messageHandlerTask()
     return NULL;
 }
 
-void *serviceAliveTask()
+int serviceAliveTask()
 {
-    return NULL;
+    return 0;
 }
 
 int nopoll_loop_wait(noPollCtx * ctx,long timeout)
@@ -156,12 +160,18 @@ void set_global_conn(noPollConn *conn)
     function_called();
 }
 
-void StartThread(void *(*start_routine) (void *))
+void StartThread(void *(*start_routine) (void *), pthread_t *threadId)
 {
     UNUSED(start_routine);
+    UNUSED(threadId);
     function_called(); 
 }
 
+void JoinThread (pthread_t threadId)
+{
+    UNUSED(threadId);
+}
+
 noPollCtx* nopoll_ctx_new(void)
 {
     function_called();
@@ -192,6 +202,16 @@ pthread_mutex_t *get_global_nano_mut(void)
     return &nano_mut;
 }
 
+pthread_cond_t *get_global_svc_con(void)
+{
+    return &svc_con;
+}
+
+pthread_mutex_t *get_global_svc_mut(void)
+{
+    return &svc_mut;
+}
+
 /*
 * Mock func to calculate time diff between start and stop time
 * This timespec_diff retuns 1 sec as diff time
@@ -205,6 +225,10 @@ void timespec_diff(struct timespec *start, struct timespec *stop,
    diff->tv_nsec = 1000;
 }
 
+void deleteAllClients (void)
+{
+}
+
 /*----------------------------------------------------------------------------*/
 /*                                   Tests                                    */
 /*----------------------------------------------------------------------------*/

tests/test_connection.c

@@ -63,6 +63,7 @@ bool LastReasonStatus;
 pthread_mutex_t close_mut;
 
 // Mock values
+bool g_shutdown = false;
 char *mock_server_addr;
 unsigned int mock_port;
 int mock_wait_status;
@@ -515,34 +516,58 @@ void test_nopoll_connect ()
   test_server.allow_insecure = 1;
   will_return (nopoll_conn_new_opts, NULL);
   expect_function_call (nopoll_conn_new_opts);
-  will_return (checkHostIp, 0);
-  expect_function_call (checkHostIp);
+  //will_return (checkHostIp, 0);
+  //expect_function_call (checkHostIp);
   assert_int_equal (nopoll_connect (&ctx, true), 0);
   assert_ptr_equal(NULL, get_global_conn());
   
   test_server.allow_insecure = 0;
   will_return (nopoll_conn_tls_new6, NULL);
   expect_function_call (nopoll_conn_tls_new6);
+  //will_return (checkHostIp, 0);
+  //expect_function_call (checkHostIp);
+  assert_int_equal (nopoll_connect (&ctx, true), 0);
+  assert_ptr_equal(NULL, get_global_conn());
+
+  will_return (nopoll_conn_tls_new, NULL);
+  expect_function_call (nopoll_conn_tls_new);
   will_return (checkHostIp, 0);
   expect_function_call (checkHostIp);
-  assert_int_equal (nopoll_connect (&ctx, true), 0);
+  assert_int_equal (nopoll_connect (&ctx, false), 0);
   assert_ptr_equal(NULL, get_global_conn());
 
   will_return (nopoll_conn_tls_new6, NULL);
   expect_function_call (nopoll_conn_tls_new6);
+  //will_return (checkHostIp, -2);
+  //expect_function_call (checkHostIp);
+  assert_int_equal (nopoll_connect (&ctx, true), 0);
+  assert_ptr_equal(NULL, get_global_conn());
+
+  will_return (nopoll_conn_tls_new, NULL);
+  expect_function_call (nopoll_conn_tls_new);
   will_return (checkHostIp, -2);
   expect_function_call (checkHostIp);
-  assert_int_equal (nopoll_connect (&ctx, true), 0);
+  assert_int_equal (nopoll_connect (&ctx, false), 0);
   assert_ptr_equal(NULL, get_global_conn());
 
   will_return (nopoll_conn_tls_new6, NULL);
   expect_function_call (nopoll_conn_tls_new6);
+  //will_return (checkHostIp, -2);
+  //expect_function_call (checkHostIp);
+  //ctx.connect_timer.start_time.tv_sec -= (15*60);
+  //will_return(kill, 1);
+  //expect_function_call(kill);
+  assert_int_equal (nopoll_connect (&ctx, true), 0);
+  assert_ptr_equal(NULL, get_global_conn());
+
+  will_return (nopoll_conn_tls_new, NULL);
+  expect_function_call (nopoll_conn_tls_new);
   will_return (checkHostIp, -2);
   expect_function_call (checkHostIp);
   ctx.connect_timer.start_time.tv_sec -= (15*60);
   will_return(kill, 1);
   expect_function_call(kill);
-  assert_int_equal (nopoll_connect (&ctx, true), 0);
+  assert_int_equal (nopoll_connect (&ctx, false), 0);
   assert_ptr_equal(NULL, get_global_conn());
 
   init_expire_timer (&ctx.connect_timer);
@@ -714,8 +739,8 @@ void test_connect_and_wait ()
   test_server.allow_insecure = 0;
   will_return (nopoll_conn_tls_new6, NULL);
   expect_function_call (nopoll_conn_tls_new6);
-  will_return (checkHostIp, 0);
-  expect_function_call (checkHostIp);
+  //will_return (checkHostIp, 0);
+  //expect_function_call (checkHostIp);
   assert_int_equal (connect_and_wait (&ctx), CONN_WAIT_RETRY_DNS);
   
   Cfg.flags = 0;
@@ -723,8 +748,8 @@ void test_connect_and_wait ()
 
   will_return (nopoll_conn_tls_new6, NULL);
   expect_function_call (nopoll_conn_tls_new6);
-  will_return (checkHostIp, 0);
-  expect_function_call (checkHostIp);
+  //will_return (checkHostIp, 0);
+  //expect_function_call (checkHostIp);
   will_return (nopoll_conn_tls_new, NULL);
   expect_function_call (nopoll_conn_tls_new);
   will_return (checkHostIp, 0);

tests/test_crud_interface.c

@@ -35,9 +35,11 @@
 /*----------------------------------------------------------------------------*/
 /*                            File Scoped Variables                           */
 /*----------------------------------------------------------------------------*/
+bool g_shutdown  = false;
 extern CrudMsg *crudMsgQ;
 int numLoops = 1;
 wrp_msg_t *temp = NULL;
+
 /*----------------------------------------------------------------------------*/
 /*                                   Mocks                                    */
 /*----------------------------------------------------------------------------*/

tests/test_crud_internal.c

@@ -165,7 +165,7 @@ void test_retrieveFromMemory()
 	assert_int_equal (ret, 0);
 	ret = retrieveFromMemory("webpa-protocol", &jsonresponse );
 	assert_int_equal (ret, 0);
-	ret = retrieveFromMemory("webpa-inteface-used", &jsonresponse );
+	ret = retrieveFromMemory("webpa-interface-used", &jsonresponse );
 	assert_int_equal (ret, 0);
 	ret = retrieveFromMemory("webpa-backoff-max", &jsonresponse );
 	assert_int_equal (ret, 0);
@@ -200,7 +200,7 @@ void test_retrieveFromMemoryFailure()
 	assert_int_equal (ret, -1);
 	ret = retrieveFromMemory("webpa-protocol", &jsonresponse );
 	assert_int_equal (ret, -1);
-	ret = retrieveFromMemory("webpa-inteface-used", &jsonresponse );
+	ret = retrieveFromMemory("webpa-interface-used", &jsonresponse );
 	assert_int_equal (ret, -1);
 	ret = retrieveFromMemory("webpa-backoff-max", &jsonresponse );
 	assert_int_equal (ret, 0);

tests/test_downstream.c

@@ -61,6 +61,10 @@ reg_list_item_t * get_global_node(void)
     return mock_ptr_type(reg_list_item_t *);
 }
 
+void release_global_node (void)
+{
+}
+
 ssize_t wrp_to_struct( const void *bytes, const size_t length,
                        const enum wrp_format fmt, wrp_msg_t **msg )
 {
@@ -68,7 +72,7 @@ ssize_t wrp_to_struct( const void *bytes, const size_t length,
     function_called();
     *msg = (wrp_msg_t*) malloc(sizeof(wrp_msg_t));
     memset(*msg, 0, sizeof(wrp_msg_t));
-	(*msg)->msg_type = WRP_MSG_TYPE__REQ;
+	(*msg)->msg_type = WRP_MSG_TYPE__REQ;   
 	(*msg)->u.req.dest = (char *) malloc(sizeof(char) *100);
 	(*msg)->u.req.partner_ids = (partners_t *) malloc(sizeof(partners_t));
 	(*msg)->u.req.partner_ids->count = 1;
@@ -79,8 +83,10 @@ ssize_t wrp_to_struct( const void *bytes, const size_t length,
 	{
 			(*msg)->msg_type = WRP_MSG_TYPE__CREATE;
 			parStrncpy((*msg)->u.crud.dest,"mac:1122334455/parodus", 100);
-			(*msg)->u.crud.source = "tag-update";
-			(*msg)->u.crud.transaction_uuid = "1234";
+                        (*msg)->u.crud.source = (char *) malloc(sizeof(char) *40);
+			parStrncpy ((*msg)->u.crud.source, "tag-update", 40);
+                        (*msg)->u.crud.transaction_uuid = (char *) malloc(sizeof(char) *40);
+			parStrncpy ((*msg)->u.crud.transaction_uuid, "1234", 40);
 	}
     return (ssize_t) mock();
 }

tests/test_downstream_more.c

@@ -244,6 +244,10 @@ reg_list_item_t *get_global_node(void)
     return NULL;
 }
 
+void release_global_node (void)
+{
+}
+
 void wrp_free_struct( wrp_msg_t *msg )
 {
     if( WRP_MSG_TYPE__EVENT == tests[i].s.msg_type ) {

tests/test_service_alive.c

@@ -33,6 +33,18 @@ static void *keep_alive_thread();
 static void add_client();
 int sock1;
 pthread_t threadId;
+pthread_mutex_t crud_mut=PTHREAD_MUTEX_INITIALIZER;
+pthread_cond_t crud_con=PTHREAD_COND_INITIALIZER;
+
+pthread_cond_t *get_global_crud_con(void)
+{
+    return &crud_con;
+}
+
+pthread_mutex_t *get_global_crud_mut(void)
+{
+    return &crud_mut;
+}
 
 /*----------------------------------------------------------------------------*/
 /*                                   Tests                                    */
@@ -48,12 +60,14 @@ void *CRUDHandlerTask()
 {
 	return NULL;
 }
+
 static void add_client()
 {
 	const wrp_msg_t reg = { .msg_type = WRP_MSG_TYPE__SVC_REGISTRATION,
 	  .u.reg.service_name = "service_client",
 	  .u.reg.url = TEST_SERVICE_URL};
 	  
+	pthread_t test_tid;
 	void *bytes;
 	int size =0;
 	int rv;
@@ -71,7 +85,7 @@ static void add_client()
 	ParodusPrint("decoded service_name:%s\n", message->u.reg.service_name);
 	ParodusPrint("decoded dest:%s\n", message->u.reg.url);
 	
-	StartThread(client_rcv_task);
+	StartThread(client_rcv_task, &test_tid);
 	status = addToList(&message);
 	ParodusPrint("addToList status is %d\n", status);
 	
@@ -134,7 +148,10 @@ static void *keep_alive_thread()
 	//ParodusPrint("keep_alive threadId is %d\n", threadId);
 	sleep(2);
 	ParodusPrint("Starting serviceAliveTask..\n");
-	serviceAliveTask();
+	while (true) {
+	  serviceAliveTask();
+	  sleep (30);
+	}
 	return 0;
 }
 

tests/test_spin_thread_e.c

@@ -23,6 +23,8 @@
 #include "../src/ParodusInternal.h"
 #include "../src/spin_thread.h"
 
+pthread_t test_tid;
+
 /*----------------------------------------------------------------------------*/
 /*                                   Mocks                                    */
 /*----------------------------------------------------------------------------*/
@@ -54,7 +56,7 @@ void *_routine(void *v)
 
 void test_StartThread_error()
 {
-    StartThread(&_routine);
+    StartThread(&_routine, &test_tid);
 }
 
 void add_suites( CU_pSuite *suite )

tests/test_spin_thread_s.c

@@ -23,6 +23,8 @@
 #include "../src/ParodusInternal.h"
 #include "../src/spin_thread.h"
 
+pthread_t test_tid;
+
 /*----------------------------------------------------------------------------*/
 /*                                   Mocks                                    */
 /*----------------------------------------------------------------------------*/
@@ -44,7 +46,7 @@ void *_routine(void *v)
 
 void test_StartThread_success()
 {
-    StartThread(&_routine);
+    StartThread(&_routine, &test_tid);
 }
 
 void add_suites( CU_pSuite *suite )

tests/test_thread_tasks.c

@@ -30,6 +30,7 @@
 /*----------------------------------------------------------------------------*/
 /*                            File Scoped Variables                           */
 /*----------------------------------------------------------------------------*/
+bool g_shutdown  = false;
 ParodusMsg *ParodusMsgQ;
 pthread_mutex_t g_mutex;
 pthread_cond_t g_cond;

tests/test_token.c

@@ -158,6 +158,19 @@ extern void read_key_from_file (const char *fname, char *buf, size_t buflen);
 extern const char *get_tok (const char *src, int delim, char *result, int resultsize);
 extern unsigned int get_algo_mask (const char *algo_str);
 
+pthread_mutex_t crud_mut=PTHREAD_MUTEX_INITIALIZER;
+pthread_cond_t crud_con=PTHREAD_COND_INITIALIZER;
+
+pthread_cond_t *get_global_crud_con(void)
+{
+    return &crud_con;
+}
+
+pthread_mutex_t *get_global_crud_mut(void)
+{
+    return &crud_mut;
+}
+
 void addCRUDmsgToQueue(wrp_msg_t *crudMsg)
 {
 	(void)crudMsg;

tests/test_token_stub.c

@@ -23,7 +23,7 @@
 #include <cmocka.h>
 #include <assert.h>
 #include <wrp-c.h>
-
+#include <pthread.h>
 #include "../src/token.h"
 
 
@@ -31,6 +31,20 @@
 /*                                   Mocks                                    */
 /*----------------------------------------------------------------------------*/
 
+
+pthread_mutex_t crud_mut=PTHREAD_MUTEX_INITIALIZER;
+pthread_cond_t crud_con=PTHREAD_COND_INITIALIZER;
+
+pthread_cond_t *get_global_crud_con(void)
+{
+    return &crud_con;
+}
+
+pthread_mutex_t *get_global_crud_mut(void)
+{
+    return &crud_mut;
+}
+
 void addCRUDmsgToQueue(wrp_msg_t *crudMsg)
 {
 	(void)crudMsg;

tests/test_upstream.c

@@ -38,10 +38,12 @@
 /*----------------------------------------------------------------------------*/
 static noPollConn *conn;
 static char *reconnect_reason = "webpa_process_starts";
+bool g_shutdown  = false;
 static ParodusCfg parodusCfg;
 extern size_t metaPackSize;
 extern UpStreamMsg *UpStreamMsgQ;
 int numLoops = 1;
+int deviceIDNull =0;
 wrp_msg_t *temp = NULL;
 extern pthread_mutex_t nano_mut;
 extern pthread_cond_t nano_con;
@@ -66,6 +68,10 @@ reg_list_item_t * get_global_node(void)
     return mock_ptr_type(reg_list_item_t *);
 }
 
+void release_global_node (void)
+{
+}
+
 int get_numOfClients()
 {
     function_called();
@@ -92,8 +98,21 @@ void sendMessage(noPollConn *conn, void *msg, size_t len)
     function_called();
 }
 
+void set_parodus_cfg(ParodusCfg *cfg)
+{
+    memcpy(&parodusCfg, cfg, sizeof(ParodusCfg));
+}
+
 ParodusCfg *get_parodus_cfg(void) 
 {
+	ParodusCfg cfg;
+	memset(&cfg,0,sizeof(cfg));
+	parStrncpy(cfg.hw_mac , "14cfe2142xxx", sizeof(cfg.hw_mac));
+	if(deviceIDNull)
+	{
+		parStrncpy(cfg.hw_mac , "", sizeof(cfg.hw_mac));
+	}
+	set_parodus_cfg(&cfg);
     return &parodusCfg;
 }
 
@@ -241,6 +260,8 @@ void test_handleUpstreamNull()
     UpStreamMsgQ = NULL;
     will_return(nn_socket, 1);
     expect_function_call(nn_socket);
+    will_return(nn_setsockopt, 0);
+    expect_function_call(nn_setsockopt);
     will_return(nn_bind, 1);
     expect_function_call(nn_bind);
     will_return(nn_recv, 12);
@@ -260,6 +281,8 @@ void test_handle_upstream()
     UpStreamMsgQ->next->next = NULL;
     will_return(nn_socket, 1);
     expect_function_call(nn_socket);
+    will_return(nn_setsockopt, 0);
+    expect_function_call(nn_setsockopt);
     will_return(nn_bind, 1);
     expect_function_call(nn_bind);
     will_return(nn_recv, 12);
@@ -273,6 +296,8 @@ void err_handleUpstreamBindFailure()
 {
     will_return(nn_socket, 1);
     expect_function_call(nn_socket);
+    will_return(nn_setsockopt, 0);
+    expect_function_call(nn_setsockopt);
     will_return(nn_bind, -1);
     expect_function_call(nn_bind);
     handle_upstream();
@@ -414,12 +439,12 @@ void test_processUpstreamMessageRegMsg()
     will_return(wrp_to_struct, 12);
     expect_function_call(wrp_to_struct);
 
-    will_return(get_numOfClients, 1);
-    expect_function_call(get_numOfClients);
-
     will_return(get_global_node, (intptr_t)head);
     expect_function_call(get_global_node);
 
+    will_return(get_numOfClients, 1);
+    expect_function_call(get_numOfClients);
+
     will_return(nn_shutdown, 1);
     expect_function_call(nn_shutdown);
 
@@ -473,6 +498,9 @@ void test_processUpstreamMessageRegMsgNoClients()
     will_return(wrp_to_struct, 12);
     expect_function_call(wrp_to_struct);
 
+    will_return(get_global_node, (intptr_t)head);
+    expect_function_call(get_global_node);
+
     will_return(get_numOfClients, 0);
     expect_function_call(get_numOfClients);
 
@@ -570,12 +598,12 @@ void err_processUpstreamMessageRegMsg()
     will_return(wrp_to_struct, 12);
     expect_function_call(wrp_to_struct);
 
-    will_return(get_numOfClients, 1);
-    expect_function_call(get_numOfClients);
-
     will_return(get_global_node, (intptr_t)head);
     expect_function_call(get_global_node);
 
+    will_return(get_numOfClients, 1);
+    expect_function_call(get_numOfClients);
+
     will_return(nn_shutdown, -1);
     expect_function_call(nn_shutdown);
 
@@ -726,16 +754,16 @@ void test_processUpstreamMsg_cloud_status()
 
 void test_processUpstreamMsg_sendToClient()
 {
-    numLoops = 2;
-    metaPackSize = 20;
+	numLoops = 2;
+	metaPackSize = 20;
 	UpStreamMsgQ = (UpStreamMsg *) malloc(sizeof(UpStreamMsg));
 	UpStreamMsgQ->msg = strdup("First Message");
 	UpStreamMsgQ->len = 13;
 	UpStreamMsgQ->next= NULL;
 	UpStreamMsgQ->next = (UpStreamMsg *) malloc(sizeof(UpStreamMsg));
-    UpStreamMsgQ->next->msg = strdup("Second Message");
-    UpStreamMsgQ->next->len = 15;
-    UpStreamMsgQ->next->next = NULL;
+	UpStreamMsgQ->next->msg = strdup("Second Message");
+	UpStreamMsgQ->next->len = 15;
+	UpStreamMsgQ->next->next = NULL;
 
 	temp = (wrp_msg_t *) malloc(sizeof(wrp_msg_t));
 	memset(temp,0,sizeof(wrp_msg_t));
@@ -758,86 +786,66 @@ void test_processUpstreamMsg_sendToClient()
 
 	expect_function_call(wrp_free_struct);
 	processUpstreamMessage();
-    free(temp);
-    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);
+	free(temp);
+	free(UpStreamMsgQ);
 	UpStreamMsgQ = NULL;
 }
-void err_processUpstreamMessageWithoutMac()
+
+void test_processUpstreamMsg_serviceNameNULL()
 {
-    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;
+	numLoops = 1;
+	metaPackSize = 20;
+	UpStreamMsgQ = (UpStreamMsg *) malloc(sizeof(UpStreamMsg));
+	UpStreamMsgQ->msg = strdup("First Message");
+	UpStreamMsgQ->len = 13;
+	UpStreamMsgQ->next= NULL;
+	UpStreamMsgQ->next = (UpStreamMsg *) malloc(sizeof(UpStreamMsg));
+	UpStreamMsgQ->next->msg = strdup("Second Message");
+	UpStreamMsgQ->next->len = 15;
+	UpStreamMsgQ->next->next = NULL;
+
+	temp = (wrp_msg_t *) malloc(sizeof(wrp_msg_t));
+	memset(temp,0,sizeof(wrp_msg_t));
+	temp->msg_type = 6;
+	temp->u.crud.dest = strdup("mac:14cfe2142xxx/");
+	temp->u.crud.source = strdup("mac:14cfe2142xxx/parodus/cloud-status");
+	temp->u.crud.transaction_uuid = strdup("123");
+
+	will_return(wrp_to_struct, 12);
+	expect_function_call(wrp_to_struct);
+	expect_function_call(wrp_free_struct);
+	processUpstreamMessage();
+	free(temp);
+	free(UpStreamMsgQ);
+	UpStreamMsgQ = NULL;
 }
 
+void err_processUpstreamMsg_deviceID()
+{
+	numLoops = 1;
+	metaPackSize = 20;
+	deviceIDNull = 1;
+	UpStreamMsgQ = (UpStreamMsg *) malloc(sizeof(UpStreamMsg));
+	UpStreamMsgQ->msg = "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 = 6;
+	temp->u.crud.dest = "mac:14cfe2142xxx/parodus/cloud-status";
+	temp->u.crud.source = "mac:14cfe2142xxx/config";
+	temp->u.crud.transaction_uuid = "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                             */
 /*----------------------------------------------------------------------------*/
@@ -870,9 +878,8 @@ 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),
+	cmocka_unit_test(test_processUpstreamMsg_serviceNameNULL),
+	cmocka_unit_test(err_processUpstreamMsg_deviceID)
     };
 
     return cmocka_run_group_tests(tests, NULL, NULL);