- Adapted indenation to new agreed upon system
- Added support for second generation scriptcontext based counter
This commit is contained in:
koennecke
364
asyncqueue.c
364
asyncqueue.c
@@ -25,38 +25,38 @@
|
||||
typedef struct __async_command AQ_Cmd, *pAQ_Cmd;
|
||||
|
||||
struct __async_command {
|
||||
pAQ_Cmd next;
|
||||
pAsyncTxn tran;
|
||||
pAsyncUnit unit;
|
||||
int timeout;
|
||||
int retries;
|
||||
int active;
|
||||
pAQ_Cmd next;
|
||||
pAsyncTxn tran;
|
||||
pAsyncUnit unit;
|
||||
int timeout;
|
||||
int retries;
|
||||
int active;
|
||||
};
|
||||
|
||||
struct __AsyncUnit {
|
||||
pAsyncUnit next;
|
||||
pAsyncUnit next;
|
||||
pAsyncQueue queue;
|
||||
AQU_Notify notify_func;
|
||||
void* notify_cntx;
|
||||
AQU_Notify notify_func;
|
||||
void *notify_cntx;
|
||||
};
|
||||
|
||||
struct __AsyncQueue {
|
||||
pObjectDescriptor pDes;
|
||||
char* queue_name;
|
||||
char* pHost;
|
||||
int iPort;
|
||||
int iDelay; /* intercommand delay in milliseconds */
|
||||
int timeout;
|
||||
int retries;
|
||||
struct timeval tvLastCmd; /* time of completion of last command */
|
||||
int unit_count; /* number of units connected */
|
||||
pAsyncUnit units; /* head of unit chain */
|
||||
pAQ_Cmd command_head; /* first/next command in queue */
|
||||
pAQ_Cmd command_tail; /* last command in queue */
|
||||
pNWContext nw_ctx; /* NetWait context handle */
|
||||
pNWTimer nw_tmr; /* NetWait timer handle */
|
||||
mkChannel* pSock; /* socket address */
|
||||
pAsyncProtocol protocol;
|
||||
char *queue_name;
|
||||
char *pHost;
|
||||
int iPort;
|
||||
int iDelay; /* intercommand delay in milliseconds */
|
||||
int timeout;
|
||||
int retries;
|
||||
struct timeval tvLastCmd; /* time of completion of last command */
|
||||
int unit_count; /* number of units connected */
|
||||
pAsyncUnit units; /* head of unit chain */
|
||||
pAQ_Cmd command_head; /* first/next command in queue */
|
||||
pAQ_Cmd command_tail; /* last command in queue */
|
||||
pNWContext nw_ctx; /* NetWait context handle */
|
||||
pNWTimer nw_tmr; /* NetWait timer handle */
|
||||
mkChannel *pSock; /* socket address */
|
||||
pAsyncProtocol protocol;
|
||||
};
|
||||
|
||||
static pAsyncQueue queue_array[FD_SETSIZE];
|
||||
@@ -66,39 +66,37 @@ static int queue_index = 0;
|
||||
CreateSocketAdress stolen from Tcl. Thanks to John Ousterhout
|
||||
*/
|
||||
|
||||
static int
|
||||
CreateSocketAdress(
|
||||
struct sockaddr_in *sockaddrPtr, /* Socket address */
|
||||
char *host, /* Host. NULL implies INADDR_ANY */
|
||||
int port) /* Port number */
|
||||
{
|
||||
struct hostent *hostent; /* Host database entry */
|
||||
struct in_addr addr; /* For 64/32 bit madness */
|
||||
static int CreateSocketAdress(struct sockaddr_in *sockaddrPtr, /* Socket address */
|
||||
char *host, /* Host. NULL implies INADDR_ANY */
|
||||
int port)
|
||||
{ /* Port number */
|
||||
struct hostent *hostent; /* Host database entry */
|
||||
struct in_addr addr; /* For 64/32 bit madness */
|
||||
|
||||
(void) memset((char *) sockaddrPtr, '\0', sizeof(struct sockaddr_in));
|
||||
sockaddrPtr->sin_family = AF_INET;
|
||||
sockaddrPtr->sin_port = htons((unsigned short) (port & 0xFFFF));
|
||||
if (host == NULL) {
|
||||
addr.s_addr = INADDR_ANY;
|
||||
(void) memset((char *) sockaddrPtr, '\0', sizeof(struct sockaddr_in));
|
||||
sockaddrPtr->sin_family = AF_INET;
|
||||
sockaddrPtr->sin_port = htons((unsigned short) (port & 0xFFFF));
|
||||
if (host == NULL) {
|
||||
addr.s_addr = INADDR_ANY;
|
||||
} else {
|
||||
hostent = gethostbyname(host);
|
||||
if (hostent != NULL) {
|
||||
memcpy((char *) &addr,
|
||||
(char *) hostent->h_addr_list[0], (size_t) hostent->h_length);
|
||||
} else {
|
||||
hostent = gethostbyname(host);
|
||||
if (hostent != NULL) {
|
||||
memcpy((char *) &addr,
|
||||
(char *) hostent->h_addr_list[0], (size_t) hostent->h_length);
|
||||
} else {
|
||||
addr.s_addr = inet_addr(host);
|
||||
if (addr.s_addr == (unsigned long)-1) {
|
||||
return 0; /* error */
|
||||
}
|
||||
}
|
||||
addr.s_addr = inet_addr(host);
|
||||
if (addr.s_addr == (unsigned long) -1) {
|
||||
return 0; /* error */
|
||||
}
|
||||
}
|
||||
/*
|
||||
* There is a rumor that this assignment may require care on
|
||||
* some 64 bit machines.
|
||||
*/
|
||||
}
|
||||
/*
|
||||
* There is a rumor that this assignment may require care on
|
||||
* some 64 bit machines.
|
||||
*/
|
||||
|
||||
sockaddrPtr->sin_addr.s_addr = addr.s_addr;
|
||||
return 1;
|
||||
sockaddrPtr->sin_addr.s_addr = addr.s_addr;
|
||||
return 1;
|
||||
}
|
||||
|
||||
static void AQ_Notify(pAsyncQueue self, int event)
|
||||
@@ -129,14 +127,14 @@ static int AQ_Reconnect(pAsyncQueue self)
|
||||
return 1;
|
||||
}
|
||||
|
||||
static int CommandTimeout(void* cntx, int mode);
|
||||
static int DelayedStart(void* cntx, int mode);
|
||||
static int CommandTimeout(void *cntx, int mode);
|
||||
static int DelayedStart(void *cntx, int mode);
|
||||
static int PopCommand(pAsyncQueue self);
|
||||
|
||||
static int StartCommand(pAsyncQueue self)
|
||||
{
|
||||
pAQ_Cmd myCmd = self->command_head;
|
||||
mkChannel* sock = self->pSock;
|
||||
mkChannel *sock = self->pSock;
|
||||
|
||||
if (myCmd == NULL)
|
||||
return OKOK;
|
||||
@@ -166,8 +164,7 @@ static int StartCommand(pAsyncQueue self)
|
||||
int delay = when.tv_sec - now.tv_sec;
|
||||
delay *= 1000;
|
||||
delay += (when.tv_usec - now.tv_usec + (1000 - 1)) / 1000;
|
||||
NetWatchRegisterTimer(&self->nw_tmr, delay,
|
||||
DelayedStart, self);
|
||||
NetWatchRegisterTimer(&self->nw_tmr, delay, DelayedStart, self);
|
||||
return OKOK;
|
||||
}
|
||||
}
|
||||
@@ -180,12 +177,12 @@ static int StartCommand(pAsyncQueue self)
|
||||
char reply[1];
|
||||
int iRet;
|
||||
iRet = NETRead(sock, reply, 1, 0);
|
||||
if (iRet < 0) { /* EOF */
|
||||
if (iRet < 0) { /* EOF */
|
||||
iRet = AQ_Reconnect(self);
|
||||
if (iRet <= 0) {
|
||||
myCmd->tran->txn_state = ATX_DISCO;
|
||||
if(myCmd->tran->handleResponse){
|
||||
myCmd->tran->handleResponse(myCmd->tran);
|
||||
if (myCmd->tran->handleResponse) {
|
||||
myCmd->tran->handleResponse(myCmd->tran);
|
||||
}
|
||||
PopCommand(self);
|
||||
return 0;
|
||||
@@ -197,10 +194,9 @@ static int StartCommand(pAsyncQueue self)
|
||||
*/
|
||||
if (myCmd->timeout > 0)
|
||||
NetWatchRegisterTimer(&self->nw_tmr, myCmd->timeout,
|
||||
CommandTimeout, self);
|
||||
CommandTimeout, self);
|
||||
else
|
||||
NetWatchRegisterTimer(&self->nw_tmr, 30000,
|
||||
CommandTimeout, self);
|
||||
NetWatchRegisterTimer(&self->nw_tmr, 30000, CommandTimeout, self);
|
||||
myCmd->active = 1;
|
||||
return self->protocol->sendCommand(self->protocol, myCmd->tran);
|
||||
}
|
||||
@@ -219,8 +215,7 @@ static int QueCommandHead(pAsyncQueue self, pAQ_Cmd cmd)
|
||||
if (self->command_head->active) {
|
||||
cmd->next = self->command_head->next;
|
||||
self->command_head->next = cmd;
|
||||
}
|
||||
else {
|
||||
} else {
|
||||
cmd->next = self->command_head;
|
||||
self->command_head = cmd;
|
||||
}
|
||||
@@ -258,8 +253,7 @@ static int PopCommand(pAsyncQueue self)
|
||||
pAQ_Cmd pNew = myCmd->next;
|
||||
self->command_head = pNew;
|
||||
StartCommand(self);
|
||||
}
|
||||
else
|
||||
} else
|
||||
self->command_head = self->command_tail = NULL;
|
||||
free(myCmd->tran->out_buf);
|
||||
free(myCmd->tran->inp_buf);
|
||||
@@ -268,7 +262,7 @@ static int PopCommand(pAsyncQueue self)
|
||||
return 1;
|
||||
}
|
||||
|
||||
static int CommandTimeout(void* cntx, int mode)
|
||||
static int CommandTimeout(void *cntx, int mode)
|
||||
{
|
||||
pAsyncQueue self = (pAsyncQueue) cntx;
|
||||
pAQ_Cmd myCmd = self->command_head;
|
||||
@@ -276,24 +270,24 @@ static int CommandTimeout(void* cntx, int mode)
|
||||
if (myCmd->retries > 0) {
|
||||
--myCmd->retries;
|
||||
StartCommand(self);
|
||||
}
|
||||
else {
|
||||
} else {
|
||||
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 (myCmd->tran->handleResponse)
|
||||
myCmd->tran->handleResponse(myCmd->tran);
|
||||
PopCommand(self); /* remove command */
|
||||
}
|
||||
else if (iRet == AQU_RETRY_CMD)
|
||||
StartCommand(self); /* restart command */
|
||||
PopCommand(self); /* remove command */
|
||||
} else if (iRet == AQU_RETRY_CMD)
|
||||
StartCommand(self); /* restart command */
|
||||
else if (iRet == AQU_RECONNECT)
|
||||
AQ_Reconnect(self);
|
||||
}
|
||||
return 1;
|
||||
}
|
||||
|
||||
static int DelayedStart(void* cntx, int mode)
|
||||
static int DelayedStart(void *cntx, int mode)
|
||||
{
|
||||
pAsyncQueue self = (pAsyncQueue) cntx;
|
||||
self->nw_tmr = 0;
|
||||
@@ -301,7 +295,7 @@ static int DelayedStart(void* cntx, int mode)
|
||||
return 1;
|
||||
}
|
||||
|
||||
static int MyCallback(void* context, int mode)
|
||||
static int MyCallback(void *context, int mode)
|
||||
{
|
||||
pAsyncQueue self = (pAsyncQueue) context;
|
||||
|
||||
@@ -311,40 +305,40 @@ static int MyCallback(void* context, int mode)
|
||||
|
||||
iRet = NETRead(self->pSock, reply, 1, 0);
|
||||
/* printf(" iRet, char = %d, %d\n", iRet, (int)reply[0]); */
|
||||
if (iRet < 0) { /* EOF */
|
||||
if (iRet < 0) { /* EOF */
|
||||
iRet = AQ_Reconnect(self);
|
||||
if (iRet <= 0){
|
||||
/* changed to call handleResponse with a bad status code: MK
|
||||
*/
|
||||
pAQ_Cmd myCmd = self->command_head;
|
||||
if(myCmd){
|
||||
myCmd->tran->txn_state = ATX_DISCO;
|
||||
if(myCmd->tran->handleResponse){
|
||||
myCmd->tran->handleResponse(myCmd->tran);
|
||||
}
|
||||
PopCommand(self);
|
||||
}
|
||||
if (iRet <= 0) {
|
||||
/* changed to call handleResponse with a bad status code: MK
|
||||
*/
|
||||
pAQ_Cmd myCmd = self->command_head;
|
||||
if (myCmd) {
|
||||
myCmd->tran->txn_state = ATX_DISCO;
|
||||
if (myCmd->tran->handleResponse) {
|
||||
myCmd->tran->handleResponse(myCmd->tran);
|
||||
}
|
||||
PopCommand(self);
|
||||
}
|
||||
return iRet;
|
||||
}
|
||||
/* restart the command */
|
||||
StartCommand(self);
|
||||
return 1;
|
||||
}
|
||||
if (iRet == 0) { /* TODO: timeout or error */
|
||||
if (iRet == 0) { /* TODO: timeout or error */
|
||||
return 0;
|
||||
} else {
|
||||
pAQ_Cmd myCmd = self->command_head;
|
||||
if (myCmd) {
|
||||
iRet = self->protocol->handleInput(self->protocol, myCmd->tran, reply[0]);
|
||||
iRet =
|
||||
self->protocol->handleInput(self->protocol, myCmd->tran,
|
||||
reply[0]);
|
||||
if (iRet == 0 || iRet == AQU_POP_CMD) { /* end of command */
|
||||
if (myCmd->tran->handleResponse)
|
||||
myCmd->tran->handleResponse(myCmd->tran);
|
||||
PopCommand(self);
|
||||
}
|
||||
else if (iRet < 0) /* TODO: error */
|
||||
} else if (iRet < 0) /* TODO: error */
|
||||
;
|
||||
}
|
||||
else {
|
||||
} else {
|
||||
/* TODO: handle unsolicited input */
|
||||
}
|
||||
}
|
||||
@@ -391,9 +385,9 @@ int AsyncUnitEnqueueTxn(pAsyncUnit unit, pAsyncTxn pTxn)
|
||||
}
|
||||
|
||||
pAsyncTxn AsyncUnitPrepareTxn(pAsyncUnit unit,
|
||||
const char* command, int cmd_len,
|
||||
AsyncTxnHandler callback, void* context,
|
||||
int rsp_len)
|
||||
const char *command, int cmd_len,
|
||||
AsyncTxnHandler callback, void *context,
|
||||
int rsp_len)
|
||||
{
|
||||
pAsyncTxn myTxn = NULL;
|
||||
|
||||
@@ -406,10 +400,11 @@ pAsyncTxn AsyncUnitPrepareTxn(pAsyncUnit unit,
|
||||
memset(myTxn, 0, sizeof(AsyncTxn));
|
||||
if (unit->queue->protocol->prepareTxn) {
|
||||
int iRet;
|
||||
iRet = unit->queue->protocol->prepareTxn(unit->queue->protocol, myTxn, command, cmd_len, rsp_len);
|
||||
}
|
||||
else {
|
||||
myTxn->out_buf = (char*) malloc(cmd_len + 5);
|
||||
iRet =
|
||||
unit->queue->protocol->prepareTxn(unit->queue->protocol, myTxn,
|
||||
command, cmd_len, rsp_len);
|
||||
} else {
|
||||
myTxn->out_buf = (char *) malloc(cmd_len + 5);
|
||||
if (myTxn->out_buf == NULL) {
|
||||
SICSLogWrite("ERROR: Out of memory in AsyncUnitPrepareTxn", eError);
|
||||
free(myTxn);
|
||||
@@ -428,8 +423,8 @@ pAsyncTxn AsyncUnitPrepareTxn(pAsyncUnit unit,
|
||||
if (rsp_len == 0)
|
||||
myTxn->inp_buf = NULL;
|
||||
else {
|
||||
if(myTxn->inp_buf != NULL){
|
||||
free(myTxn->inp_buf);
|
||||
if (myTxn->inp_buf != NULL) {
|
||||
free(myTxn->inp_buf);
|
||||
}
|
||||
myTxn->inp_buf = malloc(rsp_len + 1);
|
||||
if (myTxn->inp_buf == NULL) {
|
||||
@@ -448,13 +443,12 @@ pAsyncTxn AsyncUnitPrepareTxn(pAsyncUnit unit,
|
||||
}
|
||||
|
||||
int AsyncUnitSendTxn(pAsyncUnit unit,
|
||||
const char* command, int cmd_len,
|
||||
AsyncTxnHandler callback, void* context,
|
||||
int rsp_len)
|
||||
const char *command, int cmd_len,
|
||||
AsyncTxnHandler callback, void *context, int rsp_len)
|
||||
{
|
||||
pAsyncTxn myTxn = NULL;
|
||||
myTxn = AsyncUnitPrepareTxn(unit, command, cmd_len,
|
||||
callback, context, rsp_len);
|
||||
callback, context, rsp_len);
|
||||
if (myTxn == NULL)
|
||||
return -1;
|
||||
return AsyncUnitEnqueueTxn(unit, myTxn);
|
||||
@@ -462,15 +456,16 @@ int AsyncUnitSendTxn(pAsyncUnit unit,
|
||||
|
||||
|
||||
typedef struct txn_s {
|
||||
char* transReply;
|
||||
char *transReply;
|
||||
int transWait;
|
||||
} TXN, *pTXN;
|
||||
|
||||
/**
|
||||
* \brief TransCallback is the callback for the general command transaction.
|
||||
*/
|
||||
static int TransCallback(pAsyncTxn pCmd) {
|
||||
char* resp = pCmd->inp_buf;
|
||||
static int TransCallback(pAsyncTxn pCmd)
|
||||
{
|
||||
char *resp = pCmd->inp_buf;
|
||||
int resp_len = pCmd->inp_idx;
|
||||
pTXN self = (pTXN) pCmd->cntx;
|
||||
|
||||
@@ -479,8 +474,7 @@ static int TransCallback(pAsyncTxn pCmd) {
|
||||
self->transReply[resp_len] = '\0';
|
||||
self->transReply[0] = '\0';
|
||||
self->transWait = -1;
|
||||
}
|
||||
else {
|
||||
} else {
|
||||
memcpy(self->transReply, resp, resp_len);
|
||||
self->transReply[resp_len] = '\0';
|
||||
self->transWait = 0;
|
||||
@@ -489,16 +483,14 @@ static int TransCallback(pAsyncTxn pCmd) {
|
||||
}
|
||||
|
||||
int AsyncUnitTransact(pAsyncUnit unit,
|
||||
const char* command, int cmd_len,
|
||||
char* response, int rsp_len)
|
||||
const char *command, int cmd_len,
|
||||
char *response, int rsp_len)
|
||||
{
|
||||
TXN txn;
|
||||
assert(unit);
|
||||
txn.transReply = response;
|
||||
txn.transWait = 1;
|
||||
AsyncUnitSendTxn(unit,
|
||||
command, cmd_len,
|
||||
TransCallback, &txn, rsp_len);
|
||||
AsyncUnitSendTxn(unit, command, cmd_len, TransCallback, &txn, rsp_len);
|
||||
while (txn.transWait == 1)
|
||||
TaskYield(pServ->pTasker);
|
||||
if (txn.transWait < 0)
|
||||
@@ -506,17 +498,17 @@ int AsyncUnitTransact(pAsyncUnit unit,
|
||||
return 1;
|
||||
}
|
||||
|
||||
int AsyncUnitWrite(pAsyncUnit unit, void* buffer, int buflen)
|
||||
int AsyncUnitWrite(pAsyncUnit unit, void *buffer, int buflen)
|
||||
{
|
||||
int iRet;
|
||||
mkChannel* sock;
|
||||
mkChannel *sock;
|
||||
assert(unit);
|
||||
assert(unit->queue);
|
||||
if (buflen > 0) {
|
||||
sock = AsyncUnitGetSocket(unit);
|
||||
iRet = NETWrite(sock, buffer, buflen);
|
||||
/* TODO handle errors */
|
||||
if (iRet < 0) { /* EOF */
|
||||
if (iRet < 0) { /* EOF */
|
||||
iRet = AQ_Reconnect(unit->queue);
|
||||
if (iRet == 0)
|
||||
return 0;
|
||||
@@ -525,7 +517,7 @@ int AsyncUnitWrite(pAsyncUnit unit, void* buffer, int buflen)
|
||||
return 1;
|
||||
}
|
||||
|
||||
void AsyncUnitSetNotify(pAsyncUnit unit, void* context, AQU_Notify notify)
|
||||
void AsyncUnitSetNotify(pAsyncUnit unit, void *context, AQU_Notify notify)
|
||||
{
|
||||
assert(unit);
|
||||
unit->notify_func = notify;
|
||||
@@ -578,7 +570,7 @@ void AsyncUnitSetProtocol(pAsyncUnit unit, pAsyncProtocol protocol)
|
||||
unit->queue->protocol = protocol;
|
||||
}
|
||||
|
||||
mkChannel* AsyncUnitGetSocket(pAsyncUnit unit)
|
||||
mkChannel *AsyncUnitGetSocket(pAsyncUnit unit)
|
||||
{
|
||||
assert(unit);
|
||||
assert(unit->queue);
|
||||
@@ -595,8 +587,8 @@ int AsyncUnitReconnect(pAsyncUnit unit)
|
||||
return iRet;
|
||||
}
|
||||
|
||||
int AsyncQueueAction(SConnection *pCon, SicsInterp *pSics,
|
||||
void *pData, int argc, char *argv[])
|
||||
int AsyncQueueAction(SConnection * pCon, SicsInterp * pSics,
|
||||
void *pData, int argc, char *argv[])
|
||||
{
|
||||
char line[132];
|
||||
pAsyncQueue self = (pAsyncQueue) pData;
|
||||
@@ -610,8 +602,7 @@ int AsyncQueueAction(SConnection *pCon, SicsInterp *pSics,
|
||||
cmd[0] = '\0';
|
||||
for (i = 2; i < argc; ++i) {
|
||||
j = snprintf(&cmd[idx], 10240 - idx, "%s%s",
|
||||
(i > 2) ? " " : "",
|
||||
argv[i]);
|
||||
(i > 2) ? " " : "", argv[i]);
|
||||
if (j < 0)
|
||||
break;
|
||||
idx += j;
|
||||
@@ -635,8 +626,7 @@ int AsyncQueueAction(SConnection *pCon, SicsInterp *pSics,
|
||||
snprintf(line, 132, "Invalid argument: %s", argv[2]);
|
||||
SCWrite(pCon, line, eError);
|
||||
return 0;
|
||||
}
|
||||
else {
|
||||
} else {
|
||||
if (delay < 0 || delay > 30000) {
|
||||
snprintf(line, 132, "Value out of range: %d", delay);
|
||||
SCWrite(pCon, line, eError);
|
||||
@@ -645,8 +635,7 @@ int AsyncQueueAction(SConnection *pCon, SicsInterp *pSics,
|
||||
self->iDelay = delay;
|
||||
return OKOK;
|
||||
}
|
||||
}
|
||||
else {
|
||||
} else {
|
||||
snprintf(line, 132, "%s.delay = %d", argv[0], self->iDelay);
|
||||
SCWrite(pCon, line, eValue);
|
||||
return OKOK;
|
||||
@@ -662,8 +651,7 @@ int AsyncQueueAction(SConnection *pCon, SicsInterp *pSics,
|
||||
snprintf(line, 132, "Invalid argument: %s", argv[2]);
|
||||
SCWrite(pCon, line, eError);
|
||||
return 0;
|
||||
}
|
||||
else {
|
||||
} else {
|
||||
if (timeout < 0 || timeout > 30000) {
|
||||
snprintf(line, 132, "Value out of range: %d", timeout);
|
||||
SCWrite(pCon, line, eError);
|
||||
@@ -672,8 +660,7 @@ int AsyncQueueAction(SConnection *pCon, SicsInterp *pSics,
|
||||
self->timeout = timeout;
|
||||
return OKOK;
|
||||
}
|
||||
}
|
||||
else {
|
||||
} else {
|
||||
snprintf(line, 132, "%s.timeout = %d", argv[0], self->timeout);
|
||||
SCWrite(pCon, line, eValue);
|
||||
return OKOK;
|
||||
@@ -689,8 +676,7 @@ int AsyncQueueAction(SConnection *pCon, SicsInterp *pSics,
|
||||
snprintf(line, 132, "Invalid argument: %s", argv[2]);
|
||||
SCWrite(pCon, line, eError);
|
||||
return 0;
|
||||
}
|
||||
else {
|
||||
} else {
|
||||
if (retries < 0 || retries > 30000) {
|
||||
snprintf(line, 132, "Value out of range: %d", retries);
|
||||
SCWrite(pCon, line, eError);
|
||||
@@ -699,8 +685,7 @@ int AsyncQueueAction(SConnection *pCon, SicsInterp *pSics,
|
||||
self->retries = retries;
|
||||
return OKOK;
|
||||
}
|
||||
}
|
||||
else {
|
||||
} else {
|
||||
snprintf(line, 132, "%s.retries = %d", argv[0], self->retries);
|
||||
SCWrite(pCon, line, eValue);
|
||||
return OKOK;
|
||||
@@ -713,40 +698,43 @@ int AsyncQueueAction(SConnection *pCon, SicsInterp *pSics,
|
||||
return 0;
|
||||
}
|
||||
|
||||
static pAsyncQueue AQ_Create(const char* host, const char* port)
|
||||
static pAsyncQueue AQ_Create(const char *host, const char *port)
|
||||
{
|
||||
int i;
|
||||
pAsyncQueue self = NULL;
|
||||
mkChannel* channel = NULL;
|
||||
mkChannel *channel = NULL;
|
||||
|
||||
if (host == NULL)
|
||||
return NULL;
|
||||
|
||||
/* try the AsyncQueue with this name */
|
||||
self = (pAsyncQueue) FindCommandData(pServ->pSics,(char *) host, "AsyncQueue");
|
||||
self =
|
||||
(pAsyncQueue) FindCommandData(pServ->pSics, (char *) host,
|
||||
"AsyncQueue");
|
||||
|
||||
/* try host and port */
|
||||
if (self == NULL && port) {
|
||||
int port_no = atoi(port);
|
||||
if (port_no == 0) {
|
||||
struct servent *sp=NULL;
|
||||
struct servent *sp = NULL;
|
||||
sp = getservbyname(port, NULL);
|
||||
if (sp)
|
||||
port_no = ntohs(sp->s_port);
|
||||
}
|
||||
if (port_no > 0) {
|
||||
struct sockaddr_in sa;
|
||||
if (CreateSocketAdress(&sa,(char *) host, port_no)) {
|
||||
if (CreateSocketAdress(&sa, (char *) host, port_no)) {
|
||||
/* look for queue with same address */
|
||||
for (i = 0; i < queue_index; ++i)
|
||||
if (queue_array[i]->pSock->adresse.sin_port == sa.sin_port
|
||||
&& queue_array[i]->pSock->adresse.sin_addr.s_addr == sa.sin_addr.s_addr) {
|
||||
&& queue_array[i]->pSock->adresse.sin_addr.s_addr ==
|
||||
sa.sin_addr.s_addr) {
|
||||
self = queue_array[i];
|
||||
break;
|
||||
}
|
||||
}
|
||||
if (self == NULL) {
|
||||
channel = NETConnectWithFlags((char *)host, port_no, 0);
|
||||
channel = NETConnectWithFlags((char *) host, port_no, 0);
|
||||
/* TODO handle asynchronous connection */
|
||||
}
|
||||
}
|
||||
@@ -780,13 +768,11 @@ static int AQ_Init(pAsyncQueue self)
|
||||
/* Init the controller */
|
||||
if (self->nw_ctx == NULL)
|
||||
NetWatchRegisterCallback(&self->nw_ctx,
|
||||
self->pSock->sockid,
|
||||
MyCallback,
|
||||
self);
|
||||
self->pSock->sockid, MyCallback, self);
|
||||
return 1;
|
||||
}
|
||||
|
||||
static void AQ_Kill(void* pData)
|
||||
static void AQ_Kill(void *pData)
|
||||
{
|
||||
int i;
|
||||
pAsyncQueue self = (pAsyncQueue) pData;
|
||||
@@ -814,42 +800,48 @@ static void AQ_Kill(void* pData)
|
||||
*
|
||||
* MakeAsyncQueue queueName protocolName hostName portname
|
||||
*/
|
||||
int AsyncQueueFactory(SConnection *pCon, SicsInterp *pSics,
|
||||
void *pData, int argc, char *argv[])
|
||||
int AsyncQueueFactory(SConnection * pCon, SicsInterp * pSics,
|
||||
void *pData, int argc, char *argv[])
|
||||
{
|
||||
pAsyncQueue pNew = NULL;
|
||||
mkChannel* channel = NULL;
|
||||
mkChannel *channel = NULL;
|
||||
pAsyncProtocol pPro = NULL;
|
||||
int port_no;
|
||||
int iRet = 0;
|
||||
|
||||
if (argc < 5) {
|
||||
SCWrite(pCon,"ERROR: insufficient arguments to AsyncQueueFactory", eError);
|
||||
SCWrite(pCon, "ERROR: insufficient arguments to AsyncQueueFactory",
|
||||
eError);
|
||||
return 0;
|
||||
}
|
||||
|
||||
/* try to find an existing queue with this name */
|
||||
pNew = (pAsyncQueue) FindCommandData(pServ->pSics, argv[1], "AsyncQueue");
|
||||
pNew =
|
||||
(pAsyncQueue) FindCommandData(pServ->pSics, argv[1], "AsyncQueue");
|
||||
if (pNew != NULL) {
|
||||
char line[132];
|
||||
snprintf(line, 132, "WARNING: AsyncQueue '%s' already exists", argv[1]);
|
||||
snprintf(line, 132, "WARNING: AsyncQueue '%s' already exists",
|
||||
argv[1]);
|
||||
SCWrite(pCon, line, eError);
|
||||
SCSendOK(pCon);
|
||||
return 1;
|
||||
}
|
||||
|
||||
/* try to find an existing protocol with this name */
|
||||
pPro = (pAsyncProtocol) FindCommandData(pServ->pSics, argv[2], "AsyncProtocol");
|
||||
pPro =
|
||||
(pAsyncProtocol) FindCommandData(pServ->pSics, argv[2],
|
||||
"AsyncProtocol");
|
||||
if (pPro == NULL) {
|
||||
char line[132];
|
||||
snprintf(line, 132, "WARNING: AsyncQueue protocol '%s' not found", argv[2]);
|
||||
snprintf(line, 132, "WARNING: AsyncQueue protocol '%s' not found",
|
||||
argv[2]);
|
||||
SCWrite(pCon, line, eError);
|
||||
return 0;
|
||||
}
|
||||
|
||||
port_no = atoi(argv[4]);
|
||||
if (port_no == 0) {
|
||||
struct servent *sp=NULL;
|
||||
struct servent *sp = NULL;
|
||||
sp = getservbyname(argv[4], NULL);
|
||||
if (sp)
|
||||
port_no = ntohs(sp->s_port);
|
||||
@@ -861,17 +853,18 @@ int AsyncQueueFactory(SConnection *pCon, SicsInterp *pSics,
|
||||
/* look for queue with same address */
|
||||
for (i = 0; i < queue_index; ++i)
|
||||
if (queue_array[i]->pSock->adresse.sin_port == sa.sin_port
|
||||
&& queue_array[i]->pSock->adresse.sin_addr.s_addr == sa.sin_addr.s_addr) {
|
||||
&& queue_array[i]->pSock->adresse.sin_addr.s_addr ==
|
||||
sa.sin_addr.s_addr) {
|
||||
char line[132];
|
||||
snprintf(line, 132, "WARNING: AsyncQueue '%s' has same address as %s",
|
||||
argv[1],
|
||||
queue_array[i]->queue_name);
|
||||
snprintf(line, 132,
|
||||
"WARNING: AsyncQueue '%s' has same address as %s",
|
||||
argv[1], queue_array[i]->queue_name);
|
||||
SCWrite(pCon, line, eError);
|
||||
}
|
||||
}
|
||||
}
|
||||
/* TODO: implement asynchronous connection */
|
||||
channel = NETConnectWithFlags(argv[3], port_no, 0);
|
||||
}
|
||||
}
|
||||
|
||||
if (channel == NULL) {
|
||||
char line[132];
|
||||
@@ -896,13 +889,12 @@ int AsyncQueueFactory(SConnection *pCon, SicsInterp *pSics,
|
||||
queue_array[queue_index++] = pNew;
|
||||
|
||||
AQ_Init(pNew);
|
||||
|
||||
|
||||
/*
|
||||
create the command
|
||||
*/
|
||||
create the command
|
||||
*/
|
||||
iRet = AddCommand(pSics, argv[1], AsyncQueueAction, AQ_Kill, pNew);
|
||||
if(!iRet)
|
||||
{
|
||||
if (!iRet) {
|
||||
char line[132];
|
||||
snprintf(line, 123, "ERROR: add command %s failed", argv[1]);
|
||||
SCWrite(pCon, line, eError);
|
||||
@@ -920,7 +912,8 @@ int AsyncQueueFactory(SConnection *pCon, SicsInterp *pSics,
|
||||
* \param handle the handle to the AsyncQueue object
|
||||
* \return 0 for FAILURE, 1 for SUCCESS
|
||||
*/
|
||||
int AsyncUnitCreateHost(const char* host, const char* port, pAsyncUnit* handle)
|
||||
int AsyncUnitCreateHost(const char *host, const char *port,
|
||||
pAsyncUnit * handle)
|
||||
{
|
||||
int status;
|
||||
pAsyncQueue self = NULL;
|
||||
@@ -948,7 +941,8 @@ int AsyncUnitCreateHost(const char* host, const char* port, pAsyncUnit* handle)
|
||||
return 1;
|
||||
}
|
||||
|
||||
int AsyncUnitCreate(const char* host, pAsyncUnit* handle) {
|
||||
int AsyncUnitCreate(const char *host, pAsyncUnit * handle)
|
||||
{
|
||||
return AsyncUnitCreateHost(host, NULL, handle);
|
||||
}
|
||||
|
||||
@@ -957,7 +951,7 @@ int AsyncUnitDestroy(pAsyncUnit unit)
|
||||
assert(unit);
|
||||
assert(unit->queue);
|
||||
pAsyncQueue self = unit->queue;
|
||||
pAsyncUnit* pNxt = &self->units;
|
||||
pAsyncUnit *pNxt = &self->units;
|
||||
while (*pNxt) {
|
||||
if (*pNxt == unit) {
|
||||
*pNxt = (*pNxt)->next;
|
||||
@@ -973,15 +967,15 @@ int AsyncUnitDestroy(pAsyncUnit unit)
|
||||
return 1;
|
||||
}
|
||||
|
||||
pAsyncUnit AsyncUnitFromQueue(pAsyncQueue queue){
|
||||
pAsyncUnit result = NULL;
|
||||
|
||||
result = malloc(sizeof(AsyncUnit));
|
||||
if(result == NULL){
|
||||
return NULL;
|
||||
}
|
||||
memset(result,0,sizeof(AsyncUnit));
|
||||
result->queue = queue;
|
||||
return result;
|
||||
}
|
||||
pAsyncUnit AsyncUnitFromQueue(pAsyncQueue queue)
|
||||
{
|
||||
pAsyncUnit result = NULL;
|
||||
|
||||
result = malloc(sizeof(AsyncUnit));
|
||||
if (result == NULL) {
|
||||
return NULL;
|
||||
}
|
||||
memset(result, 0, sizeof(AsyncUnit));
|
||||
result->queue = queue;
|
||||
return result;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user