sics/asyncprotocol.c

#include <sics.h>
#include <asyncprotocol.h>
#include <asyncqueue.h>

int defaultSendCommand(pAsyncProtocol p, pAsyncTxn txn) {
  int i, iRet;
  int state;
  const char *term = "\r\n";
  if (p->sendTerminator)
    term = p->sendTerminator;
  state = 0;
  for (i = 0; i < txn->out_len; ++i) {
    if (txn->out_buf[i] == 0x00) { /* end of transmission */
      break;
    }
    else if (txn->out_buf[i] == term[state]) {
      ++state;
      continue;
    }
    state = 0;
  }
  txn->txn_state = 0;
  iRet = AsyncUnitWrite(txn->unit, txn->out_buf, txn->out_len);
  if (iRet <= 0)
    return iRet;
  if (term[state] != 0)
    iRet = AsyncUnitWrite(txn->unit, (void *)term, strlen(term));
  return iRet;
}

int defaultHandleInput(pAsyncProtocol p, pAsyncTxn txn, int ch) {
  const char *term = "\r\n";
  if (txn->txn_state == 0) {
    int i;
    for (i = 0; i < 10; ++i)
      if (p->replyTerminator[i] && ch == p->replyTerminator[i][0]) {
        txn->txn_state = i << 16;
        break;
      }
  }
  term = p->replyTerminator[txn->txn_state >> 16];
  if (ch == term[txn->txn_state & 0xffff])
    ++txn->txn_state;
  else
    txn->txn_state = 0;

  if (txn->inp_idx < txn->inp_len)
    txn->inp_buf[txn->inp_idx++] = ch;
  if (term[txn->txn_state & 0xffff] == 0) {
    if (txn->inp_idx < txn->inp_len)
      txn->inp_buf[txn->inp_idx] = '\0';
    return AQU_POP_CMD;
  }
  return 1;
}

int defaultHandleEvent(pAsyncProtocol p, pAsyncTxn txn, int event) {
  /* TODO: what could or should we do to handle the event */
  return AQU_POP_CMD;
}

int defaultPrepareTxn(pAsyncProtocol p, pAsyncTxn txn, const char* cmd, int cmd_len, int rsp_len) {
  int i;
  int state;
  const char *term = "\r\n";
  if (p->sendTerminator)
    term = p->sendTerminator;
  state = 0;
  for (i = 0; i < cmd_len; ++i) {
    if (cmd[i] == term[state]) {
      ++state;
      continue;
    }
    state = 0;
  }
  if (term[state] == 0) {
    /* outgoing command is correctly terminated */
    txn->out_buf = malloc(cmd_len + 1);
    if (txn->out_buf == NULL) {
      SICSLogWrite("Out of memory in AsyncProtocol::defaultPrepareTxn", eError);
      return 0;
    }
    memcpy(txn->out_buf, cmd, cmd_len + 1);
  }
  else {
    /* outgoing command is NOT correctly terminated */
    int tlen = strlen(term);
    txn->out_buf = malloc(cmd_len + tlen + 1);
    if (txn->out_buf == NULL) {
      SICSLogWrite("Out of memory in AsyncProtocol::defaultPrepareTxn", eError);
      return 0;
    }
    memcpy(txn->out_buf, cmd, cmd_len);
    memcpy(txn->out_buf + cmd_len, term, tlen + 1);
    cmd_len += tlen;
  }
  txn->out_len = cmd_len;
  txn->out_idx = 0;
  if(txn->inp_buf != NULL){
    free(txn->inp_buf);
  }
  txn->inp_len = rsp_len;
  txn->inp_idx = 0;
  txn->txn_state = 0;
  txn->txn_status = 0;
  return 1;
}

static const char* hex = "0123456789ABCDEF";

/*--------------------------------------------------------------------*/
static void encodeTerminator(char *result, char *terminator)
{
  if (terminator)
    while (*terminator) {
      if (*terminator <= 32 || *terminator >= 127) {
        *result++ = '0';
        *result++ = 'x';
        *result++ = hex[(*terminator >> 4) &0xF];
        *result++ = hex[(*terminator) &0xF];
        ++terminator;
      }
      else {
        *result++ = *terminator++;
      }
    }
  *result = '\0';
  return;
}

