linse/sics
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
22b0e8ec83f1c4aeeb3b88884e933c46d6bf48d8
sics/asyncqueue.c
koennecke 22b0e8ec83 *** empty log message ***
2007-11-27 13:36:15 +00:00

988 lines
24 KiB
C
Raw Blame History

/*
* A S Y N C Q U E U E
*
* This module manages AsyncQueue communications.
*
* The AsyncQueue is an asynchronous queue between drivers and the device. It
* supports multiple logical units on a single device controller that share a
* single command channel.
*
* Douglas Clowes, February 2007
*
*/
#include <sys/time.h>
#include <arpa/inet.h>
#include <netinet/tcp.h>
#include <netdb.h>
#include <sics.h>
#include <rs232controller.h>
#include "network.h"
#include "asyncqueue.h"
#include "nwatch.h"
typedef struct __async_command AQ_Cmd, *pAQ_Cmd;
struct __async_command {
pAQ_Cmd next;
pAsyncTxn tran;
pAsyncUnit unit;
int timeout;
int retries;
int active;
};
struct __AsyncUnit {
pAsyncUnit next;
pAsyncQueue queue;
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;
};
static pAsyncQueue queue_array[FD_SETSIZE];
static int queue_index = 0;
/* ---------------------------- Local ------------------------------------
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 */
(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 {
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.
*/
sockaddrPtr->sin_addr.s_addr = addr.s_addr;
return 1;
}
static void AQ_Notify(pAsyncQueue self, int event)
{
pAsyncUnit unit;
for (unit = self->units; unit; unit = unit->next)
if (unit->notify_func != NULL)
unit->notify_func(unit->notify_cntx, event);
}
static int AQ_Reconnect(pAsyncQueue self)
{
int iRet;
int sock;
int flag = 1;
char line[132];
iRet = NETReconnect(self->pSock);
if (iRet <= 0) {
snprintf(line, 132, "Disconnect on AsyncQueue '%s'", self->queue_name);
SICSLogWrite(line, eStatus);
AQ_Notify(self, AQU_DISCONNECT);
return iRet;
}
snprintf(line, 132, "Reconnect on AsyncQueue '%s'", self->queue_name);
SICSLogWrite(line, eStatus);
AQ_Notify(self, AQU_RECONNECT);
return 1;
}
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;
if (myCmd == NULL)
return OKOK;
/*
* Remove any old command timeout timer
*/
if (self->nw_tmr)
NetWatchRemoveTimer(self->nw_tmr);
/*
* Implement the inter-command delay
*/
if (self->iDelay) {
struct timeval now, when;
gettimeofday(&now, NULL);
if (self->tvLastCmd.tv_sec == 0)
self->tvLastCmd = now;
when.tv_sec = self->tvLastCmd.tv_sec;
when.tv_usec = self->tvLastCmd.tv_usec + 1000 * self->iDelay;
if (when.tv_usec >= 1000000) {
when.tv_sec += when.tv_usec / 1000000;
when.tv_usec %= 1000000;
}
if (when.tv_sec > now.tv_sec ||
(when.tv_sec == now.tv_sec && when.tv_usec > now.tv_usec)) {
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);
return OKOK;
}
}
/*
* Discard any input before sending command
*/
while (NETAvailable(sock, 0)) {
/* TODO: handle unsolicited input */
char reply[1];
int iRet;
iRet = NETRead(sock, reply, 1, 0);
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);
}
PopCommand(self);
return 0;
}
}
}
/*
* Add a new command timeout timer
*/
if (myCmd->timeout > 0)
NetWatchRegisterTimer(&self->nw_tmr, myCmd->timeout,
CommandTimeout, self);
else
NetWatchRegisterTimer(&self->nw_tmr, 30000,
CommandTimeout, self);
myCmd->active = 1;
return self->protocol->sendCommand(self->protocol, myCmd->tran);
}
static int QueCommandHead(pAsyncQueue self, pAQ_Cmd cmd)
{
cmd->next = NULL;
/*
* If the command queue is empty, start transmission
*/
if (self->command_head == NULL) {
self->command_head = self->command_tail = cmd;
StartCommand(self);
return 1;
}
if (self->command_head->active) {
cmd->next = self->command_head->next;
self->command_head->next = cmd;
}
else {
cmd->next = self->command_head;
self->command_head = cmd;
}
if (cmd->next == NULL)
self->command_tail = cmd;
return 1;
}
static int QueCommand(pAsyncQueue self, pAQ_Cmd cmd)
{
cmd->next = NULL;
/*
* If the command queue is empty, start transmission
*/
if (self->command_head == NULL) {
self->command_head = self->command_tail = cmd;
StartCommand(self);
return 1;
}
self->command_tail->next = cmd;
self->command_tail = cmd;
return 1;
}
static int PopCommand(pAsyncQueue self)
{
pAQ_Cmd myCmd = self->command_head;
if (self->nw_tmr)
NetWatchRemoveTimer(self->nw_tmr);
self->nw_tmr = 0;
gettimeofday(&self->tvLastCmd, NULL);
/*
* If this is not the last in queue, start transmission
*/
if (myCmd->next) {
pAQ_Cmd pNew = myCmd->next;
self->command_head = pNew;
StartCommand(self);
}
else
self->command_head = self->command_tail = NULL;
free(myCmd->tran->out_buf);
free(myCmd->tran->inp_buf);
free(myCmd->tran);
free(myCmd);
return 1;
}
static int CommandTimeout(void* cntx, int mode)
{
pAsyncQueue self = (pAsyncQueue) cntx;
pAQ_Cmd myCmd = self->command_head;
self->nw_tmr = 0;
if (myCmd->retries > 0) {
--myCmd->retries;
StartCommand(self);
}
else {
int iRet;
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 */
else if (iRet == AQU_RECONNECT)
AQ_Reconnect(self);
}
return 1;
}
static int DelayedStart(void* cntx, int mode)
{
pAsyncQueue self = (pAsyncQueue) cntx;
self->nw_tmr = 0;
StartCommand(self);
return 1;
}
static int MyCallback(void* context, int mode)
{
pAsyncQueue self = (pAsyncQueue) context;
if (mode & nwatch_read) {
int iRet;
char reply[1];
iRet = NETRead(self->pSock, reply, 1, 0);
/* printf(" iRet, char = %d, %d\n", iRet, (int)reply[0]); */
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);
}
return iRet;
}
/* restart the command */
StartCommand(self);
return 1;
}
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]);
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 {
/* TODO: handle unsolicited input */
}
}
}
return 1;
}
int AsyncUnitEnqueueHead(pAsyncUnit unit, pAsyncTxn context)
{
pAQ_Cmd myCmd = NULL;
assert(unit && unit->queue && unit->queue->protocol);
myCmd = (pAQ_Cmd) malloc(sizeof(AQ_Cmd));
if (myCmd == NULL) {
SICSLogWrite("ERROR: Out of memory in AsyncUnitEnqueHead", eError);
return 0;
}
memset(myCmd, 0, sizeof(AQ_Cmd));
myCmd->tran = context;
myCmd->unit = unit;
myCmd->timeout = unit->queue->timeout;
myCmd->retries = unit->queue->retries;
myCmd->active = 0;
return QueCommandHead(unit->queue, myCmd);
}
int AsyncUnitEnqueueTxn(pAsyncUnit unit, pAsyncTxn pTxn)
{
pAQ_Cmd myCmd = NULL;
assert(unit && unit->queue && unit->queue->protocol);
myCmd = (pAQ_Cmd) malloc(sizeof(AQ_Cmd));
if (myCmd == NULL) {
SICSLogWrite("ERROR: Out of memory in AsyncUnitEnqueueTxn", eError);
return 0;
}
memset(myCmd, 0, sizeof(AQ_Cmd));
myCmd->tran = pTxn;
myCmd->unit = unit;
myCmd->timeout = unit->queue->timeout;
myCmd->retries = unit->queue->retries;
myCmd->active = 0;
return QueCommand(unit->queue, myCmd);
}
pAsyncTxn AsyncUnitPrepareTxn(pAsyncUnit unit,
const char* command, int cmd_len,
AsyncTxnHandler callback, void* context,
int rsp_len)
{
pAsyncTxn myTxn = NULL;
assert(unit);
myTxn = (pAsyncTxn) malloc(sizeof(AsyncTxn));
if (myTxn == NULL) {
SICSLogWrite("ERROR: Out of memory in AsyncUnitPrepareTxn", eError);
return 0;
}
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);
if (myTxn->out_buf == NULL) {
SICSLogWrite("ERROR: Out of memory in AsyncUnitPrepareTxn", eError);
free(myTxn);
return 0;
}
memcpy(myTxn->out_buf, command, cmd_len);
myTxn->out_len = cmd_len;
if (myTxn->out_len < 2 ||
myTxn->out_buf[myTxn->out_len - 1] != 0x0A ||
myTxn->out_buf[myTxn->out_len - 2] != 0x0D) {
myTxn->out_buf[myTxn->out_len++] = 0x0D;
myTxn->out_buf[myTxn->out_len++] = 0x0A;
}
myTxn->out_buf[myTxn->out_len] = '\0';
}
if (rsp_len == 0)
myTxn->inp_buf = NULL;
else {
if(myTxn->inp_buf != NULL){
free(myTxn->inp_buf);
}
myTxn->inp_buf = malloc(rsp_len + 1);
if (myTxn->inp_buf == NULL) {
SICSLogWrite("ERROR: Out of memory in AsyncUnitPrepareTxn", eError);
free(myTxn->out_buf);
free(myTxn);
return 0;
}
memset(myTxn->inp_buf, 0, rsp_len + 1);
}
myTxn->inp_len = rsp_len;
myTxn->unit = unit;
myTxn->handleResponse = callback;
myTxn->cntx = context;
return myTxn;
}
int AsyncUnitSendTxn(pAsyncUnit unit,
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);
if (myTxn == NULL)
return -1;
return AsyncUnitEnqueueTxn(unit, myTxn);
}
typedef struct txn_s {
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;
int resp_len = pCmd->inp_idx;
pTXN self = (pTXN) pCmd->cntx;
if (pCmd->txn_status == ATX_TIMEOUT) {
memcpy(self->transReply, resp, resp_len);
self->transReply[resp_len] = '\0';
self->transReply[0] = '\0';
self->transWait = -1;
}
else {
memcpy(self->transReply, resp, resp_len);
self->transReply[resp_len] = '\0';
self->transWait = 0;
}
return 0;
}
int AsyncUnitTransact(pAsyncUnit unit,
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);
while (txn.transWait == 1)
TaskYield(pServ->pTasker);
if (txn.transWait < 0)
return txn.transWait;
return 1;
}
int AsyncUnitWrite(pAsyncUnit unit, void* buffer, int buflen)
{
int iRet;
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 */
iRet = AQ_Reconnect(unit->queue);
if (iRet == 0)
return 0;
}
}
return 1;
}
void AsyncUnitSetNotify(pAsyncUnit unit, void* context, AQU_Notify notify)
{
assert(unit);
unit->notify_func = notify;
unit->notify_cntx = context;
}
int AsyncUnitGetDelay(pAsyncUnit unit)
{
assert(unit);
return unit->queue->iDelay;
}
void AsyncUnitSetDelay(pAsyncUnit unit, int iDelay)
{
assert(unit);
unit->queue->iDelay = iDelay;
}
int AsyncUnitGetTimeout(pAsyncUnit unit)
{
assert(unit);
return unit->queue->timeout;
}
void AsyncUnitSetTimeout(pAsyncUnit unit, int timeout)
{
assert(unit);
unit->queue->timeout = timeout;
}
int AsyncUnitGetRetries(pAsyncUnit unit)
{
assert(unit);
return unit->queue->retries;
}
void AsyncUnitSetRetries(pAsyncUnit unit, int retries)
{
assert(unit);
unit->queue->retries = retries;
}
pAsyncProtocol AsyncUnitGetProtocol(pAsyncUnit unit)
{
return unit->queue->protocol;
}
void AsyncUnitSetProtocol(pAsyncUnit unit, pAsyncProtocol protocol)
{
unit->queue->protocol = protocol;
}
mkChannel* AsyncUnitGetSocket(pAsyncUnit unit)
{
assert(unit);
assert(unit->queue);
return unit->queue->pSock;
}
int AsyncUnitReconnect(pAsyncUnit unit)
{
int iRet;
assert(unit);
assert(unit->queue);
iRet = AQ_Reconnect(unit->queue);
/* TODO: handle in-progress */
return iRet;
}
int AsyncQueueAction(SConnection *pCon, SicsInterp *pSics,
void *pData, int argc, char *argv[])
{
char line[132];
pAsyncQueue self = (pAsyncQueue) pData;
if (argc > 1) {
if (strcasecmp("send", argv[1]) == 0) {
AsyncUnit myUnit;
char cmd[10240];
char rsp[10240];
int idx = 0;
int i, j;
cmd[0] = '\0';
for (i = 2; i < argc; ++i) {
j = snprintf(&cmd[idx], 10240 - idx, "%s%s",
(i > 2) ? " " : "",
argv[i]);
if (j < 0)
break;
idx += j;
}
memset(&myUnit, 0, sizeof(AsyncUnit));
myUnit.queue = self;
AsyncUnitTransact(&myUnit, cmd, idx, rsp, 10240);
SCWrite(pCon, rsp, eValue);
return 1;
}
if (strcasecmp(argv[1], "reconnect") == 0) {
AQ_Reconnect(self);
return OKOK;
}
if (strcasecmp(argv[1], "delay") == 0) {
if (argc > 2) {
int delay;
int iRet;
iRet = sscanf(argv[2], "%d", &delay);
if (iRet != 1) {
snprintf(line, 132, "Invalid argument: %s", argv[2]);
SCWrite(pCon, line, eError);
return 0;
}
else {
if (delay < 0 || delay > 30000) {
snprintf(line, 132, "Value out of range: %d", delay);
SCWrite(pCon, line, eError);
return 0;
}
self->iDelay = delay;
return OKOK;
}
}
else {
snprintf(line, 132, "%s.delay = %d", argv[0], self->iDelay);
SCWrite(pCon, line, eStatus);
return OKOK;
}
return OKOK;
}
if (strcasecmp(argv[1], "timeout") == 0) {
if (argc > 2) {
int timeout;
int iRet;
iRet = sscanf(argv[2], "%d", &timeout);
if (iRet != 1) {
snprintf(line, 132, "Invalid argument: %s", argv[2]);
SCWrite(pCon, line, eError);
return 0;
}
else {
if (timeout < 0 || timeout > 30000) {
snprintf(line, 132, "Value out of range: %d", timeout);
SCWrite(pCon, line, eError);
return 0;
}
self->timeout = timeout;
return OKOK;
}
}
else {
snprintf(line, 132, "%s.timeout = %d", argv[0], self->timeout);
SCWrite(pCon, line, eStatus);
return OKOK;
}
return OKOK;
}
if (strcasecmp(argv[1], "retries") == 0) {
if (argc > 2) {
int retries;
int iRet;
iRet = sscanf(argv[2], "%d", &retries);
if (iRet != 1) {
snprintf(line, 132, "Invalid argument: %s", argv[2]);
SCWrite(pCon, line, eError);
return 0;
}
else {
if (retries < 0 || retries > 30000) {
snprintf(line, 132, "Value out of range: %d", retries);
SCWrite(pCon, line, eError);
return 0;
}
self->retries = retries;
return OKOK;
}
}
else {
snprintf(line, 132, "%s.retries = %d", argv[0], self->retries);
SCWrite(pCon, line, eStatus);
return OKOK;
}
return OKOK;
}
}
snprintf(line, 132, "%s does not understand %s", argv[0], argv[1]);
SCWrite(pCon, line, eError);
return 0;
}
static pAsyncQueue AQ_Create(const char* host, const char* port)
{
int i;
pAsyncQueue self = NULL;
mkChannel* channel = NULL;
if (host == NULL)
return NULL;
/* try the AsyncQueue with this name */
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;
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)) {
/* 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) {
self = queue_array[i];
break;
}
}
if (self == NULL) {
channel = NETConnectWithFlags((char *)host, port_no, 0);
/* TODO handle asynchronous connection */
}
}
}
if (self == NULL) {
if (channel == NULL)
return NULL;
self = (pAsyncQueue) malloc(sizeof(AsyncQueue));
if (self == NULL)
return NULL;
memset(self, 0, sizeof(AsyncQueue));
self->pSock = channel;
self->pDes = CreateDescriptor("AsyncQueue");
queue_array[queue_index++] = self;
}
for (i = 0; i < queue_index; ++i)
if (queue_array[i] == self) {
break;
}
if (i == queue_index)
queue_array[queue_index++] = self;
return self;
}
static int AQ_Init(pAsyncQueue self)
{
/* Init the controller */
if (self->nw_ctx == NULL)
NetWatchRegisterCallback(&self->nw_ctx,
self->pSock->sockid,
MyCallback,
self);
return 1;
}
static void AQ_Kill(void* pData)
{
int i;
pAsyncQueue self = (pAsyncQueue) pData;
for (i = 0; i < queue_index; ++i)
if (queue_array[i] == self) {
--queue_index;
if (queue_index > 0)
queue_array[i] = queue_array[queue_index];
if (self->nw_ctx)
NetWatchRemoveCallback(self->nw_ctx);
if (self->nw_tmr)
NetWatchRemoveTimer(self->nw_tmr);
if (self->queue_name)
free(self->queue_name);
NETClosePort(self->pSock);
free(self->pSock);
DeleteDescriptor(self->pDes);
free(self);
return;
}
}
/*
* \brief make a AsyncQueue from the command line
*
* MakeAsyncQueue queueName protocolName hostName portname
*/
int AsyncQueueFactory(SConnection *pCon, SicsInterp *pSics,
void *pData, int argc, char *argv[])
{
pAsyncQueue pNew = NULL;
mkChannel* channel = NULL;
pAsyncProtocol pPro = NULL;
int port_no;
int iRet = 0;
if (argc < 5) {
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");
if (pNew != NULL) {
char line[132];
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");
if (pPro == NULL) {
char line[132];
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;
sp = getservbyname(argv[4], NULL);
if (sp)
port_no = ntohs(sp->s_port);
}
if (port_no > 0) {
struct sockaddr_in sa;
if (CreateSocketAdress(&sa, argv[3], port_no)) {
int i;
/* 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) {
char line[132];
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];
snprintf(line, 132, "ERROR: AsyncQueue '%s' cannot connect", argv[1]);
SCWrite(pCon, line, eError);
return 0;
}
pNew = (pAsyncQueue) malloc(sizeof(AsyncQueue));
if (pNew == NULL) {
char line[132];
snprintf(line, 132, "ERROR: AsyncQueue '%s' memory failure", argv[1]);
SCWrite(pCon, line, eError);
return 0;
}
memset(pNew, 0, sizeof(AsyncQueue));
pNew->pDes = CreateDescriptor("AsyncQueue");
pNew->queue_name = strdup(argv[1]);
pNew->protocol = pPro;
pNew->pSock = channel;
queue_array[queue_index++] = pNew;
AQ_Init(pNew);
/*
create the command
*/
iRet = AddCommand(pSics, argv[1], AsyncQueueAction, AQ_Kill, pNew);
if(!iRet)
{
char line[132];
snprintf(line, 123, "ERROR: add command %s failed", argv[1]);
SCWrite(pCon, line, eError);
AQ_Kill(pNew);
return 0;
}
SCSendOK(pCon);
return 1;
}
/*
* \brief make a AsyncQueue from a named rs232 controller
*
* \param name the name of the SICS "RS232 Controller" object
* \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 status;
pAsyncQueue self = NULL;
pAsyncUnit unit = NULL;
*handle = NULL;
self = AQ_Create(host, port);
if (self == NULL)
return 0;
status = AQ_Init(self);
unit = (pAsyncUnit) malloc(sizeof(AsyncUnit));
if (unit == NULL) {
SICSLogWrite("ERROR: Out of memory in AsyncUnitCreateHost", eError);
*handle = NULL;
return 0;
}
memset(unit, 0, sizeof(AsyncUnit));
++self->unit_count;
unit->queue = self;
unit->next = self->units;
self->units = unit;
*handle = unit;
return 1;
}
int AsyncUnitCreate(const char* host, pAsyncUnit* handle) {
return AsyncUnitCreateHost(host, NULL, handle);
}
int AsyncUnitDestroy(pAsyncUnit unit)
{
assert(unit);
assert(unit->queue);
pAsyncQueue self = unit->queue;
pAsyncUnit* pNxt = &self->units;
while (*pNxt) {
if (*pNxt == unit) {
*pNxt = (*pNxt)->next;
break;
}
pNxt = &(*pNxt)->next;
}
--self->unit_count;
if (self->unit_count <= 0) {
AQ_Kill(self);
}
free(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;
}
Reference in New Issue View Git Blame Copy Permalink