binarytermreader.c

Go to the documentation of this file.

/*
 * Copyright (C) 2006  SRI International
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or (at your option) any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
 *
 * SRI International: 333 Ravenswood Ave, Menlo Park, CA 94025
 */

#ifdef _WINDOWS
#include <windows.h>
#include <winsock.h>
#else
#include <sys/types.h>
#include <sys/socket.h>
#include <unistd.h>     /* close        */
#include <sys/time.h>   /* For polling time */
#endif

#include <stdlib.h>
#include <string.h>
#include <math.h>
#include <stdio.h>

#include "binarytermreader.h"
#include "javadistoglib.h"
#include "binarytermtypes.h"
#include "libicl.h"
#include "libicl_private.h"
#include "glib.h"
#include "liboaa.h"

#ifdef _WINDOWS
// VC++ does not support const ints inside of array
// sizes. The following is an "enum hack" to get around
// this bug.
enum { INITBUFFERSZ = 8388608 };
enum { AMOUNTTOREAD = 8192 };
#else
static const int INITBUFFERSZ = 8388608;
static const int AMOUNTTOREAD = 8192;
#endif

struct BinaryTermReaderStruct
{
  TermReader* superReader;
  GByteArray* currentBuffer;
}
;

void binaryTermReader_cleanup(TermReader*);
ICLTerm* binaryTermReader_getNext(TermReader*, double);
ICLTerm* binaryTermReader_parseBuffer(BinaryTermReader* br);
void binaryTermReader_fillData(BinaryTermReader* br, double timeout);
void binaryTermReader_addToCurrentBuffer(BinaryTermReader* br, char* buf, ssize_t len);
void binaryTermReader_cleanupTermList(GQueue*);

BinaryTermReader* binaryTermReader_create(TermReader* t, gint listenSocket)
{
  BinaryTermReader* r = (BinaryTermReader*)malloc(sizeof(BinaryTermReader));
  r->superReader = t;
  r->currentBuffer = g_byte_array_new();
  termReader_setReaderSpecificData(t, r);
  termReader_setSocket(t, listenSocket);
  termReader_setType(t, BINARYTERMREADERTYPE);
  termReader_setGetNextCallback(t, binaryTermReader_getNext);
  termReader_setCleanupCallback(t, binaryTermReader_cleanup);
  termReader_setError(t, TERMREADER_OKAY);
  return r;
}

void binaryTermReader_cleanup(TermReader* reader)
{
  BinaryTermReader* br = (BinaryTermReader*)termReader_getReaderSpecificData(reader);
  if(br->currentBuffer != NULL) {
    free(br->currentBuffer);
  }
  if(br != NULL) {
    free(br);
  }
}

