mirror of
https://github.com/outbackdingo/parodus.git
synced 2026-01-28 02:20:02 +00:00
Compare commits
10 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
85a0d9b71d | ||
|
|
03a5b96e0c | ||
|
|
066bdc180d | ||
|
|
118e8ee32f | ||
|
|
f7c9f483f9 | ||
|
|
502f56400e | ||
|
|
3e557ae4b0 | ||
|
|
0602fb243b | ||
|
|
0b0ba77bd6 | ||
|
|
0561c67e5c |
47
src/config.c
47
src/config.c
@@ -711,6 +711,53 @@ int parseCommandLine(int argc,char **argv,ParodusCfg * cfg)
|
||||
return 0;
|
||||
}
|
||||
|
||||
void free_cfg(ParodusCfg *cfg)
|
||||
{
|
||||
if(cfg != NULL)
|
||||
{
|
||||
if (cfg->mtls_client_cert_path != NULL )
|
||||
{
|
||||
free(cfg->mtls_client_cert_path);
|
||||
cfg->mtls_client_cert_path = NULL;
|
||||
}
|
||||
if(cfg->connection_health_file != NULL)
|
||||
{
|
||||
free(cfg->connection_health_file);
|
||||
cfg->connection_health_file = NULL;
|
||||
}
|
||||
if(cfg->token_server_url != NULL)
|
||||
{
|
||||
free(cfg->token_server_url );
|
||||
cfg->token_server_url = NULL;
|
||||
}
|
||||
if(cfg->mtls_client_key_path != NULL)
|
||||
{
|
||||
free(cfg->mtls_client_key_path);
|
||||
cfg->mtls_client_key_path = NULL;
|
||||
}
|
||||
if(cfg->client_cert_path != NULL)
|
||||
{
|
||||
free(cfg->client_cert_path);
|
||||
cfg->client_cert_path = NULL;
|
||||
}
|
||||
if(cfg->crud_config_file != NULL)
|
||||
{
|
||||
free(cfg->crud_config_file);
|
||||
cfg->crud_config_file = NULL;
|
||||
}
|
||||
if(cfg->close_reason_file != NULL)
|
||||
{
|
||||
free(cfg->close_reason_file);
|
||||
cfg->close_reason_file = NULL;
|
||||
}
|
||||
if(cfg->cloud_disconnect != NULL)
|
||||
{
|
||||
free(cfg->cloud_disconnect);
|
||||
cfg->cloud_disconnect = NULL;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
void setDefaultValuesToCfg(ParodusCfg *cfg)
|
||||
{
|
||||
if(cfg == NULL)
|
||||
|
||||
@@ -135,7 +135,7 @@ void loadParodusCfg(ParodusCfg * config,ParodusCfg *cfg);
|
||||
* or -1 if error
|
||||
*/
|
||||
int parseCommandLine(int argc,char **argv,ParodusCfg * cfg);
|
||||
|
||||
void free_cfg(ParodusCfg *cfg);
|
||||
void setDefaultValuesToCfg(ParodusCfg *cfg);
|
||||
// Accessor for the global config structure.
|
||||
ParodusCfg *get_parodus_cfg(void);
|
||||
|
||||
@@ -231,6 +231,9 @@ void createSocketConnection(void (* initKeypress)())
|
||||
nopoll_ctx_unref(ctx);
|
||||
nopoll_cleanup_library();
|
||||
curl_global_cleanup();
|
||||
clear_metadata();
|
||||
rdk_logger_deinit();
|
||||
free_server_list(&server_list);
|
||||
}
|
||||
|
||||
void shutdownSocketConnection(char *reason) {
|
||||
|
||||
@@ -131,7 +131,7 @@ int main( int argc, char **argv)
|
||||
curl_global_init(CURL_GLOBAL_DEFAULT);
|
||||
|
||||
createSocketConnection( NULL);
|
||||
|
||||
free_cfg(cfg);
|
||||
return 0;
|
||||
}
|
||||
|
||||
|
||||
@@ -110,6 +110,11 @@ void packMetaData()
|
||||
ParodusError("Failed to encode metadata\n");
|
||||
}
|
||||
}
|
||||
|
||||
void clear_metadata(){
|
||||
if(metadataPack != NULL)
|
||||
free(metadataPack);
|
||||
}
|
||||
|
||||
/*
|
||||
* @brief To handle UpStream messages which is received from nanomsg server socket
|
||||
|
||||
@@ -59,6 +59,7 @@ int subscribeCurrentActiveInterfaceEvent();
|
||||
UpStreamMsg * get_global_UpStreamMsgQ(void);
|
||||
pthread_cond_t *get_global_nano_con(void);
|
||||
pthread_mutex_t *get_global_nano_mut(void);
|
||||
void clear_metadata();
|
||||
|
||||
#ifdef __cplusplus
|
||||
}
|
||||
|
||||
@@ -31,6 +31,7 @@
|
||||
#include "config.h"
|
||||
#include "time.h"
|
||||
#include "heartBeat.h"
|
||||
#include "close_retry.h"
|
||||
|
||||
static pthread_t processThreadId = 0;
|
||||
static unsigned int XmidtQsize = 0;
|
||||
@@ -70,6 +71,22 @@ bool highQosValueCheck(int qos)
|
||||
return false;
|
||||
}
|
||||
|
||||
//To handle high priority low qos message to confirm send success and ignore cloud ack.
|
||||
bool higherPriorityLowQosCheck(int qos)
|
||||
{
|
||||
if(qos > 20 && qos < 25)
|
||||
{
|
||||
ParodusInfo("The low qos msg with higher priority\n");
|
||||
return true;
|
||||
}
|
||||
else
|
||||
{
|
||||
ParodusPrint("The qos is not higher priority low qos\n");
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
XmidtMsg * get_global_xmidthead(void)
|
||||
{
|
||||
XmidtMsg *tmp = NULL;
|
||||
@@ -122,9 +139,9 @@ void decrement_XmidtQsize()
|
||||
int checkCloudConn()
|
||||
{
|
||||
int ret = 1;
|
||||
if (!cloud_status_is_online ())
|
||||
if (get_close_retry() || !cloud_status_is_online ())
|
||||
{
|
||||
ParodusInfo("cloud status is not online, wait till connection up\n");
|
||||
ParodusInfo("close_retry is in progress or cloud status is not online, wait till connection up\n");
|
||||
|
||||
int rv;
|
||||
struct timespec ts;
|
||||
@@ -214,8 +231,15 @@ int xmidtQOptmize()
|
||||
{
|
||||
if(get_XmidtQsize() > 0 && get_XmidtQsize() == get_parodus_cfg()->max_queue_size)
|
||||
{
|
||||
ParodusInfo("Max queue size reached, delete low qos %d transid %s\n", tempMsg->u.event.qos, tempMsg->u.event.transaction_uuid);
|
||||
del = 2;
|
||||
if(higherPriorityLowQosCheck(tempMsg->u.event.qos))
|
||||
{
|
||||
ParodusInfo("Skip max queue size delete for qos %d transid %s\n", tempMsg->u.event.qos, tempMsg->u.event.transaction_uuid);
|
||||
}
|
||||
else
|
||||
{
|
||||
ParodusInfo("Max queue size reached, delete low qos %d transid %s\n", tempMsg->u.event.qos, tempMsg->u.event.transaction_uuid);
|
||||
del = 2;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -746,10 +770,9 @@ int sendXmidtEventToServer(XmidtMsg *msgnode, wrp_msg_t * msg, rbusMethodAsyncHa
|
||||
while(sendRetStatus) //If SendMessage is failed condition
|
||||
{
|
||||
ParodusError("sendXmidtEventToServer is Failed\n");
|
||||
if(highQosValueCheck(qos))
|
||||
if((highQosValueCheck(qos)) || (higherPriorityLowQosCheck(qos)))
|
||||
{
|
||||
ParodusPrint("The event is having high qos retry again\n");
|
||||
ParodusInfo("Wait till connection is Up\n");
|
||||
ParodusPrint("The event is having high qos retry again, wait till connection is Up\n");
|
||||
rv = checkCloudConn();
|
||||
if(rv == 2)
|
||||
{
|
||||
@@ -757,7 +780,7 @@ int sendXmidtEventToServer(XmidtMsg *msgnode, wrp_msg_t * msg, rbusMethodAsyncHa
|
||||
break;
|
||||
}
|
||||
ParodusInfo("Received cloud status signal proceed to retry\n");
|
||||
printSendMsgData("send to server after cloud reconnect", notif_wrp_msg->u.event.qos, notif_wrp_msg->u.event.dest, notif_wrp_msg->u.event.transaction_uuid);
|
||||
printSendMsgData("send to server after cloud reconnect", notif_wrp_msg->u.event.qos, notif_wrp_msg->u.event.dest, notif_wrp_msg->u.event.transaction_uuid);
|
||||
}
|
||||
else
|
||||
{
|
||||
@@ -797,13 +820,26 @@ int sendXmidtEventToServer(XmidtMsg *msgnode, wrp_msg_t * msg, rbusMethodAsyncHa
|
||||
}
|
||||
else
|
||||
{
|
||||
ParodusInfo("Low qos event, send success callback and delete\n");
|
||||
mapXmidtStatusToStatusMessage(DELIVERED_SUCCESS, &errorMsg);
|
||||
ParodusPrint("statusMsg is %s\n",errorMsg);
|
||||
createOutParamsandSendAck(msg, asyncHandle, errorMsg, DELIVERED_SUCCESS, NULL, RBUS_ERROR_SUCCESS);
|
||||
//print_xmidMsg_list();
|
||||
updateXmidtState(msgnode, DELETE);
|
||||
print_xmidMsg_list();
|
||||
if(higherPriorityLowQosCheck(qos))
|
||||
{
|
||||
ParodusInfo("Higher priority low qos send success, ignore cloud ack\n");
|
||||
mapXmidtStatusToStatusMessage(DELIVERED_SUCCESS, &errorMsg);
|
||||
printSendMsgData(errorMsg, notif_wrp_msg->u.event.qos, notif_wrp_msg->u.event.dest, notif_wrp_msg->u.event.transaction_uuid);
|
||||
createOutParamsandSendAck(msg, asyncHandle, errorMsg, DELIVERED_SUCCESS, NULL, RBUS_ERROR_SUCCESS);
|
||||
ParodusPrint("update state to DELETE\n");
|
||||
updateXmidtState(msgnode, DELETE);
|
||||
print_xmidMsg_list();
|
||||
}
|
||||
else
|
||||
{
|
||||
ParodusInfo("Low qos event, send success callback and delete\n");
|
||||
mapXmidtStatusToStatusMessage(DELIVERED_SUCCESS, &errorMsg);
|
||||
ParodusPrint("statusMsg is %s\n",errorMsg);
|
||||
createOutParamsandSendAck(msg, asyncHandle, errorMsg, DELIVERED_SUCCESS, NULL, RBUS_ERROR_SUCCESS);
|
||||
//print_xmidMsg_list();
|
||||
updateXmidtState(msgnode, DELETE);
|
||||
print_xmidMsg_list();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1631,7 +1667,7 @@ void checkMaxQandOptimize(XmidtMsg *xmdMsg)
|
||||
wrp_msg_t * tempMsg = temp->msg;
|
||||
qos = tempMsg->u.event.qos;
|
||||
ParodusPrint("qos is %d\n", qos);
|
||||
if(highQosValueCheck(qos))
|
||||
if((highQosValueCheck(qos)) || (higherPriorityLowQosCheck(qos)))
|
||||
{
|
||||
ParodusPrint("High qos msg, skip delete\n");
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user