static int fromHex(const char* code) {
  int icode = -1;
  int result = -1;
  if (code[0] == '0' && (code[1] == 'x' || code[1] == 'X')) {
    if (code[2] >= '0' && code[2] <= '9')
      icode = (code[2] - '0');
    else if (code[2] >= 'a' && code[2] <= 'f')
      icode = 10 + (code[2] - 'a');
    else if (code[2] >= 'A' && code[2] <= 'F')
      icode = 10 + (code[2] - 'A');
    if (icode < 0)
      return -1;
    result = icode << 4;
    icode = -1;
    if (code[3] >= '0' && code[3] <= '9')
      icode = (code[3] - '0');
    else if (code[3] >= 'a' && code[3] <= 'f')
      icode = 10 + (code[3] - 'a');
    else if (code[3] >= 'A' && code[3] <= 'F')
      icode = 10 + (code[3] - 'A');
    if (icode < 0)
      return -1;
    result |= icode;
    return result;
  }
  return -1;
}
/*--------------------------------------------------------------------*/
static char *decodeTerminator(char *code)
{
  int count = 0, icode;
  char *pResult;
  char* pCh;
  char* pQt = NULL; /* pointer to quote character if found */

  if (code == NULL)
    return NULL;
  count = strlen(code);
  pResult = (char *) malloc(count + 1);
  if (!pResult) {
    SICSLogWrite("Out of memory in AsyncProtocol::decodeTerminator", eError);
    return NULL;
  }
  memset(pResult, 0, count + 1);

  pCh = pResult;
  if (*code == '\'' || *code == '"') /* check for leading quote */
    pQt = code++;

  while (*code) {
    if (pQt && *code == *pQt) /* check for trailing quote */
      break;

    if (code[0] == '\\' && code[1] == 'r') { /* CR */
      *pCh++ = '\r';
      code += 2;
    }
    else if (code[0] == '\\' && code[1] == 'n') { /* LF */
      *pCh++ = '\n';
      code += 2;
    }
    else if ((icode = fromHex(code)) >= 0) { /* Hex: 0xFF */
      *pCh++ = icode;
      code += 4;
    }
    else /* literal */
      *pCh++ = *code++;
  }
  *pCh = '\0';

  return pResult;
}
int AsyncProtocolNoAction(SConnection *pCon, SicsInterp *pSics,
                        void *pData, int argc, char *argv[])
{
  char line[132];
  snprintf(line, 132, "%s does not understand %s", argv[0], argv[1]);
  SCWrite(pCon, line, eError);
  return 0;
}

