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

Adjustments for handling binary protocols

Browse Source 
r2267 | dcl | 2008-01-15 12:17:58 +1100 (Tue, 15 Jan 2008) | 2 lines
This commit is contained in:
Douglas Clowes
2008-01-15 12:17:58 +11:00
parent 56643bb4c5
commit 1250d902e7
2 changed files with 119 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

141
asyncqueue.c
View File

@@ -15,11 +15,13 @@
#include <arpa/inet.h>
#include <netinet/tcp.h>
#include <netdb.h>
#include <ctype.h>
#include <sics.h>
#include <rs232controller.h>
#include "network.h"
#include "asyncqueue.h"
#include "nwatch.h"
#include <stdbool.h>
typedef struct __AsyncQueue AsyncQueue, *pAsyncQueue;
@@ -49,6 +51,7 @@ struct __AsyncQueue {
int iDelay; /* intercommand delay in milliseconds */
int timeout;
int retries;
bool translate; /* translate binary output with escaped chars */
struct timeval tvLastCmd; /* time of completion of last command */
int unit_count; /* number of units connected */
pAsyncUnit units; /* head of unit chain */
@@ -399,19 +402,23 @@ pAsyncTxn AsyncUnitPrepareTxn(pAsyncUnit unit,
myTxn = (pAsyncTxn) malloc(sizeof(AsyncTxn));
if (myTxn == NULL) {
SICSLogWrite("ERROR: Out of memory in AsyncUnitPrepareTxn", eError);
return 0;
return NULL;
}
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);
if (iRet == 0) {
free(myTxn);
return NULL;
}
}
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;
return NULL;
}
memcpy(myTxn->out_buf, command, cmd_len);
myTxn->out_len = cmd_len;
@@ -431,7 +438,7 @@ pAsyncTxn AsyncUnitPrepareTxn(pAsyncUnit unit,
SICSLogWrite("ERROR: Out of memory in AsyncUnitPrepareTxn", eError);
free(myTxn->out_buf);
free(myTxn);
return 0;
return NULL;
}
memset(myTxn->inp_buf, 0, rsp_len + 1);
}
@@ -459,6 +466,7 @@ int AsyncUnitSendTxn(pAsyncUnit unit,
typedef struct txn_s {
char* transReply;
int transWait;
int respLen;
} TXN, *pTXN;
/**
@@ -469,33 +477,37 @@ static int TransCallback(pAsyncTxn pCmd) {
int resp_len = pCmd->inp_idx;
pTXN self = (pTXN) pCmd->cntx;
if (pCmd->txn_status == ATX_TIMEOUT) {
self->respLen = resp_len;
if (resp_len > 0) {
memcpy(self->transReply, resp, resp_len);
self->transReply[resp_len] = '\0';
}
else
self->transReply[0] = '\0';
if (pCmd->txn_status == ATX_TIMEOUT)
self->transWait = -1;
}
else {
memcpy(self->transReply, resp, resp_len);
self->transReply[resp_len] = '\0';
else
self->transWait = 0;
}
return 0;
}
int AsyncUnitTransact(pAsyncUnit unit,
const char* command, int cmd_len,
char* response, int rsp_len)
char* response, int *rsp_len)
{
TXN txn;
assert(unit);
txn.transReply = response;
txn.respLen = *rsp_len;
txn.transWait = 1;
AsyncUnitSendTxn(unit,
command, cmd_len,
TransCallback, &txn, rsp_len);
TransCallback, &txn, *rsp_len);
while (txn.transWait == 1)
TaskYield(pServ->pTasker);
*rsp_len = txn.respLen;
if (txn.transWait < 0)
return txn.transWait;
return 1;
@@ -601,20 +613,72 @@ int AsyncQueueAction(SConnection *pCon, SicsInterp *pSics,
char cmd[10240];
char rsp[10240];
int idx = 0;
int i, j;
int i, j, len;
cmd[0] = '\0';
for (i = 2; i < argc; ++i) {
j = snprintf(&cmd[idx], 10240 - idx, "%s%s",
(i > 2) ? " " : "",
argv[i]);
if (j < 0)
if (idx >= 10240)
break;
idx += j;
if (i > 2)
cmd[idx++] = ' ';
len = strlen(argv[i]);
for (j = 0; j < len; ++j) {
if (idx >= 10240)
break;
if (argv[i][j] == '\\') {
++j;
if (argv[i][j] == '\\')
cmd[idx++] = '\\';
else if (argv[i][j] == 'r')
cmd[idx++] = '\r';
else if (argv[i][j] == 'n')
cmd[idx++] = '\n';
else if (isdigit(argv[i][j])) {
char* nptr = &argv[i][j];
char* endptr;
long k = strtol(nptr, &endptr, 0);
cmd[idx++] = k;
j += (endptr - nptr);
--j; /* prepare for loop increment */
} else
cmd[idx++] = argv[i][j];
} else
cmd[idx++] = argv[i][j];
}
}
cmd[idx] = '\0';
memset(&myUnit, 0, sizeof(AsyncUnit));
myUnit.queue = self;
AsyncUnitTransact(&myUnit, cmd, idx, rsp, 10240);
SCWrite(pCon, rsp, eValue);
len = 10240;
(void) AsyncUnitTransact(&myUnit, cmd, idx, rsp, &len);
/* escape control characters in response */
j = 0;
for (i = 0; i < len; ++i) {
if (j >= 10230)
break;
if (self->translate) {
if (rsp[i] < 32 || rsp[i] >= 127) {
if (rsp[i] == '\r') {
cmd[j++] = '\r';
}
else if (rsp[i] == '\n') {
cmd[j++] = '\n';
}
else {
j += snprintf(&cmd[j], 6, "\\0x%02x", rsp[i]);
}
}
else if (rsp[i] == '\\') {
cmd[j++] = '\\';
cmd[j++] = '\\';
}
else
cmd[j++] = rsp[i];
}
else
cmd[j++] = rsp[i];
}
cmd[j++] = '\0';
SCWrite(pCon, cmd, eValue);
return 1;
}
if (strcasecmp(argv[1], "reconnect") == 0) {
@@ -632,7 +696,7 @@ int AsyncQueueAction(SConnection *pCon, SicsInterp *pSics,
return 0;
}
else {
if (delay < 0 || delay > 30000) {
if (delay < 0 || delay > 300000) {
snprintf(line, 132, "Value out of range: %d", delay);
SCWrite(pCon, line, eError);
return 0;
@@ -702,6 +766,37 @@ int AsyncQueueAction(SConnection *pCon, SicsInterp *pSics,
}
return OKOK;
}
if (strcasecmp(argv[1], "translate") == 0) {
if (argc > 2) {
int value;
int iRet;
iRet = sscanf(argv[2], "%d", &value);
if (iRet != 1) {
snprintf(line, 132, "Invalid argument: %s", argv[2]);
SCWrite(pCon, line, eError);
return 0;
}
else {
if (value == 0) {
self->translate = false;
return OKOK;
}
else if (value == 1) {
self->timeout = true;
return OKOK;
}
snprintf(line, 132, "Invalid argument: %s", argv[2]);
SCWrite(pCon, line, eError);
return 0;
}
}
else {
snprintf(line, 132, "%s.timeout = %d", argv[0], self->timeout);
SCWrite(pCon, line, eStatus);
return OKOK;
}
return OKOK;
}
}
snprintf(line, 132, "%s does not understand %s", argv[0], argv[1]);
SCWrite(pCon, line, eError);
@@ -718,7 +813,7 @@ static pAsyncQueue AQ_Create(const char* host, const char* port)
return NULL;
/* try the AsyncQueue with this name */
self = (pAsyncQueue) FindCommandData(pServ->pSics, host, "AsyncQueue");
self = (pAsyncQueue) FindCommandData(pServ->pSics, (char*) host, "AsyncQueue");
/* try host and port */
if (self == NULL && port) {
@@ -731,7 +826,7 @@ static pAsyncQueue AQ_Create(const char* host, const char* port)
}
if (port_no > 0) {
struct sockaddr_in sa;
if (CreateSocketAdress(&sa, 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
@@ -741,7 +836,7 @@ static pAsyncQueue AQ_Create(const char* host, const char* port)
}
}
if (self == NULL) {
channel = NETConnectWithFlags(host, port_no, 0);
channel = NETConnectWithFlags((char *)host, port_no, 0);
/* TODO handle asynchronous connection */
}
}