ICLTerm* binaryTermReader_getNext(TermReader* reader, double timeout)
{
  ICLTerm* nextEvent = NULL;
  gboolean done = FALSE;
  gboolean timedOut = FALSE;
  gboolean hadOldData = FALSE;
  gboolean forceFill = FALSE;
  BinaryTermReader* br = (BinaryTermReader*)termReader_getReaderSpecificData(reader);
  while(!done) {
    termReader_setError(reader, TERMREADER_OKAY);
    if((br->currentBuffer->len == 0) || forceFill) {
      binaryTermReader_fillData(br, timeout);
      if(termReader_getError(reader) == TERMREADER_TIMEOUT) {
        timedOut = TRUE;
      }
    }
    else {
      forceFill = FALSE;
      hadOldData = TRUE;
    }
    if((termReader_getError(reader) != TERMREADER_OKAY) &&
       !timedOut) {
      if(termReader_getError(reader) != TERMREADER_NOCONN) {
        // Don't report an error for NOCONN...typically
        // a client has simply disconnected cleanly.
        fprintf(stderr,
                "binaryTermReader_getNext() [%s %i] bad read: %i\n",
                __FILE__, __LINE__, termReader_getError(reader));
      }
      return NULL;
    }
    if(br->currentBuffer->len > 0) {
      CHECK_LEAKS();
      nextEvent = binaryTermReader_parseBuffer(br);
      CHECK_LEAKS();
      if(nextEvent == NULL) {
        /* If there was no event in the buffer, check for a timeout */
        if(timedOut) {
          nextEvent = icl_NewTermFromString("event(timeout)");
          return nextEvent;
        }
        else if(termReader_getError(reader) == TERMREADER_BINNOTENOUGHDATA) {
          forceFill = TRUE;
          hadOldData = FALSE;
          termReader_setError(reader, TERMREADER_OKAY);
          /*printf("Not enough data in buffer--continue\n");*/
          continue;
        }
        else if(termReader_getError(reader) != TERMREADER_OKAY) {
          fprintf(stderr, "error in binarytermreader\n");
          return NULL;
        }
        else if(hadOldData) {
          forceFill = TRUE;
          hadOldData = FALSE;
          continue;
        }
        else {
          continue;
        }
      }
      else {
        //char* finalTerm;
        //finalTerm = icl_NewStringFromTerm(nextEvent);
        /* printf("binaryTermReader_getNext() found next [%s]\n", finalTerm); */
        //icl_stFree(finalTerm);
        return nextEvent;
      }
    }
    else if(timedOut) {
      nextEvent = icl_NewTermFromString("event(timeout)");
      return nextEvent;
    }
    else {
      continue;
    }
  }
  return NULL;
}

void binaryTermReader_cleanupTermList(GQueue* termList) 
{
  ICLTerm* t;
  while(!g_queue_is_empty(termList)) {
    t = (ICLTerm*)g_queue_pop_tail(termList);
    icl_Free(t);
  }
  g_queue_free(termList);
}

