Fixed license in new c file

Use rbus subscribe API instead of addListener API

CHANGELOG.md

@@ -6,6 +6,9 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 
 ## [Unreleased]
 - Add additional HTTP headers for call to Themis from Convey
+- Add callback handler for initial cloud connection status change event
+- Fix Parodus connection stuck on interface up down received together
+- Update to use nopoll version 1.0.3
 
 ## [1.1.4]
 - on connect retry, requery jwt only if it failed before 

CMakeLists.txt

@@ -86,7 +86,7 @@ add_dependencies(libtrower-base64 trower-base64)
 ExternalProject_Add(nopoll
     PREFIX ${PREFIX_DIR}/nopoll
     GIT_REPOSITORY https://github.com/Comcast/nopoll.git
-    GIT_TAG "1.0.2"
+    GIT_TAG "1.0.3"
     CONFIGURE_COMMAND COMMAND <SOURCE_DIR>/autogen.sh --prefix=${PREFIX}
                                                      --includedir=${INCLUDE_DIR}
                                                      --libdir=${LIBRARY_DIR}
@@ -253,6 +253,10 @@ if (FEATURE_DNS_QUERY)
 set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -DFEATURE_DNS_QUERY ")
 endif (FEATURE_DNS_QUERY)
 
+if (ENABLE_WEBCFGBIN)
+set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -DENABLE_WEBCFGBIN ")
+endif (ENABLE_WEBCFGBIN)
+
 link_directories ( ${LIBRARY_DIR} ${COMMON_LIBRARY_DIR} ${LIBRARY_DIR64} )
 add_subdirectory(src)
 if (BUILD_TESTING)

src/CMakeLists.txt

@@ -23,6 +23,9 @@ else()
 set(SOURCES ${SOURCES} seshat_interface_stub.c)
 endif (ENABLE_SESHAT)
 
+if (ENABLE_WEBCFGBIN)
+set(SOURCES ${SOURCES} upstream_rbus.c)
+endif (ENABLE_WEBCFGBIN)
 
 add_executable(parodus ${SOURCES})
 

src/conn_interface.c

@@ -112,6 +112,9 @@ void createSocketConnection(void (* initKeypress)())
     UpStreamMsgQ = NULL;
     StartThread(handle_upstream, &upstream_tid);
     StartThread(processUpstreamMessage, &upstream_msg_tid);
+    #ifdef ENABLE_WEBCFGBIN
+    subscribeRBUSevent();
+    #endif 
     ParodusMsgQ = NULL;
     StartThread(messageHandlerTask, &downstream_tid);
     StartThread(serviceAliveTask, &svc_alive_tid);
@@ -195,7 +198,8 @@ void createSocketConnection(void (* initKeypress)())
             }
             createNopollConnection(ctx, &server_list);
         }
-       } while(!get_close_retry() && !g_shutdown);
+	 //process exit only when g_shutdown is true.
+       } while(FOREVER() && !g_shutdown);
 
     pthread_mutex_lock (get_global_svc_mut());
     pthread_cond_signal (get_global_svc_con());

src/connection.c

