stringtermreader.c

来自「SRI international 发布的OAA框架软件」· C语言 代码 · 共 293 行

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

#include "stdpccts.h"

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

#include "stringtermreader.h"
#include "libicl_private.h"
#include "stringbuffer.h"
#include "stringbuffer_private.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 = 8;
static const int AMOUNTTOREAD = 8192;
#endif

struct StringTermReaderStruct
{
  TermReader* superReader;
  char* currentBuffer;
  size_t bufCapacity;
  size_t bufUsed;
}
;

void stringTermReader_cleanup(TermReader*);
ICLTerm* stringTermReader_getNext(TermReader*, double);
void stringTermReader_fillData(StringTermReader* sr, double timeout);
void stringTermReader_addToCurrentBuffer(StringTermReader* sr, char* buf, ssize_t len);

StringTermReader* stringTermReader_create(TermReader* t, gint listenSocket)
{
  StringTermReader* r = (StringTermReader*)malloc(sizeof(StringTermReader));
  r->superReader = t;
  r->currentBuffer = (char*)malloc(8);
  r->bufUsed = 0;
  r->bufCapacity = 8;
  termReader_setReaderSpecificData(t, r);
  termReader_setSocket(t, listenSocket);
  termReader_setType(t, STRINGTERMREADERTYPE);
  termReader_setGetNextCallback(t, stringTermReader_getNext);
  termReader_setCleanupCallback(t, stringTermReader_cleanup);
  termReader_setError(t, TERMREADER_OKAY);
  return r;
}

void stringTermReader_cleanup(TermReader* reader)
{
  StringTermReader* sr = (StringTermReader*)termReader_getReaderSpecificData(reader);
  if(sr->currentBuffer != NULL) {
    free(sr->currentBuffer);
  }
  if(sr != NULL) {
    free(sr);
  }
}

ICLTerm* stringTermReader_getNext(TermReader* reader, double timeout)
{
  ICLTerm* nextEvent = NULL;
  stringbuffer_t sbuf;
  gboolean done = FALSE;
  gboolean timedOut = FALSE;
  gboolean hadOldData = FALSE;
  gboolean forceFill = FALSE;
  StringTermReader* sr = (StringTermReader*)termReader_getReaderSpecificData(reader);
  
  CHECK_LEAKS();
  /*
  printf("stringTermReader_getNext\n");
  */
  while(!done) {
    if((sr->bufUsed == 0) || forceFill) {
      /*
      printf("stringTermReader_getNext calling fillData\n");
      */
      stringTermReader_fillData(sr, timeout);
      CHECK_LEAKS();
      if(termReader_getError(reader) == TERMREADER_TIMEOUT) {
        /*
        printf("stringTermReader_getNext timed out\n");
        */
        timedOut = TRUE;
      }
    }
    else {
      /*
      printf("stringTermReader_getNext hadOldData\n");
      */
      forceFill = FALSE;
      hadOldData = TRUE;
    }

      /*
    printf("stringTermReader_getNext bufUsed = %i\n", sr->bufUsed);
    {
      unsigned int i;
      printf("stringTermReader_getNext buffer contents: ");
      for(i = 0; i < sr->bufUsed; ++i) {
        printf(" %o ", sr->currentBuffer[i]);
      }
      printf("\n");

      {
        char* data = (char*)malloc(sr->bufUsed + 1);
        memset(data, 0, sr->bufUsed + 1);
        memcpy(data, sr->currentBuffer, sr->bufUsed);
        printf("stringTermReader_getNext current buffer contains [%s]\n", data);
        free(data);
      }
    }
    parser_setDebug(TRUE);
      */



	// XXX Unless a no-op sleep is put here, sometimes
	// communications can hang
	sleep_millis(0);

    sbuf.data = sr->currentBuffer;
    sbuf.len = sr->bufUsed;
    sbuf.index = 0;
    CHECK_LEAKS();
    if((sr->bufUsed > 0) &&
       (parser_getNetTermFromBuf(&nextEvent, &sbuf) != FALSE)) {
      size_t newUsed;
      CHECK_LEAKS();
      newUsed = sr->bufUsed - stringbuffer_getIndex(&sbuf);
      sr->currentBuffer = memmove(sr->currentBuffer,
                                  sr->currentBuffer + stringbuffer_getIndex(&sbuf),
                                  newUsed);
      sr->bufUsed = newUsed;
      /*sr->currentBuffer[sr->bufUsed] = '\0';*/
      done = TRUE;
      /*
      {
        char* ds = icl_NewStringFromTerm(nextEvent);
        printf("stringTermReader_getNext looks good from string:  got term %s\n", ds);
        icl_stFree(ds);
      }
      */
      parser_setDebug(FALSE);
      CHECK_LEAKS();
      return nextEvent;
    }
    CHECK_LEAKS();
    parser_setDebug(FALSE);

    if(timedOut) {
      CHECK_LEAKS();
      nextEvent = icl_NewTermFromData("event(timeout)", 14);
      CHECK_LEAKS();
      /*
      printf("stringTermReader_getNext timeout\n");
      */
      return nextEvent;
    }
    else if(termReader_getError(reader) != TERMREADER_OKAY) {
      /*
      printf("stringTermReader_getNext termReader_getError != TERMREADER_OKAY\n");
      */
    CHECK_LEAKS();
    return NULL;
    }
    else if(hadOldData) {
      /*
      printf("stringTermReader_getNext hadOldData, now forceFill\n");
      */
      forceFill = TRUE;
      hadOldData = FALSE;
      CHECK_LEAKS();
      continue;
    }
    else {
      /*
      printf("stringTermReader_getNext just continuing\n");
      */
      CHECK_LEAKS();
      continue;
    }
  }
  /*
  printf("stringTermReader_getNext NULL return\n");
  */
  CHECK_LEAKS();
  return NULL;
}