ICLTerm* binaryTermReader_parseBuffer(BinaryTermReader* br)
{
  GQueue* termList = g_queue_new();
  gboolean done = FALSE;
  gint32 type;
  gint32 numArgs;
  gint32 dataLen;
  ICLTerm* newTerm;
  gint64 inIclInt = 0;
  double inIclFloat = 0;
  char* inString = NULL;
  char* inData = NULL;
  char* currPos = (char*)br->currentBuffer->data;
  size_t bufLen = br->currentBuffer->len;
  size_t offset = 0;
  size_t offsetIncr = 0;
  size_t strLen = 0;
  char starter = 0;
  ICLTerm* args;
  int i;

  /* printf("binaryTermReader_parseBuffer()\n"); */

  while(!done) {
    offsetIncr = javadistoglib_readJavaInt(currPos + offset, bufLen, &type);
    if(offsetIncr == 0) {
      /* printf("binaryTermReader_parseBuffer() not enough data for int (type) at offset %i\n", offset); */
      termReader_setError(br->superReader, TERMREADER_BINNOTENOUGHDATA);
      binaryTermReader_cleanupTermList(termList);
      return NULL;
    }
    offset += offsetIncr;
    bufLen -= offsetIncr;
    numArgs = 0;
    dataLen = 0;
    newTerm = NULL;
    switch(type) {
    case BTT_SENTINEL:
      /* printf("binaryTermReader_parseBuffer() hit sentinel\n"); */
      done = TRUE;
      newTerm = (ICLTerm*)g_queue_pop_tail(termList);
      if(!g_queue_is_empty(termList)) {
        fprintf(stderr, "binaryTermReader_parseBuffer() encountered sentinel but termList not empty\n");
        termReader_setError(br->superReader, TERMREADER_BINUNEXPECTEDSENTINEL);
        binaryTermReader_cleanupTermList(termList);
        return NULL;
      }
      else {
        GByteArray* oldBuf = br->currentBuffer;
        br->currentBuffer = g_byte_array_sized_new(br->currentBuffer->len - offset);
        g_byte_array_append(br->currentBuffer, oldBuf->data + offset, oldBuf->len - offset);
        g_byte_array_free(oldBuf, TRUE);
        binaryTermReader_cleanupTermList(termList);
        return newTerm;
      }
    case BTT_ICLINT:
      /* printf("binaryTermReader_parseBuffer() hit IclInt\n"); */
      offsetIncr = javadistoglib_readJavaLong(currPos + offset, bufLen, &inIclInt);
      if(offsetIncr == 0) {
        /* printf("binaryTermReader_parseBuffer() not enough data for long (int)\n"); */
        termReader_setError(br->superReader, TERMREADER_BINNOTENOUGHDATA);
        binaryTermReader_cleanupTermList(termList);
        return NULL;
      }
      offset += offsetIncr;
      bufLen -= offsetIncr;
      newTerm = icl_NewInt(inIclInt);
      break;
    case BTT_ICLFLOAT:
      /* printf("binaryTermReader_parseBuffer() hit IclFloat at offset %i\n", offset); */
      offsetIncr = javadistoglib_readJavaDouble(currPos + offset, bufLen, &inIclFloat);
      if(offsetIncr == 0) {
        /* printf("binaryTermReader_parseBuffer() not enough data for double (float)\n"); */
        termReader_setError(br->superReader, TERMREADER_BINNOTENOUGHDATA);
        binaryTermReader_cleanupTermList(termList);
        return NULL;
      }
      offset += offsetIncr;
      bufLen -= offsetIncr;
      newTerm = icl_NewFloat(inIclFloat);
      break;
    case BTT_ICLVAR:
      /* printf("binaryTermReader_parseBuffer() hit IclVar\n"); */
      offsetIncr = javadistoglib_readJavaInt(currPos + offset, bufLen, &dataLen);
      if(offsetIncr == 0) {
        /* printf("binaryTermReader_parseBuffer() not enough data for int (dataLen for Var)\n"); */
        termReader_setError(br->superReader, TERMREADER_BINNOTENOUGHDATA);
        binaryTermReader_cleanupTermList(termList);
        return NULL;
      }
      offset += offsetIncr;
      bufLen -= offsetIncr;
      if(dataLen < 0) {
        fprintf(stderr, "binaryTermReader_parseBuffer() ICLVAR has datalen < 0\n");
        termReader_setError(br->superReader, TERMREADER_BINBADICLVAR);
        binaryTermReader_cleanupTermList(termList);
        return NULL;
      }
      if(dataLen == 0) {
        newTerm = icl_NewVar("_");
      }
      else {
        inString = (char*)malloc((dataLen + 1) * sizeof(char));
        offsetIncr = javadistoglib_readJavaChars(currPos + offset, bufLen, dataLen, dataLen + 1, inString);
        if(offsetIncr == 0) {
          /* printf("binaryTermReader_parseBuffer() not enough data for chars (var)\n"); */
          termReader_setError(br->superReader, TERMREADER_BINNOTENOUGHDATA);
          free(inString);
          binaryTermReader_cleanupTermList(termList);
          return NULL;
        }
        offset += offsetIncr;
        bufLen -= offsetIncr;
        newTerm = icl_NewVarNoCopy(inString);
        inString = NULL;
      }
      break;
    case BTT_ICLSTR:
      /* printf("binaryTermReader_parseBuffer() hit IclStr offset for dataLen is %i\n", offset); */
      offsetIncr = javadistoglib_readJavaInt(currPos + offset, bufLen, &dataLen);
      if(offsetIncr == 0) {
        termReader_setError(br->superReader, TERMREADER_BINNOTENOUGHDATA);
        binaryTermReader_cleanupTermList(termList);
        return NULL;
      }
      offset += offsetIncr;
      bufLen -= offsetIncr;
      if(dataLen <= 0) {
        fprintf(stderr, "binaryTermReader_parseBuffer() ICLSTR has datalen <= 0\n");
        termReader_setError(br->superReader, TERMREADER_BINBADICLSTR);
        binaryTermReader_cleanupTermList(termList);
        return NULL;
      }
      inString = (char*)malloc((dataLen + 1) * sizeof(char));
      offsetIncr = javadistoglib_readJavaChars(currPos + offset, bufLen, dataLen, dataLen + 1, inString);
      if(offsetIncr == 0) {
        /* printf("binaryTermReader_parseBuffer() not enough data for chars (Str)\n"); */
        termReader_setError(br->superReader, TERMREADER_BINNOTENOUGHDATA);
        binaryTermReader_cleanupTermList(termList);
        free(inString);
        return NULL;
      }
      offset += offsetIncr;
      bufLen -= offsetIncr;
      icl_safeStRemoveQuotes(inString, dataLen, &strLen);
      icl_safeStUndoubleQuotes(inString, strLen, &strLen);
      newTerm = icl_NewStrNoCopy(inString);
      inString = NULL;
      break;
    case BTT_ICLDATAQ:
      offsetIncr = javadistoglib_readJavaInt(currPos + offset, bufLen, &dataLen);
      if(offsetIncr == 0) {
        termReader_setError(br->superReader, TERMREADER_BINNOTENOUGHDATA);
        binaryTermReader_cleanupTermList(termList);
        return NULL;
      }
      offset += offsetIncr;
      bufLen -= offsetIncr;
      if(dataLen <= 0) {
        fprintf(stderr, "binaryTermReader_parseBuffer() ICLDATAQ has datalen <= 0\n");
      }
      inData = (char*)malloc(dataLen * sizeof(char));
      offsetIncr = javadistoglib_readJavaBytes(currPos + offset, bufLen, dataLen, dataLen, inData);
      if(offsetIncr == 0) {
        termReader_setError(br->superReader, TERMREADER_BINNOTENOUGHDATA);
        binaryTermReader_cleanupTermList(termList);
        free(inData);
        return NULL;
      }
      offset += offsetIncr;
      newTerm = icl_NewDataQNoCopy(inData, dataLen);
      inData = NULL;
      break;
    case BTT_ICLSTRUCT:
      /* printf("binaryTermReader_parseBuffer() hit IclStruct\n"); */
      offsetIncr = javadistoglib_readJavaInt(currPos + offset, bufLen, &numArgs);
      if(offsetIncr == 0) {
        /* printf("binaryTermReader_parseBuffer() not enough data for int (numArgs for struct)\n"); */
        termReader_setError(br->superReader, TERMREADER_BINNOTENOUGHDATA);
        binaryTermReader_cleanupTermList(termList);
        return NULL;
      }
      offset += offsetIncr;
      bufLen -= offsetIncr;
      offsetIncr = javadistoglib_readJavaInt(currPos + offset, bufLen, &dataLen);
      offset += offsetIncr;
      bufLen -= offsetIncr;
      if(offsetIncr == 0) {
        /* printf("binaryTermReader_parseBuffer() not enough data for int (dataLen for struct)\n"); */
        termReader_setError(br->superReader, TERMREADER_BINNOTENOUGHDATA);
        binaryTermReader_cleanupTermList(termList);
        return NULL;
      }
      if(dataLen <= 0) {
        fprintf(stderr, "binaryTermReader_parseBuffer() ICLSTRUCT has datalen <= 0\n");
        termReader_setError(br->superReader, TERMREADER_BINBADICLSTRUCT);
        binaryTermReader_cleanupTermList(termList);
        return NULL;
      }
      inString = (char*)malloc((dataLen + 1) * sizeof(char));
      offsetIncr = javadistoglib_readJavaChars(currPos + offset, bufLen, dataLen, dataLen + 1, inString);
      if(offsetIncr == 0) {
        /* printf("binaryTermReader_parseBuffer() not enough data for chars (functor)\n"); */
        termReader_setError(br->superReader, TERMREADER_BINNOTENOUGHDATA);
        binaryTermReader_cleanupTermList(termList);
        free(inString);
        return NULL;
      }
      offset += offsetIncr;
      bufLen -= offsetIncr;
      args = icl_NewList(NULL);
      for(i = 0; i < numArgs; ++i) {
        if(g_queue_is_empty(termList)) {
          fprintf(stderr, "binaryTermReader_parseBuffer() ICLSTRUCT not enough args\n");
          termReader_setError(br->superReader, TERMREADER_BINNOTENOUGHARGS);
          binaryTermReader_cleanupTermList(termList);
          return NULL;
        }
        icl_AddToList(args, (ICLTerm*)g_queue_pop_tail(termList), TRUE);
      }
      newTerm = icl_NewStructFromList(inString, args);
      free(inString);
      inString = NULL;
      break;
    case BTT_ICLLIST:
      /* printf("binaryTermReader_parseBuffer() hit IclList\n"); */
      offsetIncr = javadistoglib_readJavaInt(currPos + offset, bufLen, &numArgs);
      if(offsetIncr == 0) {
        /* printf("binaryTermReader_parseBuffer() not enough data for int (numArgs for list)\n"); */
        termReader_setError(br->superReader, TERMREADER_BINNOTENOUGHDATA);
        binaryTermReader_cleanupTermList(termList);
        return NULL;
      }
      offset += offsetIncr;
      bufLen -= offsetIncr;
      newTerm = icl_NewList(NULL);
      for(i = 0; i < numArgs; ++i) {
        if(g_queue_is_empty(termList)) {
          fprintf(stderr, "binaryTermReader_parseBuffer() ICLLIST not enough args\n");
          termReader_setError(br->superReader, TERMREADER_BINNOTENOUGHARGS);
          binaryTermReader_cleanupTermList(termList);
          return NULL;
        }
        icl_AddToList(newTerm, (ICLTerm*)g_queue_pop_tail(termList), TRUE);
      }
      break;
    case BTT_ICLGROUP:
      /* printf("binaryTermReader_parseBuffer() hit IclGroup\n"); */
      offsetIncr = javadistoglib_readJavaInt(currPos + offset, bufLen, &numArgs);
      if(offsetIncr == 0) {
        /* printf("binaryTermReader_parseBuffer() not enough data for int (numArgs for group)\n"); */
        termReader_setError(br->superReader, TERMREADER_BINNOTENOUGHDATA);
        binaryTermReader_cleanupTermList(termList);
        return NULL;
      }
      offset += offsetIncr;
      bufLen -= offsetIncr;
      offsetIncr = javadistoglib_readJavaChar(currPos + offset, bufLen, &starter);
      if(offsetIncr == 0) {
        /* printf("binaryTermReader_parseBuffer() not enough data for char (starter for group)\n"); */
        termReader_setError(br->superReader, TERMREADER_BINNOTENOUGHDATA);
        binaryTermReader_cleanupTermList(termList);
        return NULL;
      }
      offset += offsetIncr;
      bufLen -= offsetIncr;
      newTerm = icl_NewGroup(starter, ",", NULL);
      for(i = 0; i < numArgs; ++i) {
        if(g_queue_is_empty(termList)) {
          fprintf(stderr, "binaryTermReader_parseBuffer() ICLGROUP not enough args\n");
          termReader_setError(br->superReader, TERMREADER_BINNOTENOUGHARGS);
          binaryTermReader_cleanupTermList(termList);
          return NULL;
        }
        icl_AddToList(newTerm, (ICLTerm*)g_queue_pop_tail(termList), TRUE);
      }
      break;
    default:
      {
        char* format;
        if(sizeof(offset) == 4) {
          format = "binaryTermReader_parseBuffer() encountered unexpected type: [%i] at offset %d\n";
        }
        else {
          format = "binaryTermReader_parseBuffer() encountered unexpected type: [%i] at offset %ld\n";          
        }
        fprintf(stderr, format, type, offset - offsetIncr);
      }
      termReader_setError(br->superReader, TERMREADER_BINUNEXPECTEDTYPE);
      binaryTermReader_cleanupTermList(termList);
      return NULL;
    }
    g_queue_push_tail(termList, newTerm);
  }
  fprintf(stderr, "binaryTermReader_parseBuffer() unexpected exit from loop\n");
  termReader_setError(br->superReader, TERMREADER_BINUNREACHABLE);
  binaryTermReader_cleanupTermList(termList);
  return NULL;
}

