linse/sics
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Purge the asyncqueue command queue periodically while disconnected

Browse Source
This commit is contained in:
Douglas Clowes
2013-07-30 13:33:26 +10:00
parent 606e73b221
commit 77bb495fe2
1 changed files with 4 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
asyncqueue.c
View File

@@ -192,6 +192,8 @@ static int TimedReconnect(void *cntx, int mode)
SICSLogPrintf(eStatus, "Function: %s:%s\n", self->queue_name, SICSLogPrintf(eStatus, "Function: %s:%s\n", self->queue_name,
__func__); __func__);
AQ_Purge(self);
/* TODO: if self->pSock is NULL we haven't connected yet */ /* TODO: if self->pSock is NULL we haven't connected yet */
iRet = NETReconnect(self->pSock); iRet = NETReconnect(self->pSock);
@@ -234,6 +236,7 @@ static int TimedReconnect(void *cntx, int mode)
self->state = eAsyncConnected; self->state = eAsyncConnected;
snprintf(line, 132, "Reconnect on AsyncQueue '%s'", self->queue_name); snprintf(line, 132, "Reconnect on AsyncQueue '%s'", self->queue_name);
SICSLogWrite(line, eStatus); SICSLogWrite(line, eStatus);
AQ_Purge(self);
AQ_Notify(self, AQU_RECONNECT); AQ_Notify(self, AQU_RECONNECT);
return 1; return 1;
} }
@@ -292,6 +295,7 @@ static int AQ_Reconnect(pAsyncQueue self)
self->state = eAsyncConnected; self->state = eAsyncConnected;
snprintf(line, 132, "Reconnect on AsyncQueue '%s'", self->queue_name); snprintf(line, 132, "Reconnect on AsyncQueue '%s'", self->queue_name);
SICSLogWrite(line, eStatus); SICSLogWrite(line, eStatus);
AQ_Purge(self);
AQ_Notify(self, AQU_RECONNECT); AQ_Notify(self, AQU_RECONNECT);
return 1; return 1;
} }
@@ -387,26 +391,6 @@ static int StartCommand(pAsyncQueue self)
static int QueCommandHead(pAsyncQueue self, pAQ_Cmd cmd) static int QueCommandHead(pAsyncQueue self, pAQ_Cmd cmd)
{ {
cmd->next = NULL; cmd->next = NULL;
#if 0
if (self->state != eAsyncConnected) {
if (cmd->tran->handleResponse) {
cmd->tran->txn_status == ATX_TIMEOUT; /* TODO should be ATX_DISCO */
cmd->tran->handleResponse(cmd->tran);
}
/*
* Remove this transaction from the queue
*/
if (cmd->next) {
self->command_head = cmd->next;
} else
self->command_head = self->command_tail = NULL;
free(cmd->tran->out_buf);
free(cmd->tran->inp_buf);
free(cmd->tran);
free(cmd);
return 0;
}
#endif
/* /*
* If the command queue is empty, start transmission * If the command queue is empty, start transmission
*/ */
@@ -430,26 +414,6 @@ static int QueCommandHead(pAsyncQueue self, pAQ_Cmd cmd)
static int QueCommand(pAsyncQueue self, pAQ_Cmd cmd) static int QueCommand(pAsyncQueue self, pAQ_Cmd cmd)
{ {
cmd->next = NULL; cmd->next = NULL;
#if 0
if (self->state != eAsyncConnected) {
if (cmd->tran->handleResponse) {
cmd->tran->txn_status == ATX_TIMEOUT; /* TODO should be ATX_DISCO */
cmd->tran->handleResponse(cmd->tran);
}
/*
* Remove this transaction from the queue
*/
if (cmd->next) {
self->command_head = cmd->next;
} else
self->command_head = self->command_tail = NULL;
free(cmd->tran->out_buf);
free(cmd->tran->inp_buf);
free(cmd->tran);
free(cmd);
return 0;
}
#endif
/* /*
* If the command queue is empty, start transmission * If the command queue is empty, start transmission
*/ */