binarytermsender.c

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

#include <string.h>
#include "binarytermsender.h"
#include "binarytermtypes.h"
#include "glibtojava.h"
#include "libicl_private.h"

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

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

struct BinaryTermSenderStruct
{
  TermSender* superSender;
}
;

typedef struct TermRecordStruct
{
  ICLTerm* term;
  int numTimesVisited;
}
TermRecord;

static TermRecord* binaryTermSender_newTermRecord(ICLTerm* term)
{
  TermRecord* t = (TermRecord*)malloc(sizeof(TermRecord));
  t->numTimesVisited = 0;
  t->term = term;
  return t;
}

void binaryTermSender_sendTerm(TermSender* ts, ICLTerm* toSend);
void binaryTermSender_cleanup(TermSender* ts);

BinaryTermSender* binaryTermSender_create(TermSender* t, gint sendSocket)
{
  BinaryTermSender* s = (BinaryTermSender*)malloc(sizeof(BinaryTermSender));
  s->superSender = t;
  termSender_setSendTermCallback(t, binaryTermSender_sendTerm);
  termSender_setCleanupCallback(t, binaryTermSender_cleanup);
  termSender_setType(t, BINARYTERMSENDERTYPE);
  termSender_setSocket(t, sendSocket);
  termSender_setSenderSpecificData(t, s);
  return s;
}

/**
 *
 */
