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

Handle commands where no response is expected

Browse Source 
r2299 | dcl | 2008-01-30 13:46:32 +1100 (Wed, 30 Jan 2008) | 2 lines
This commit is contained in:
Douglas Clowes
2008-01-30 13:46:32 +11:00
parent e3a9c18322
commit fb967be939
1 changed files with 15 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
asyncqueue.c
View File

@@ -134,7 +134,7 @@ static int AQ_Reconnect(pAsyncQueue self)
return 1; return 1;
} }
static int PopCommand(pAsyncQueue self);
static int CommandTimeout(void* cntx, int mode); static int CommandTimeout(void* cntx, int mode);
static int DelayedStart(void* cntx, int mode); static int DelayedStart(void* cntx, int mode);
@@ -142,6 +142,7 @@ static int StartCommand(pAsyncQueue self)
{ {
pAQ_Cmd myCmd = self->command_head; pAQ_Cmd myCmd = self->command_head;
mkChannel* sock = self->pSock; mkChannel* sock = self->pSock;
int iRet = 0;
if (myCmd == NULL) if (myCmd == NULL)
return OKOK; return OKOK;
@@ -183,7 +184,6 @@ static int StartCommand(pAsyncQueue self)
while (NETAvailable(sock, 0)) { while (NETAvailable(sock, 0)) {
/* TODO: handle unsolicited input */ /* TODO: handle unsolicited input */
char reply[1]; char reply[1];
int iRet;
iRet = NETRead(sock, reply, 1, 0); iRet = NETRead(sock, reply, 1, 0);
if (iRet < 0) { /* EOF */ if (iRet < 0) { /* EOF */
iRet = AQ_Reconnect(self); iRet = AQ_Reconnect(self);
@@ -191,6 +191,15 @@ static int StartCommand(pAsyncQueue self)
return 0; return 0;
} }
} }
myCmd->tran->txn_status == ATX_ACTIVE;
iRet = self->protocol->sendCommand(self->protocol, myCmd->tran);
/*
* Handle case of no response expected
*/
if (iRet > 0)
if (myCmd->tran->txn_status == ATX_COMPLETE)
return PopCommand(self);
/* /*
* Add a new command timeout timer * Add a new command timeout timer
*/ */
@@ -201,7 +210,7 @@ static int StartCommand(pAsyncQueue self)
NetWatchRegisterTimer(&self->nw_tmr, 30000, NetWatchRegisterTimer(&self->nw_tmr, 30000,
CommandTimeout, self); CommandTimeout, self);
myCmd->active = 1; myCmd->active = 1;
return self->protocol->sendCommand(self->protocol, myCmd->tran); return iRet;
} }
static int QueCommandHead(pAsyncQueue self, pAQ_Cmd cmd) static int QueCommandHead(pAsyncQueue self, pAQ_Cmd cmd)
@@ -251,6 +260,9 @@ static int PopCommand(pAsyncQueue self)
NetWatchRemoveTimer(self->nw_tmr); NetWatchRemoveTimer(self->nw_tmr);
self->nw_tmr = 0; self->nw_tmr = 0;
gettimeofday(&self->tvLastCmd, NULL); gettimeofday(&self->tvLastCmd, NULL);
/* Process any callback */
if (myCmd->tran->handleResponse)
myCmd->tran->handleResponse(myCmd->tran);
/* /*
* If this is not the last in queue, start transmission * If this is not the last in queue, start transmission
*/ */
@@ -281,8 +293,6 @@ static int CommandTimeout(void* cntx, int mode)
int iRet; int iRet;
iRet = self->protocol->handleEvent(self->protocol, myCmd->tran, AQU_TIMEOUT); iRet = self->protocol->handleEvent(self->protocol, myCmd->tran, AQU_TIMEOUT);
if (iRet == AQU_POP_CMD) { if (iRet == AQU_POP_CMD) {
if (myCmd->tran->handleResponse)
myCmd->tran->handleResponse(myCmd->tran);
PopCommand(self); /* remove command */ PopCommand(self); /* remove command */
} }
else if (iRet == AQU_RETRY_CMD) else if (iRet == AQU_RETRY_CMD)
@@ -328,8 +338,6 @@ static int MyCallback(void* context, int mode)
for (i = 0; i < nchars; ++i) { for (i = 0; i < nchars; ++i) {
iRet = self->protocol->handleInput(self->protocol, myCmd->tran, reply[i]); iRet = self->protocol->handleInput(self->protocol, myCmd->tran, reply[i]);
if (iRet == 0 || iRet == AQU_POP_CMD) { /* end of command */ if (iRet == 0 || iRet == AQU_POP_CMD) { /* end of command */
if (myCmd->tran->handleResponse)
myCmd->tran->handleResponse(myCmd->tran);
PopCommand(self); PopCommand(self);
break; break;
} }