Compare commits

...

8 Commits

Author SHA1 Message Date
Sadhyama Vengilat
8957afc6bc Revert traceroute changes 2023-04-28 17:45:36 +05:30
sadhyama
118e8ee32f Merge pull request #413 from shilpa24balaji/xmidt_send_loop
Xmidt send loop
2023-04-11 12:26:39 +05:30
Shilpa Seshadri
f7c9f483f9 Merge branch 'master' of https://github.com/xmidt-org/parodus into xmidt_send_loop 2023-04-05 19:04:56 +05:30
shilpa24balaji
502f56400e Merge pull request #418 from sadhyama/lowqos_retry
XmidtSendData to handle higher priority low qos
2023-03-30 20:24:10 +05:30
Sadhyama Vengilat
3e557ae4b0 To skip max queue size delete for higher priority low qos 2023-03-30 18:34:55 +05:30
Sadhyama Vengilat
0602fb243b XmidtSendData to handle higher priority low qos msgs to confirm send success 2023-03-28 18:45:31 +05:30
Shilpa Seshadri
0b0ba77bd6 Wait until close retry is in progress for xmidtEventSendToServer 2023-01-06 00:16:27 +05:30
Shilpa Seshadri
0561c67e5c To avoid continuous xmidtEventSendToServer retry loop 2022-12-15 20:42:34 +05:30
2 changed files with 54 additions and 21 deletions

View File

@@ -320,10 +320,7 @@ void *processUpstreamMessage()
}
else if(msgType == WRP_MSG_TYPE__EVENT)
{
(msg->u.event.headers != NULL && msg->u.event.headers->headers[0] != NULL && msg->u.event.headers->headers[1] != NULL) ? ParodusInfo(" Received upstream event data: dest '%s' traceParent: %s traceState: %s\n", msg->u.event.dest, msg->u.event.headers->headers[0], msg->u.event.headers->headers[1]) : ParodusInfo(" Received upstream event data: dest '%s'\n", msg->u.event.dest);
if(msg->u.event.transaction_uuid != NULL) {
ParodusInfo("transaction_uuid in event: %s\n", msg->u.event.transaction_uuid);
}
ParodusInfo(" Received upstream event data: dest '%s'\n", msg->u.event.dest);
partners_t *partnersList = NULL;
int j = 0;
@@ -382,7 +379,7 @@ void *processUpstreamMessage()
//Sending to server for msgTypes 3, 5, 6, 7, 8.
if( WRP_MSG_TYPE__REQ == msgType )
{
(msg->u.req.headers != NULL && msg->u.req.headers->headers[0] != NULL && msg->u.req.headers->headers[1] != NULL) ? ParodusInfo(" Received upstream data with MsgType: %d dest: '%s' transaction_uuid: %s traceParent: %s traceState: %s\n", msgType, msg->u.req.dest, msg->u.req.transaction_uuid, msg->u.req.headers->headers[0], msg->u.req.headers->headers[1]) : ParodusInfo(" Received upstream data with MsgType: %d dest: '%s' transaction_uuid: %s\n", msgType, msg->u.req.dest, msg->u.req.transaction_uuid);
ParodusInfo(" Received upstream data with MsgType: %d dest: '%s' transaction_uuid: %s\n", msgType, msg->u.req.dest, msg->u.req.transaction_uuid );
sendUpstreamMsgToServer(&message->msg, message->len);
}
else

View File

@@ -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");
}