void binaryTermSender_sendTerm(TermSender* ts, ICLTerm* inputTerm)
{
  /**
   * The proper way to do things is to send chunks of data at a time--that
   * way, we don't have to have everything in memory.  So first, you'd validate
   * the term, to make sure you could send it, then you actually send it.  Any
   * errors would then be fatal.
   * As it stands, errors cause terms not to be sent at all.
   */
  GByteArray* sendBuf = g_byte_array_new();
  GQueue* termRecordList = g_queue_new();
  TermRecord* currentRecord;
  ssize_t sent;

  /*
   * Send data in a postfix order.  So, if we had a term like,
   * ev_solve(one, [two, three(four, [five])])
   * we end up sending something that looks like (after wrapping it
   * with term):
   * atom length 4 data is 'five'
   * list length 1
   * atom length 4 data is 'four'
   * struct functor with length 5 and named 'three' with two args
   * atom length 3 data is 'two'
   * list length 2
   * atom length 3 data is 'one'
   * struct functor with length 8 and named 'ev_solve' with two args
   * struct functor with length 4 and named 'term' with 1 arg
   */
  g_queue_push_tail(termRecordList, binaryTermSender_newTermRecord(inputTerm));
  while(!g_queue_is_empty(termRecordList)) {
    currentRecord = (TermRecord*)g_queue_pop_tail(termRecordList);
    ++currentRecord->numTimesVisited;
    if(currentRecord->numTimesVisited == 2) {
      if(icl_IsStruct(currentRecord->term)) {
        gchar* functor = icl_Functor(currentRecord->term);
        sendBuf = glibtojava_writeJavaInt(sendBuf, BTT_ICLSTRUCT);
        sendBuf = glibtojava_writeJavaInt(sendBuf, icl_NumTerms(currentRecord->term));
        sendBuf = glibtojava_writeJavaInt(sendBuf, strlen(functor));
        sendBuf = glibtojava_writeJavaChars(sendBuf, strlen(functor), functor);
        free(currentRecord);
      }
      else if(icl_IsList(currentRecord->term)) {
        sendBuf = glibtojava_writeJavaInt(sendBuf, BTT_ICLLIST);
        sendBuf = glibtojava_writeJavaInt(sendBuf, icl_NumTerms(currentRecord->term));
        free(currentRecord);
      }
      else if(icl_IsGroup(currentRecord->term)) {
        gchar starter;
        gchar* sep;
        icl_GetGroupChars(currentRecord->term, &starter, &sep);
        sendBuf = glibtojava_writeJavaInt(sendBuf, BTT_ICLGROUP);
        sendBuf = glibtojava_writeJavaInt(sendBuf, icl_NumTerms(currentRecord->term));
        sendBuf = glibtojava_writeJavaChar(sendBuf, starter);
        free(currentRecord);
      }
      else {
        fprintf(stderr, "binaryTermSender_sendTerm() Unexpected type visited twice\n");
        termSender_setError(ts, TERMSENDER_BINUNKNOWNTYPETWICE);
        return;
      }
      continue;
    }
    else {
      if(icl_IsInt(currentRecord->term)) {
        sendBuf = glibtojava_writeJavaInt(sendBuf, BTT_ICLINT);
        sendBuf = glibtojava_writeJavaLong(sendBuf, icl_Int(currentRecord->term));
        free(currentRecord);
      }
      else if(icl_IsFloat(currentRecord->term)) {
        sendBuf = glibtojava_writeJavaInt(sendBuf, BTT_ICLFLOAT);
        sendBuf = glibtojava_writeJavaDouble(sendBuf, icl_Float(currentRecord->term));
        free(currentRecord);
      }
      else if(icl_IsVar(currentRecord->term)) {
        gchar* name = icl_Str(currentRecord->term);
        sendBuf = glibtojava_writeJavaInt(sendBuf, BTT_ICLVAR);
        sendBuf = glibtojava_writeJavaInt(sendBuf, strlen(name));
        sendBuf = glibtojava_writeJavaChars(sendBuf, strlen(name), name);
        free(currentRecord);
      }
      else if(icl_IsStr(currentRecord->term)) {
        gchar* name = icl_Str(currentRecord->term);
        sendBuf = glibtojava_writeJavaInt(sendBuf, BTT_ICLSTR);
        sendBuf = glibtojava_writeJavaInt(sendBuf, strlen(name));
        sendBuf = glibtojava_writeJavaChars(sendBuf, strlen(name), name);
        free(currentRecord);
      }
      else if(icl_IsDataQ(currentRecord->term)) {
        gchar* data = icl_DataQ(currentRecord->term);
        sendBuf = glibtojava_writeJavaInt(sendBuf, BTT_ICLDATAQ);
        sendBuf = glibtojava_writeJavaInt(sendBuf, currentRecord->term->len);
        sendBuf = glibtojava_writeJavaBytes(sendBuf, currentRecord->term->len, data);
        printf("sendBuf (%p) added iclDataQ of length %i, and is now of size %i\n", sendBuf->data, currentRecord->term->len, sendBuf->len);
        free(currentRecord);
      }
      else if(icl_IsStruct(currentRecord->term) ||
              icl_IsList(currentRecord->term) ||
              icl_IsGroup(currentRecord->term)) {
        ICLListType* args = icl_List(currentRecord->term);
        g_queue_push_tail(termRecordList, currentRecord);
        while(args != NULL) {
          if(icl_IsValid(icl_ListElement(args))) {
            g_queue_push_tail(termRecordList, binaryTermSender_newTermRecord(icl_ListElement(args)));
          }
          else {
            fprintf(stderr, "binaryTermSender_sendTerm() invalid element to add to queue\n");
            termSender_setError(ts, TERMSENDER_BININVALIDELEMENT);
            return;
          }
          args = icl_ListNextElement(args);
        }
      }
      else {
        fprintf(stderr, "binaryTermSender_sendTerm() Unexpected type visited once\n");
        termSender_setError(ts, TERMSENDER_BINUNKNOWNTYPEONCE);
        return;
      }
    }
  }
  /*
   * Add the 'term' struct which has one argument, which is the single term that came before
   */
  {
    gchar* functor = "term";
    sendBuf = glibtojava_writeJavaInt(sendBuf, BTT_ICLSTRUCT);
    sendBuf = glibtojava_writeJavaInt(sendBuf, 1);
    sendBuf = glibtojava_writeJavaInt(sendBuf, 4);
    sendBuf = glibtojava_writeJavaChars(sendBuf, 4, functor);
  }
  
  sendBuf = glibtojava_writeJavaInt(sendBuf, BTT_SENTINEL);

#ifdef _WINDOWS
  if(send(termSender_getSocket(ts), sendBuf->data, sendBuf->len, 0) == SOCKET_ERROR) {
    termSender_setError(ts, TERMSENDER_BADSEND);
    return;
  }
#else
  sent = write(termSender_getSocket(ts), sendBuf->data, sendBuf->len);
  if((sent < 0) ||
     ((sent > 0) && ((size_t)sent != sendBuf->len)))
  {
    termSender_setError(ts, TERMSENDER_BADSEND);
    return;
  }
  fsync(termSender_getSocket(ts));
#endif

  g_queue_free(termRecordList);
  g_byte_array_free(sendBuf, TRUE);
}

void binaryTermSender_cleanup(TermSender* ts)
{
  if(termSender_getSenderSpecificData(ts) != NULL) {
    free(termSender_getSenderSpecificData(ts));
    termSender_setSenderSpecificData(ts, NULL);
  }
}