binarytermreader.c

来自「SRI international 发布的OAA框架软件」· C语言 代码 · 共 542 行 · 第 1/2 页

#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 = 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);