Merge pull request #281 from sadhyama/parodus-interface

To correct the spelling 'webpa-inteface-used' in convey header
2026-01-27 18:20:04 +00:00 · 2019-02-07 13:05:10 -08:00 · 2019-02-07 16:42:42 +05:30 · 2019-02-05 16:20:35 -08:00 · 2019-01-28 16:09:08 -08:00 · 2019-01-28 15:53:29 -08:00
43 changed files with 942 additions and 696 deletions
--- a/.travis.yml
+++ b/.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
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -5,8 +5,30 @@ 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
+- 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
@@ -20,7 +42,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).
--- a/CMakeLists.txt
+++ b/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,8 @@ 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 "1b023c41bb2d6dbbf493c202ed81f06c84d5b51b"
+    #GIT_TAG "1.0.1"
    CMAKE_ARGS += -DCMAKE_INSTALL_PREFIX=${INSTALL_DIR} -DBUILD_TESTING=OFF
 )
 add_library(libcjwt STATIC SHARED IMPORTED)
@@ -218,7 +204,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)
--- a/src/ParodusInternal.h
+++ b/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
@@ -106,7 +107,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;

@@ -132,6 +134,7 @@ typedef struct {
 /*----------------------------------------------------------------------------*/
 /*                            File Scoped Variables                           */
 /*----------------------------------------------------------------------------*/
+extern bool g_shutdown;
 extern ParodusMsg *ParodusMsgQ;
 int numLoops;
 /*----------------------------------------------------------------------------*/
--- a/src/client_list.c
+++ b/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;
 }
--- a/src/client_list.h
+++ b/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
 }
--- a/src/config.c
+++ b/src/config.c
@@ -615,7 +615,6 @@ void createNewAuthToken(char *newToken, size_t len)
 {
 	//Call create script
 	char output[12] = {'\0'};
-	memset (newToken, 0, len);
 	execute_token_script(output,get_parodus_cfg()->token_acquisition_script,sizeof(output),get_parodus_cfg()->hw_mac,get_parodus_cfg()->hw_serial_number);
  	if (strlen(output)>0  && strcmp(output,"SUCCESS")==0)
 	{
@@ -637,7 +636,7 @@ void getAuthToken(ParodusCfg *cfg)
 {
 	//local var to update cfg->webpa_auth_token only in success case
 	char output[4069] = {'\0'} ;
-	memset (cfg->webpa_auth_token, 0, sizeof(cfg->webpa_auth_token));
+	
 	if( strlen(cfg->token_read_script) !=0 && strlen(cfg->token_acquisition_script) !=0)
    	{
 		execute_token_script(output,cfg->token_read_script,sizeof(output),cfg->hw_mac,cfg->hw_serial_number);
@@ -694,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);
 }
--- a/src/config.h
+++ b/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"
--- a/src/conn_interface.c
+++ b/src/conn_interface.c
@@ -51,12 +51,19 @@
 /*----------------------------------------------------------------------------*/
 /*                            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                             */
 /*----------------------------------------------------------------------------*/
@@ -68,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
@@ -93,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) 
    {
@@ -107,6 +115,8 @@ void createSocketConnection(void (* initKeypress)())

    seshat_registered = __registerWithSeshat();
    
+    clock_gettime(CLOCK_REALTIME, &start_svc_alive_timer);
+
    do
    {
        struct timespec start, stop, diff;
@@ -157,21 +167,47 @@ 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());
+       } 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();
 }

+void shutdownSocketConnection(void) {
+   g_shutdown = true;
+}
+
--- a/src/conn_interface.h
+++ b/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
 }
--- a/src/connection.c
+++ b/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 );
@@ -192,19 +193,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;
 }  

@@ -383,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");
@@ -405,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;
@@ -412,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;
@@ -430,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);
@@ -503,14 +508,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 +519,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 +536,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,22 +550,23 @@ 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)
+	while (!g_shutdown)
 	{
 	  query_dns_status = find_servers (&conn_ctx.server_list);
 	  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 +583,12 @@ 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) && 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);
        free_header_info (&conn_ctx.header_info);
        free_server_list (&conn_ctx.server_list);