void binaryTermReader_fillData(BinaryTermReader* br, double timeout)
{
  char buf[AMOUNTTOREAD + 1];
  struct timeval time;
  struct timeval* timep = NULL;
  int lastFdPlusOne;
  int selectRes = 0;
  ssize_t numBytes = 0;
  fd_set readfds;

  /* printf("binaryTermReader_fillData()\n"); */
  if(timeout > 0) {
    long i = (long)floor(timeout);
    time.tv_sec = i;
    time.tv_usec = (long)floor((timeout - i) * 1e6);
    timep = &time;
  }

  lastFdPlusOne = termReader_getSocket(br->superReader) + 1;
  FD_ZERO(&readfds);
  FD_SET(termReader_getSocket(br->superReader), &readfds);

  selectRes = select(lastFdPlusOne, &readfds, NULL, NULL, timep);

  switch(selectRes) {
  case -1:
    perror("binaryTermReader_fillData bad select");
    termReader_setError(br->superReader, TERMREADER_SELECTERR);
    return;
  case 0:
    /*fprintf(stderr, "binaryTermReader_fillData time out\n"); */
    termReader_setError(br->superReader, TERMREADER_TIMEOUT);
    return;
  default:
#ifdef _WINDOWS
    numBytes = recv(termReader_getSocket(br->superReader), buf, AMOUNTTOREAD, 0);
#else
    numBytes = read(termReader_getSocket(br->superReader), buf, AMOUNTTOREAD);
#endif

    switch(numBytes) {
    case -1:
      perror("binaryTermReader_fillData bad read");
      termReader_setError(br->superReader, TERMREADER_READERR);
      return;
    case 0:
      /*fprintf(stderr, "binaryTermReader_fillData no connection\n");*/
      termReader_setError(br->superReader, TERMREADER_NOCONN);
      return;
    }
  }

  binaryTermReader_addToCurrentBuffer(br, buf, numBytes);
  termReader_setError(br->superReader, TERMREADER_OKAY);
  /*
  {
    printf("binaryTermReader_fillData() added %i bytes; buffer is now size %i\n", numBytes, br->currentBuffer->len);
  }
  */
}

void binaryTermReader_addToCurrentBuffer(BinaryTermReader* br, char* buf, ssize_t len)
{
  g_byte_array_append(br->currentBuffer, (guint8*)buf, len);
  /*
  size_t needCapacity = len + br->currentBuffer->len + 1;
  char* oldBuf;

  if(needCapacity > br->bufCapacity) {
    while(br->bufCapacity < needCapacity) {
      br->bufCapacity *= 2;
    }
    oldBuf = br->currentBuffer;
    br->currentBuffer = (char*)realloc(br->currentBuffer, br->bufCapacity);
    if(br->currentBuffer == NULL) {
      br->currentBuffer = (char*)malloc(br->bufCapacity);
      memcpy(br->currentBuffer, oldBuf, br->currentBuffer->len);
      free(oldBuf);
    }
  }
  memcpy(br->currentBuffer + br->currentBuffer->len, buf, len);
  br->currentBuffer->len += len;
  */
  /*
    printf("binaryTermReader_addToCurrentBuffer() used = %i\n", br->bufUsed);
  */
}

---