@@ -62,6 +62,8 @@ enum {
 /*                            File Scoped Variables                           */
 /*----------------------------------------------------------------------------*/
 
+parodusOnConnStatusChangeHandler on_conn_status_change;
+
 parodusOnPingStatusChangeHandler on_ping_status_change;
 
 pthread_mutex_t backoff_delay_mut=PTHREAD_MUTEX_INITIALIZER;
@@ -697,7 +699,8 @@ int wait_while_interface_down()
   	  if (rtn != 0)
   	    ParodusError 
   	      ("Error on pthread_cond_wait (%d) in wait_while_interface_down\n", rtn);
-  	  if ((rtn != 0) || g_shutdown) {
+	  if (g_shutdown) {
+	    ParodusInfo("Received g_shutdown during interface down wait, returning\n");
 	    return -1;
 	  }
 	}
@@ -717,6 +720,7 @@ int createNopollConnection(noPollCtx *ctx, server_list_t *server_list)
   struct timespec connect_time,*connectTimePtr;
   connectTimePtr = &connect_time;
   backoff_timer_t backoff_timer;
+  static int init_conn_failure=1;
   
   if(ctx == NULL) {
         return nopoll_false;
@@ -752,6 +756,14 @@ int createNopollConnection(noPollCtx *ctx, server_list_t *server_list)
 	  }
 	  /* if we failed to connect, don't reuse the redirect server */	
       free_server (&conn_ctx.server_list->redirect);
+
+      /* On initial connect failure, invoke conn status change event as "failed" only 1 time*/
+      if((NULL != on_conn_status_change) && init && init_conn_failure)
+      {
+    	  on_conn_status_change("failed");
+	  init_conn_failure=0;
+      }
+
 #ifdef FEATURE_DNS_QUERY
       /* if we don't already have a valid jwt, look up server information */
       if (server_is_null (&conn_ctx.server_list->jwt))
@@ -776,6 +788,12 @@ int createNopollConnection(noPollCtx *ctx, server_list_t *server_list)
 	get_parodus_cfg()->cloud_status = CLOUD_STATUS_ONLINE;
 	ParodusInfo("cloud_status set as %s after successful connection\n", get_parodus_cfg()->cloud_status);
 
+	/* On initial connect success, invoke conn status change event as "success" */
+	if((NULL != on_conn_status_change) && init)
+	{
+		on_conn_status_change("success");
+	}
+
 	// Invoke the ping status change event callback as "received" ping
 	if(NULL != on_ping_status_change)
 	{
@@ -934,3 +952,8 @@ void registerParodusOnPingStatusChangeHandler(parodusOnPingStatusChangeHandler c
 	on_ping_status_change = callback_func;
 }
 
+void registerParodusOnConnStatusChangeHandler(parodusOnConnStatusChangeHandler callback_func)
+{
+	on_conn_status_change = callback_func;
+}
+

src/connection.h

@@ -37,6 +37,15 @@ extern "C" {
 #define SHUTDOWN_REASON_SYSTEM_RESTART  "system_restarting"
 #define SHUTDOWN_REASON_SIGTERM 		"SIGTERM"
 
+/**
+* parodusOnConnStatusChangeHandler - Function pointer
+* Used to define callback function to do additional processing
+* when websocket cloud connection status change event
+* i.e. "cloud-conn-status" as "fail" or "success"
+*/
+typedef void (*parodusOnConnStatusChangeHandler) (char * status);
+extern parodusOnConnStatusChangeHandler on_conn_status_change;
+
 /**
 * parodusOnPingStatusChangeHandler - Function pointer
 * Used to define callback function to do additional processing 
@@ -79,6 +88,9 @@ void set_cloud_disconnect_time(int disconnTime);
 void start_conn_in_progress (unsigned long start_time);
 void stop_conn_in_progress (void);
 
+// To Register parodusOnConnStatusChangeHandler Callback function
+void registerParodusOnConnStatusChangeHandler(parodusOnConnStatusChangeHandler on_conn_status_change);
+
 // To Register parodusOnPingStatusChangeHandler Callback function
 void registerParodusOnPingStatusChangeHandler(parodusOnPingStatusChangeHandler on_ping_status_change);
 

src/upstream.h

@@ -47,6 +47,7 @@ typedef struct UpStreamMsg__
 void packMetaData();
 void *handle_upstream();
 void *processUpstreamMessage();
+void registerRBUSlistener();
 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);

src/upstream_rbus.c

@@ -0,0 +1,123 @@
+/**
+ * Copyright 2021 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.
+ *
+ */
+/**
+ * @file upstream_rbus.c
+ *
+ * @description This is used for parodus-RBUS communication 
+ * to send notification events upstream to cloud.
+ *
+ */
+
+#include <stdlib.h>
+#include <rbus.h>
+#include "upstream.h"
+#include "ParodusInternal.h"
+#include "partners_check.h"
+
+#define WEBCFG_UPSTREAM_EVENT "Webconfig.Upstream"
+
+void processWebconfigUpstreamEvent(rbusHandle_t handle, rbusEvent_t const* event, rbusEventSubscription_t* subscription);
+
+void subscribeAsyncHandler( rbusHandle_t handle, rbusEventSubscription_t* subscription, rbusError_t error);
+
+/* API to register RBUS listener to receive messages from webconfig */
+void subscribeRBUSevent()
+{
+	rbusError_t err;
+    int rc = RBUS_ERROR_SUCCESS;
+	rbusHandle_t rbus_Handle;
+
+	err = rbus_open(&rbus_Handle, "parodus");
+	if (err)
+	{
+		ParodusError("rbus_open failed :%s\n", rbusError_ToString(err));
+		return;
+	}
+    rc = rbusEvent_SubscribeAsync(rbus_Handle,WEBCFG_UPSTREAM_EVENT,processWebconfigUpstreamEvent,subscribeAsyncHandler,"parodus",10*60);
+    if(rc != RBUS_ERROR_SUCCESS)
+        ParodusError("rbusEvent_Subscribe failed: %d, %s\n", rc, rbusError_ToString(rc));
+    else
+        ParodusInfo("rbusEvent_Subscribe was successful\n");
+}
+
+void processWebconfigUpstreamEvent(rbusHandle_t handle, rbusEvent_t const* event, rbusEventSubscription_t* subscription)
+{
+    (void)handle;
+    (void)subscription;
+    
+    int rv=-1;
+	wrp_msg_t *event_msg;
+	void *bytes;
+    int len;
+    rbusValue_t value = NULL;
+
+    value = rbusObject_GetValue(event->data, "value");
+    bytes = rbusValue_GetBytes(value, &len);
+
+	rv = wrp_to_struct( bytes, len, WRP_BYTES, &event_msg );
+	if(rv > 0)
+	{
+		ParodusInfo(" Received upstream event data: dest '%s'\n", event_msg->u.event.dest);
+		partners_t *partnersList = NULL;
+		int j = 0;
+
+		int ret = validate_partner_id(event_msg, &partnersList);
+		if(ret == 1)
+		{
+		    wrp_msg_t *eventMsg = (wrp_msg_t *) malloc(sizeof(wrp_msg_t));
+		    eventMsg->msg_type = event_msg->msg_type;
+		    eventMsg->u.event.content_type=event_msg->u.event.content_type;
+		    eventMsg->u.event.source=event_msg->u.event.source;
+		    eventMsg->u.event.dest=event_msg->u.event.dest;
+		    eventMsg->u.event.payload=event_msg->u.event.payload;
+		    eventMsg->u.event.payload_size=event_msg->u.event.payload_size;
+		    eventMsg->u.event.headers=event_msg->u.event.headers;
+		    eventMsg->u.event.metadata=event_msg->u.event.metadata;
+		    eventMsg->u.event.partner_ids = partnersList;
+
+		   int size = wrp_struct_to( eventMsg, WRP_BYTES, &bytes );
+		   if(size > 0)
+		   {
+		       sendUpstreamMsgToServer(&bytes, size);
+		   }
+		   free(eventMsg);
+		   free(bytes);
+		   bytes = NULL;
+		}
+		else
+		{
+		   sendUpstreamMsgToServer((void **)(&bytes), 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;
+	}
+}
+
+void subscribeAsyncHandler( rbusHandle_t handle, rbusEventSubscription_t* subscription, rbusError_t error)
+{
+	ParodusInfo("subscribeAsyncHandler event %s, error %d - %s\n",subscription->eventName, error, rbusError_ToString(error));
+}

tests/test_conn_interface.c

@@ -44,7 +44,7 @@ pthread_mutex_t svc_mut=PTHREAD_MUTEX_INITIALIZER;
 pthread_cond_t svc_con=PTHREAD_COND_INITIALIZER;
 int numLoops;
 parodusOnPingStatusChangeHandler on_ping_status_change;
- 
+
 /*----------------------------------------------------------------------------*/
 /*                                   Mocks                                    */
 /*----------------------------------------------------------------------------*/
@@ -335,6 +335,7 @@ void test_createSocketConnection()
 
 void test_createSocketConnection1()
 {
+    numLoops =0;
     noPollCtx *ctx;
     ParodusCfg cfg;
     memset(&cfg,0, sizeof(ParodusCfg));
@@ -364,11 +365,11 @@ void test_createSocketConnection1()
     expect_function_call(nopoll_ctx_unref);
     expect_function_call(nopoll_cleanup_library);
     createSocketConnection(NULL);
-    
 }
 
 void test_PingMissIntervalTime()
 {
+    numLoops = 6;
     noPollCtx *ctx;
     ParodusCfg cfg;
     memset(&cfg,0,sizeof(ParodusCfg));
@@ -386,7 +387,6 @@ void test_PingMissIntervalTime()
     //Max ping timeout is 6 sec
     cfg.webpa_ping_timeout = 6;
     set_parodus_cfg(&cfg);
-    
     reset_close_retry();
     expect_function_call(nopoll_thread_handlers);
     
@@ -422,11 +422,11 @@ void test_PingMissIntervalTime()
     expect_function_call(nopoll_ctx_unref);
     expect_function_call(nopoll_cleanup_library);
     createSocketConnection(NULL);
-    
 }
 
 void err_createSocketConnection()
 {
+    numLoops =0;
     set_close_retry();
     reset_heartBeatTimer();
     expect_function_call(nopoll_thread_handlers);
@@ -459,6 +459,7 @@ void err_createSocketConnection()
 
 void test_createSocketConnection_cloud_disconn()
 {
+        numLoops =0;
 	ParodusCfg cfg;
 	memset(&cfg,0,sizeof(ParodusCfg));
 	cfg.cloud_disconnect = strdup("XPC");