--- a/src/crud_interface.c
+++ b/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);
 		}
--- a/src/crud_interface.h
+++ b/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
 }
--- a/src/downstream.c
+++ b/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;
@@ -119,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)
@@ -138,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))
 							{
@@ -153,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);
@@ -183,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;
                        }
@@ -224,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;
                }
@@ -234,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
        {
@@ -244,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;
+    }
+}
--- a/src/main.c
+++ b/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();
 	}
 	
 }
--- a/src/partners_check.c
+++ b/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]);
                }
--- a/src/service_alive.c
+++ b/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;
 }
--- a/src/service_alive.h
+++ b/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
 }
--- a/src/spin_thread.c
+++ b/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);
+}
--- a/src/spin_thread.h
+++ b/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
 }
--- a/src/thread_tasks.c
+++ b/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);
--- a/src/token.c
+++ b/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);
 	
--- a/src/upstream.c
+++ b/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,10 +210,10 @@ void *processUpstreamMessage()
    reg_list_item_t *temp = NULL;
    int matchFlag = 0;
    int status = -1;
-    char *serviceName = 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())
    {
@@ -225,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)
@@ -244,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)
@@ -288,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)
@@ -321,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
                {
@@ -333,79 +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)
 						{
-							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(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)
+								ret = getDeviceId(&device_id, &device_id_len);
+								if(ret == 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)
+									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) )
 									{
-										ParodusInfo("Send upstreamMsg successfully to registered client %s\n", serviceName);
+										 /* 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) ) 
+										{
+											/* For Parodus CRUD queue. */
+											ParodusInfo("Create RetrieveMsg and add to parodus CRUD queue\n");
+											createUpstreamRetrieveMsg(msg, &retrieve_msg);
+											addCRUDmsgToQueue(retrieve_msg);
+										}
+										else
+										{
+											/* For nanomsg clients. */
+											getServiceNameAndSendResponse(msg, &message->msg, message->len);
+										}
 									}
 									else
 									{
-										ParodusError("Failed to send upstreamMsg to registered client %s\n", serviceName);
+										/* Not for this device. Send upstream */
+										ParodusInfo("sendUpstreamMsgToServer \n");
+										sendUpstreamMsgToServer(&message->msg, message->len);
+									}
+									if(device_id != NULL)
+									{
+										free(device_id);
+										device_id = NULL;
 									}
-									free(serviceName);
-									serviceName = NULL;
 								}
 								else
 								{
-									ParodusError("serviceName is NULL,not sending cloud-status response to client\n");
+									ParodusError("Failed to get device_id\n");
 								}
-							}
-							else
-							{
-								ParodusInfo("sendUpstreamMsgToServer \n");
-								sendUpstreamMsgToServer(&message->msg, message->len);
-							}
-							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
 						{
@@ -441,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);
@@ -450,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;
--- a/src/upstream.h
+++ b/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);
--- a/tests/CMakeLists.txt
+++ b/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)
--- a/tests/libpd_test.c
+++ b/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;
-}
--- a/tests/test_client_list.c
+++ b/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");	
--- a/tests/test_close_retry.c
+++ b/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);
--- a/tests/test_conn_interface.c
+++ b/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                                    */
 /*----------------------------------------------------------------------------*/
--- a/tests/test_connection.c
+++ b/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);


 /*----------------------------------------------------------------------------*/
@@ -64,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;
@@ -117,10 +117,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 +302,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));
 }


@@ -511,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);
@@ -580,6 +609,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 +702,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);

@@ -694,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;
@@ -703,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);
@@ -771,6 +816,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 +848,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 +870,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 +918,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);
 }

--- a/tests/test_crud_interface.c
+++ b/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                                    */
 /*----------------------------------------------------------------------------*/
--- a/tests/test_crud_internal.c
+++ b/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);
--- a/tests/test_downstream.c
+++ b/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();
 }