int AsyncProtocolAction(SConnection *pCon, SicsInterp *pSics,
                        void *pData, int argc, char *argv[])
{
  pAsyncProtocol self = (pAsyncProtocol) pData;
  if (argc > 1) {
    /* handle genecic parameters like terminators */
    if (strcasecmp(argv[1], "sendterminator") == 0) {
      if (argc > 2) {
        char* pPtr = decodeTerminator(argv[2]);
        if (pPtr) {
          if (self->sendTerminator)
            free(self->sendTerminator);
          self->sendTerminator = pPtr;
        }
        SCSendOK(pCon);
      }
      else
      {
        char term[132];
        char line[1024];
        encodeTerminator(term, self->sendTerminator);
        sprintf(line, "%s.sendTerminator = \"%s\"", argv[0], term);
        SCWrite(pCon, line, eValue);
      }
      return 1;
    }
    else if (strcasecmp(argv[1], "replyterminator") == 0) {
      if (argc > 2) {
        int i;
        for (i = 0; i < 10; ++i)
          if (self->replyTerminator[i]) {
            free(self->replyTerminator[i]);
            self->replyTerminator[i] = NULL;
          }
        for (i = 0; i < 10 && i < argc - 2; ++i) {
          char* pPtr = decodeTerminator(argv[i + 2]);
          if (pPtr) {
            self->replyTerminator[i] = pPtr;
          }
        }
        SCSendOK(pCon);
      }
      else
      {
        int i;
        char term[132];
        char line[1024];
        term[0] = '\0';
        sprintf(line, "%s.replyTerminator =", argv[0]);
        for (i = 0; i < 10; ++i) {
          if (self->replyTerminator[i] == NULL)
            break;
          term[0] = ' ';
          term[1] = '"';
          encodeTerminator(&term[2], self->replyTerminator[i]);
          strcat(term, "\"");
          strcat(line, term);
        }
        SCWrite(pCon, line, eValue);
      }
      return 1;
    }
  }
  else if (strcasecmp(argv[1], "list") == 0) {
    int ac = 2;
    char* av[3] = { argv[0], 0, 0 };
    av[1] = "sendterminator";
    AsyncProtocolAction(pCon, pSics, pData, ac, av);
    av[1] = "replyterminator";
    AsyncProtocolAction(pCon, pSics, pData, ac, av);
    return 1;
  }
  /* handle any other actions here */
  return AsyncProtocolNoAction(pCon, pSics, pData, argc,argv);
}

void defaultKillPrivate(pAsyncProtocol p) {
  if (p->privateData) {
    /* TODO: should we do anything? */
    free(p->privateData);
  }
}

void AsyncProtocolKill(void *pData) {
  pAsyncProtocol self = (pAsyncProtocol) pData;
  int i;
  if(self->pDes)
    DeleteDescriptor(self->pDes);
  if(self->sendTerminator != NULL)
    free(self->sendTerminator);
  for (i = 0; i < 10; ++i)
    if(self->replyTerminator[i] != NULL)
      free(self->replyTerminator[i]);
  if (self->killPrivate)
    self->killPrivate(self);
}

pAsyncProtocol AsyncProtocolCreate(SicsInterp *pSics, const char* protocolName,
    ObjectFunc pFunc, KillFunc pKFunc) {
  int iRet;
  pAsyncProtocol self = NULL;

  /* try to find an existing queue with this name */
  self = (pAsyncProtocol) FindCommandData(pServ->pSics, (char *)protocolName, "AsyncProtocol");
  if (self != NULL) {
    return self;
  }

  self = (pAsyncProtocol) malloc(sizeof(AsyncProtocol));
  if (self == NULL) {
    SICSLogWrite("Out of memory in AsyncProtocolCreate", eError);
    return NULL;
  }
  memset(self, 0, sizeof(AsyncProtocol));
  self->pDes = CreateDescriptor("AsyncProtocol");
  if (pFunc == NULL)
    pFunc = AsyncProtocolNoAction;
  if (pKFunc == NULL)
    pKFunc = AsyncProtocolKill;
  iRet = AddCommand(pSics, (char *)protocolName, pFunc, pKFunc, self);
  if (!iRet ) {
    SICSLogWrite("AddCommand failed in AsyncProtocolCreate", eError);
    AsyncProtocolKill(self);
    return NULL;
  }
  self->sendCommand = defaultSendCommand;
  self->handleInput = defaultHandleInput;
  self->handleEvent = defaultHandleEvent;
  self->prepareTxn = defaultPrepareTxn;
  self->killPrivate = defaultKillPrivate;
  self->sendTerminator = strdup("\r\n");
  self->replyTerminator[0] = strdup("\r\n");
  return self;
}

int AsyncProtocolFactory(SConnection *pCon, SicsInterp *pSics,
    void *pData, int argc, char *argv[]) {
  if (argc < 2) {
    SCWrite(pCon,"ERROR: insufficient arguments to AsyncProtocolFactory", eError);
    return 0;
  }
  pAsyncProtocol pNew = AsyncProtocolCreate(pSics, argv[1],
      AsyncProtocolAction, AsyncProtocolKill);
  /* handle any extra arguments here */
  pNew->privateData = NULL;
  return 1;
}