void stringTermReader_fillData(StringTermReader* sr, 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;
  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(sr->superReader) + 1;
  FD_ZERO(&readfds);
  FD_SET(termReader_getSocket(sr->superReader), &readfds);

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

  switch(selectRes) {
  case -1:
    perror("stringTermReader_fillData bad select");
    termReader_setError(sr->superReader, TERMREADER_SELECTERR);
    return;
  case 0:
    termReader_setError(sr->superReader, TERMREADER_TIMEOUT);
    return;
  default:
#ifdef _WINDOWS
    numBytes = recv(termReader_getSocket(sr->superReader), buf, AMOUNTTOREAD, 0);
#else
    numBytes = read(termReader_getSocket(sr->superReader), buf, AMOUNTTOREAD);
#endif

    switch(numBytes) {
    case -1:
#ifndef _WINDOWS
      // This error is printed in the Windows version
      // after ordinary, clean client disconnects
      // and is a distraction.
      perror("stringTermReader_fillData bad read");
#endif
      termReader_setError(sr->superReader, TERMREADER_READERR);
      return;
    case 0:
      termReader_setError(sr->superReader, TERMREADER_NOCONN);
      return;
    }
  }

  buf[numBytes] = '\0';

  stringTermReader_addToCurrentBuffer(sr, buf, numBytes);
  termReader_setError(sr->superReader, TERMREADER_OKAY);
}

void stringTermReader_addToCurrentBuffer(StringTermReader* sr, char* buf, ssize_t len)
{
  // Check our current buffer capacity--double or halve it as appropriate
  // Actually, I don't know if halving it is efficient or not...
  size_t needCapacity = len + sr->bufUsed + 1;

  if(needCapacity > sr->bufCapacity) {
    while(sr->bufCapacity < needCapacity) {
      sr->bufCapacity *= 2;
    }
    sr->currentBuffer = realloc(sr->currentBuffer, sr->bufCapacity);
  }
  else if(needCapacity < (sr->bufCapacity / 2)) {
    sr->bufCapacity /= 2;
    sr->currentBuffer = realloc(sr->currentBuffer, sr->bufCapacity);
  }
  else {
    // Just leave it alone
  }

  memcpy(sr->currentBuffer + sr->bufUsed, buf, len);
  sr->bufUsed += len;
  sr->currentBuffer[sr->bufUsed] = '\0';
}