--- a/tests/test_downstream_more.c
+++ b/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 ) {
--- a/tests/test_heartBeatTimer.c
+++ b/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);
--- a/tests/test_partners_check.c
+++ b/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),
--- a/tests/test_service_alive.c
+++ b/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;
 }

--- a/tests/test_spin_thread_e.c
+++ b/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 )
--- a/tests/test_spin_thread_s.c
+++ b/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 )
--- a/tests/test_thread_tasks.c
+++ b/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;
--- a/tests/test_token.c
+++ b/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;
--- a/tests/test_token_stub.c
+++ b/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;
--- a/tests/test_upstream.c
+++ b/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,11 +786,66 @@ void test_processUpstreamMsg_sendToClient()

 	expect_function_call(wrp_free_struct);
 	processUpstreamMessage();
-    free(temp);
-    free(UpStreamMsgQ);
-    UpStreamMsgQ = NULL;
+	free(temp);
+	free(UpStreamMsgQ);
+	UpStreamMsgQ = NULL;
 }

+void test_processUpstreamMsg_serviceNameNULL()
+{
+	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                             */
 /*----------------------------------------------------------------------------*/
@@ -795,6 +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_processUpstreamMsg_serviceNameNULL),
+	cmocka_unit_test(err_processUpstreamMsg_deviceID)
    };

    return cmocka_run_group_tests(tests, NULL, NULL);
Author	SHA1	Message	Date
shilpa24balaji	b001964b71	Merge pull request #281 from sadhyama/parodus-interface To correct the spelling 'webpa-inteface-used' in convey header	2019-02-07 13:05:10 -08:00
Sadhyama Vengilat	c33bb6c1c3	To correct the spelling 'webpa-inteface-used' in convey header	2019-02-07 16:42:42 +05:30
shilpa24balaji	379e63be00	Merge pull request #280 from bill1600/svcalive Svcalive	2019-02-05 16:20:35 -08:00
bill1600	062fdc2ce0	Merge branch 'master' into svcalive	2019-01-28 16:09:08 -08:00
Bill Williams	200522c997	change svc alive back to separate thread. Shut it down with pthread_cond_timedwait	2019-01-28 15:53:29 -08:00
Bill Williams	62540e8db8	fix main loop to keep calling svc_alive_task during cloud disconnect retry	2019-01-25 14:08:27 -08:00
shilpa24balaji	f155dd601c	Merge pull request #278 from sadhyama/master Refactored Upstream Retrieve flow	2019-01-25 12:33:32 -08:00
shilpa24balaji	b9a5770bf9	Merge pull request #277 from gbuddappagari/master To enable valgrind	2019-01-14 10:30:30 -08:00
Sadhyama Vengilat	a717b0a909	Updated changelog	2019-01-12 12:33:58 +05:30
Sadhyama Vengilat	a7b1a145a5	Added unit tests for upstream refactor changes	2019-01-12 12:32:00 +05:30
Sadhyama Vengilat	7bc498b376	Refactor Upstream Retrieve mechanism	2019-01-12 12:15:02 +05:30
Gayathri	fe00e55fff	To enable valgrind	2019-01-11 09:52:55 +05:30
shilpa24balaji	66104dcabc	Merge pull request #275 from gbuddappagari/leak_fix Fixed issues in CRUD message processing	2019-01-10 09:24:24 -08:00
Gayathri	aedd9c972a	To fix memory corruption	2019-01-10 11:03:05 +05:30
shilpa24balaji	99b09179f2	Merge pull request #274 from bill1600/fixleaks Fixleaks	2019-01-08 23:49:35 -08:00
shilpa24balaji	92020ec6ec	Update CMakeLists.txt	2019-01-08 16:57:35 -08:00
shilpa24balaji	9057322fd7	Update CMakeLists.txt	2019-01-08 16:52:39 -08:00
Bill Williams	55c70f3c6f	resolve conflict	2019-01-08 14:49:31 -08:00
Bill Williams	7c07a5dd63	fix test_downstream	2019-01-08 14:38:55 -08:00
Bill Williams	cdd0a9e355	remove unneeded patch, and fix tests	2019-01-08 10:20:34 -08:00
shilpa24balaji	392085b01b	Merge pull request #271 from gbuddappagari/master To fix memory leak in upstream event message flow	2019-01-07 20:12:28 -08:00
Sadhyama Vengilat	bd67050c80	Fix build failure when FEATURE_DNS_QUERY is disabled	2019-01-08 08:56:00 +05:30
Sadhyama Vengilat	cea408a33c	Fix RETRIEVE response processing issue due to mac check	2019-01-08 08:43:19 +05:30
Gayathri	d1522346b4	To fix crash in CRUD request processing	2019-01-09 15:21:13 +05:30
Bill Williams	22feddeadc	properly shut down tasks, and fix memory leaks	2019-01-07 14:24:55 -08:00
Gayathri	df2729084d	To fix memory leak in upstream event message flow	2019-01-04 18:34:27 +05:30
shilpa24balaji	8a61e66456	Merge pull request #267 from bill1600/untask_svc_alive change serviceAlivetask to a function call from main	2018-12-18 14:09:17 -08:00
shilpa24balaji	8a737d6c18	Merge branch 'master' into untask_svc_alive	2018-12-18 14:08:50 -08:00
Bill Williams	905cb3bade	update change log	2018-12-18 13:39:25 -08:00
Shilpa Seshadri	a2d7dfeb95	Log time difference in connect time and boot time only during boot time and not during reconnect	2018-12-17 23:21:04 -08:00
shilpa24balaji	342a8ed250	Merge pull request #268 from bill1600/send_mutex Send mutex. put mutex lock into get_global_node	2018-12-13 15:41:20 -08:00
Bill Williams	97da13ffcc	put mutex lock into get_global_node	2018-12-13 14:07:06 -08:00
Bill Williams	590a43a053	Merge branch 'master' of https://github.com/Comcast/parodus into send_mutex	2018-12-13 13:42:51 -08:00
Bill Williams	0e6e833301	put mutex lock in get_global_node	2018-12-13 13:33:33 -08:00
shilpa24balaji	4525e39de9	Merge pull request #261 from bill1600/send_mutex Send mutex	2018-12-13 11:50:48 -08:00
shilpa24balaji	8d9e9c014a	Merge branch 'master' into send_mutex	2018-12-13 11:49:10 -08:00
Bill Williams	9f6ddb2b78	change serviceAlivetask to a function call from main	2018-12-12 14:23:10 -08:00
shilpa24balaji	b48680c358	Merge pull request #265 from bill1600/backoff Backoff	2018-12-11 15:15:30 -08:00
shilpa24balaji	2304951186	Merge pull request #266 from Comcast/partner_crash Added NULL check in validate_partner_id	2018-12-11 14:25:00 -08:00
Bill Williams	85004c165c	use max count not max delay time for backoff	2018-12-11 10:01:53 -08:00
Bill Williams	94b64af4a0	update changelog	2018-12-10 16:31:13 -08:00
Bill Williams	d190b8b10c	backoff retry to include find_servers in loop	2018-12-10 16:29:30 -08:00
Shilpa Seshadri	4e24197d24	Added NULL check in validate_partner_id	2018-12-09 23:04:55 -08:00
shilpa24balaji	f4d544fc40	Merge pull request #263 from Comcast/mac_null Added mac id null check in upstream retrieve message handling	2018-12-04 17:49:15 -08:00
Shilpa Seshadri	0972ff76fd	Fixed unit tests and updated change log	2018-12-04 16:03:53 -08:00
shilpa24balaji	ae8a9beb8e	Merge pull request #262 from vathikar/Mac_null Added Code and Test cases fot NULL check for MAC	2018-12-04 17:36:38 -08:00
Weston Schmidt	08ed64c822	Merge pull request #260 from Comcast/time_diff Added log for time difference in connect time and boot time	2018-12-04 15:01:27 -08:00
Bill Williams	c2f6a64569	Updated change log for send_mutex change	2018-11-30 09:38:41 -08:00
Bill Williams	2f0244af11	make client list and nn_sends thread safe	2018-11-30 09:35:05 -08:00
vathikar	f0f2903106	Added Code and Test cases fot NULL check for MAC	2018-11-30 15:30:52 +05:30
Shilpa Seshadri	8c83c4ce01	Added log for time difference in connect time and boot time	2018-11-29 21:59:00 -08:00
shilpa24balaji	385a38f10a	Merge pull request #259 from Comcast/tags Update CMake git tags	2018-11-21 18:53:49 -08:00
Shilpa Seshadri	4abad4dc13	Reverting to old nanomsg to fix cmake error	2018-11-21 18:14:29 -08:00
Shilpa Seshadri	cbe184e00e	Use git tag version instead of commit hash and fix unit tests	2018-11-21 17:54:43 -08:00
shilpa24balaji	25a6a6cc72	Merge pull request #257 from bill1600/revert_temp_ref_to_bill1600_seshat revert temporary CMake reference to bill1600/seshat	2018-11-15 17:13:50 -08:00
Bill Williams	aabe6be96a	revert temporary CMake reference to bill1600/seshat	2018-11-14 09:49:42 -08:00
shilpa24balaji	9aa18cea1c	Merge pull request #253 from vathikar/master Added NULL check for MAC	2018-11-12 12:51:17 -08:00
shilpa24balaji	2f762fe9a1	Merge pull request #256 from bill1600/revert_from_NNG revert from NNG	2018-11-12 12:38:46 -08:00
shilpa24balaji	93363b35db	Update CHANGELOG.md	2018-11-12 12:10:30 -08:00
Bill Williams	7e8edcaca4	revert from NNG	2018-11-12 11:55:01 -08:00
shilpa24balaji	81290ffd84	Merge pull request #251 from gbuddappagari/change_log Modify changelog for partnerid change	2018-11-09 22:38:06 -08:00
vathikar	8a69bc3955	Added NULL check for MAC	2018-11-07 13:11:56 +05:30
Gayathri	3634aa0a70	Modify changelog for partnerid change	2018-10-25 16:38:36 +05:30
Weston Schmidt	08a09106cf	Merge pull request #248 from Comcast/partner_icase Partner-id comparison made case insensitive	2018-10-17 11:46:55 -07:00
Shilpa Seshadri	f730a41cdb	Partner-id comparison made case insensitive	2018-10-16 23:34:29 -07:00
shilpa24balaji	f79eeb066a	Merge pull request #247 from bill1600/5xx_tweak add unit test for 503	2018-10-08 14:27:53 -07:00
shilpa24balaji	b4a8d8b2da	Merge branch 'master' into 5xx_tweak	2018-10-08 14:13:50 -07:00
Bill Williams	b42051bf96	Updated change log for the connection retry change	2018-10-08 13:48:02 -07:00
shilpa24balaji	cd1d1ba52b	Merge pull request #245 from Comcast/mem_fix Memory leak fixes	2018-10-08 13:41:55 -07:00
Bill Williams	44554b0dfa	add unit test for 503	2018-10-08 11:26:32 -07:00
Shilpa Seshadri	1addffdeb0	Memory leak fixes	2018-10-05 16:20:18 -07:00
shilpa24balaji	ce0c43d894	Merge pull request #242 from bill1600/nng_compat switch from nanomsg to NNG	2018-10-05 12:01:34 -07:00
Bill Williams	4f1952ddcc	Fix travis build by changing link order in tests/CMakeLists.txt	2018-10-05 10:55:51 -07:00
Bill Williams	eda3823ef0	Change CMakeLists.txt to use Comcast versions of seshat and libparodus	2018-10-05 07:57:27 -07:00
Bill Williams	774f308680	use NNG release v1.0.1	2018-10-04 15:09:04 -07:00
Bill Williams	130d8f2933	switch from nanomsg to NNG	2018-10-02 16:56:26 -07:00