
/*
 *   THIS FILE IS UNDER RCS - DO NOT MODIFY UNLESS YOU HAVE
 *   CHECKED IT OUT USING THE COMMAND CHECKOUT.
 *
 *    $Id: xfrm.c,v 1.1 2009/12/16 12:00:00 scott Exp $
 *
 *    Revision history:
 *     $Log: xfrm.c,v $
 *
 *     Revision 1.1  2009/12/16 12:00:00 scott
 *     Initial revision
 *
 *
 */

/*
 * xfrm.c:  
 *
 * 
 *  Initial version:
 *      
 */

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/types.h>
#include <time.h>
#include <earthworm.h>
#include <kom.h>
#include <swap.h>
#ifdef _WINNT
#include <windows.h>
#define mutex_t HANDLE
#else
#ifdef _SOLARIS
#ifndef _LINUX
#include <synch.h>      /* mutex's                                      */
#endif
#endif
#endif
#include <trheadconv.h> /* trheadconv()                                 */
#include <xfrm.h>

#define SCNL_INC     10      /* how many more are allocated each     */
                                /*   time we run out                    */
#define NUM_BASE_CMDS 11

static int  maxSta = 0;     /* number of scnls allocated   */
static char **BaseCmd;      /* array of command definitions */
static char **ModCmds;      /* array of module-specific command defs */
static int  CmdCount;       /* number of command definitions */
static void **ParamTarget;  /* array of addrs for value of matching commands */
static void **ModParamTargets; /* ParamTargets for ModCmds */


static void InitializeParameters( XFRMWORLD* pXfrm )
{
  int i;

  /*    Initialize members of the XFRMWORLD structure                    */
  pXfrm->completionFcn          = NULL;

  /*    Initialize members of the xfrmEWH structure                      */
  pXfrm->xfrmEWH->myInstId = 0;
  pXfrm->xfrmEWH->myModId  = 0;
  pXfrm->xfrmEWH->typeError     = 0;
  pXfrm->xfrmEWH->typeHeartbeat = 0;
  pXfrm->xfrmEWH->ringInKey     = 0l;
  pXfrm->xfrmEWH->ringOutKey    = 0l;
  pXfrm->xfrmEWH->typeTrace     = 0;
  pXfrm->xfrmEWH->typeTrace2    = 0;

  /*    Initialize members of the xfrmParam structure                    */
  for( i=0; i<MAX_LOGO; i++ ) {
     sprintf( pXfrm->xfrmParam->readInstName[i], "" );
     sprintf( pXfrm->xfrmParam->readModName[i],  "" );
     sprintf( pXfrm->xfrmParam->readTypeName[i], "" );
  }
  pXfrm->xfrmParam->ringIn[0]      = '\0';
  pXfrm->xfrmParam->ringOut[0]     = '\0';
  pXfrm->xfrmParam->nlogo          = 0;
  pXfrm->xfrmParam->heartbeatInt   = 15;
  pXfrm->xfrmParam->debug          = 0;
  pXfrm->xfrmParam->logSwitch      = 1;
  pXfrm->xfrmParam->cleanStart     = 1;
  pXfrm->xfrmParam->testMode       = 0;
  pXfrm->xfrmParam->maxGap         = 1.5;

  InitializeXfrmParameters();
  
  return;
}


static int IsWild( char *str )
{
   if( strcmp( str, "*" ) == 0 ) return 1;
   return 0;
}

static int BaseProcessXfrmCommand( XFRMWORLD* pXfrm, int cmd_id, char *com ) {
  int  scnl;
  char *str;
  char cnull = '\0';    
  XFRMSCNL *scnls;
  int nSCNL;
  int ilogo;

  switch ( cmd_id ) {
  case 8: /* GetSCNL or GetSCN */
    scnl = strcmp(com,"GetSCN");
    if (pXfrm->nSCNL >= maxSta) 
    {
   /* Need to allocate more */
      maxSta += SCNL_INC;
      if ((pXfrm->scnls = (XFRMSCNL *) realloc (pXfrm->scnls, 
                         (maxSta * sizeof (XFRMSCNL)))) == NULL)
      {
        logit ("e", "%s: realloc for SCNL list failed; exiting!\n", pXfrm->mod_name );
        return -2;
      }
    }
    scnls = pXfrm->scnls;
    nSCNL = pXfrm->nSCNL;
    str = k_str(); /* read input SCNL code */
    if( !str  ||  strlen(str) >= TRACE2_STA_LEN  ||  IsWild(str) ) {
        if( !str ) str = &cnull;
        logit ("e", "%s: Invalid input SCNL code: %s "
                    "in GetSCNL cmd; exiting!\n", pXfrm->mod_name, str);
        return -2;
    }
    strncpy(scnls[nSCNL].inSta, str, TRACE2_STA_LEN) ;

    str = k_str(); /* read input component code */
    if( !str  ||  strlen(str) >= TRACE2_CHAN_LEN  ||  IsWild(str)) {
        if( !str ) str = &cnull;
        logit ("e", "%s: Invalid input component code: %s "
                    "in GetSCNL cmd; exiting!\n", pXfrm->mod_name, str);
        return -2;
    }
    strncpy(scnls[nSCNL].inChan, str, TRACE2_CHAN_LEN);

    str = k_str(); /* read input network code */
    if( !str  ||  strlen(str) >= TRACE2_NET_LEN  ||  IsWild(str)) {
        if( !str ) str = &cnull;
        logit ("e", "%s: Invalid input network code: %s "
                    "in GetSCNL cmd; exiting!\n", pXfrm->mod_name, str);
        return -2;
    }
    strncpy(scnls[nSCNL].inNet, str, TRACE2_NET_LEN);

    if( scnl ) {  /* read input location code */
      str = k_str();
      if( !str  ||  strlen(str) >= TRACE2_LOC_LEN  ||  IsWild(str)) {
          if( !str ) str = &cnull;
          logit ("e", "%s: Invalid input location code: %s "
                      "in GetSCNL cmd; exiting!\n", pXfrm->mod_name, str);
          return -2;
      }
      strncpy(scnls[nSCNL].inLoc, str, TRACE2_LOC_LEN);
    } 
    else {      /* use default blank location code */
      strncpy(scnls[nSCNL].inLoc, LOC_NULL_STRING, TRACE2_LOC_LEN);
    }

    str = k_str(); /* read output SCNL code */
    if( !str  ||  strlen(str) >= TRACE2_STA_LEN  ||  IsWild(str)) {
        if( !str ) str = &cnull;
        logit ("e", "%s: Invalid output SCNL code: %s "
                    "in GetSCNL cmd; exiting!\n", pXfrm->mod_name, str);
        return -2;
    }
    strncpy(scnls[nSCNL].outSta, str, TRACE2_STA_LEN);

    str = k_str(); /* read output component code */
    if( !str  ||  strlen(str) >= TRACE2_CHAN_LEN  ||  IsWild(str)) {
        if( !str ) str = &cnull;
        logit ("e", "%s: Invalid output component code: %s "
                    "in GetSCNL cmd; exiting!\n", pXfrm->mod_name, str);
        return -2;
    }
    strncpy(scnls[nSCNL].outChan, str, TRACE2_CHAN_LEN);

    str = k_str(); /* read output network code */
    if( !str  ||  strlen(str) >= TRACE2_NET_LEN  ||  IsWild(str)) {
        if( !str ) str = &cnull;
        logit ("e", "%s: Invalid output network code: %s "
                    "in GetSCNL cmd; exiting!\n", pXfrm->mod_name, str);
        return -2;
    }
    strncpy(scnls[nSCNL].outNet, str, TRACE2_NET_LEN);

    if( scnl ) {  /* read output location code */
      str = k_str();
      if( !str  ||  strlen(str) >= TRACE2_LOC_LEN  ||  IsWild(str)) {
          if( !str ) str = &cnull;
          logit ("e", "%s: Invalid output location code: %s "
                      "in GetSCNL cmd; exiting!\n", pXfrm->mod_name, str );
          return -2;
      }
      strncpy(scnls[nSCNL].outLoc, str, TRACE2_LOC_LEN);
    } 
    else {      /* use default blank location code */
      strncpy(scnls[nSCNL].outLoc, LOC_NULL_STRING, TRACE2_LOC_LEN);
    }
    scnls[nSCNL].status = -1; /* uninitialized */
    pXfrm->nSCNL++;
    break;
  case 10: /* GetWavesFrom */
	ilogo = pXfrm->xfrmParam->nlogo;
	if(ilogo+1 >= MAX_LOGO) 
    {
       logit( "e","%s: Too many GetWavesFrom commands; "
              "max=%d; exiting!\n", pXfrm->mod_name, MAX_LOGO );
       return -2;
    }
    if((str = k_str()) != NULL) 
      strcpy(pXfrm->xfrmParam->readInstName[ilogo], str);

    if((str = k_str()) != NULL) 
      strcpy(pXfrm->xfrmParam->readModName[ilogo], str);

    if((str = k_str()) != NULL)
    {
      if( strcmp(str,"TYPE_TRACEBUF" ) != 0 &&
          strcmp(str,"TYPE_TRACEBUF2") != 0    ) 
      {
         logit( "e","%s: GetWavesFrom: invalid msg type: %s "
                    "(must be TYPE_TRACEBUF or TYPE_TRACEBUF2); "
                    "exiting!\n", pXfrm->mod_name, str );
         return -2;
      }
      strcpy(pXfrm->xfrmParam->readTypeName[ilogo], str);
    }
    pXfrm->xfrmParam->nlogo++;
    break;
  default:
    return ProcessXfrmCommand( cmd_id, com );
  }
  return cmd_id;
}

/*      Function: BaseReadXfrmConfig                                            */
static int BaseReadXfrmConfig (char *configfile, XFRMWORLD* pXfrm )
{
  char     		*init;
  /* init flags, one byte for each required command */
  int      		nmiss;
  /* number of required commands that were missed   */
  char    		*com;
  char    		*str;
  char          cnull = '\0';         
  int      		nfiles;
  int      		success;
  int      		i;
  int           ret;

  BaseCmd = calloc( sizeof(char*), CmdCount );
  ParamTarget = calloc( sizeof(void*), CmdCount );
  init = malloc( CmdCount );
  if ( BaseCmd==NULL || ParamTarget==NULL || init==NULL) 
  {
    logit ("e",
             "%s: Allocation failure reading config file %s; exiting!\n", 
             pXfrm->mod_name, configfile);
    return EW_FAILURE;
  }
  BaseCmd[0] = "RSMyModId";
  ParamTarget[0] = &(pXfrm->xfrmParam->myModName);
  BaseCmd[1] = "RSInRing";
  ParamTarget[1] = &(pXfrm->xfrmParam->ringIn);
  BaseCmd[2] = "RSOutRing";
  ParamTarget[2] = &(pXfrm->xfrmParam->ringOut);
  BaseCmd[3] = "RIHeartBeatInterval";
  ParamTarget[3] = &(pXfrm->xfrmParam->heartbeatInt);
  BaseCmd[4] = "RSLogFile";
  ParamTarget[4] = &(pXfrm->xfrmParam->logSwitch);
  BaseCmd[5] = "RVMaxGap";
  ParamTarget[5] = &(pXfrm->xfrmParam->maxGap);
  BaseCmd[6] = "OFTestMode";
  ParamTarget[6] = &(pXfrm->xfrmParam->testMode);
  BaseCmd[7] = "OFDebug";
  ParamTarget[7] = &(pXfrm->xfrmParam->debug);
  BaseCmd[8] = "RXGetSCNL";
  BaseCmd[9] = "OFProcessRejected";
  ParamTarget[9] = &(pXfrm->xfrmParam->processRejected);
  BaseCmd[10] = "RXGetWavesFrom";

  for ( i=NUM_BASE_CMDS; i<CmdCount; i++ )
  {
    BaseCmd[i] = ModCmds[i-NUM_BASE_CMDS];
    ParamTarget[i] = ModParamTargets[i-NUM_BASE_CMDS];
  }
  
  /* Set to zero one init flag for each required command */
  for (i = 0; i < CmdCount; i++) 
  {
    init[i] = 0;
  }

  /* Open the main configuration file 
   **********************************/
  nfiles = k_open (configfile); 
  if (nfiles == 0) 
  {
    logit ("e",
             "%s: Error opening command file <%s>; exiting!\n", 
             pXfrm->mod_name, configfile);
    return EW_FAILURE;
  }

  /* Process all command files
   ***************************/
  while (nfiles > 0)   /* While there are command files open */
  {
    while (k_rd ())        /* Read next line from active file  */
    {  
      com = k_str ();         /* Get the first token from line */

    /* Ignore blank lines & comments
     *******************************/
      if (!com) continue;
      if (com[0] == '#') continue;

      /* Open a nested configuration file */
      if (com[0] == '@') 
      {
        success = nfiles + 1;
        nfiles  = k_open (&com[1]);
        if (nfiles != success) 
        {
          logit ("e", 
                   "%s: Error opening command file <%s>; exiting!\n", 
                   pXfrm->mod_name, &com[1]);
          return EW_FAILURE;
        }
        continue;
      }
    
      if ( strcmp( com, "GetSCN" ) == 0 ) 
      {
        com = BaseCmd[i]+2;
      }
      
      ret = -1;
      for ( i=0; i<CmdCount; i++ )
      {
        if ( strcmp( com, BaseCmd[i]+2 ) == 0 ) /*(k_its (BaseCmd[i]+2)) */
        {
          char type = BaseCmd[i][1];
          char kind = BaseCmd[i][0];
          switch ( type ) {
          case 'S': /* String */
            if ((str = k_str ()) != NULL )
            {
              strcpy( ParamTarget[i], str );
            } else {
              kind = 0;
            }
            break;
          case 'I': /* Integer */
            *((int*)(ParamTarget[i])) = k_long();
            break;
          case 'V': /* Value ? */
            *((double*)(ParamTarget[i])) = k_val();
            break;
          case 'F': /* Flag */
            *((int*)(ParamTarget[i])) = 1;
          default: /* Unknown; user-defined? */
            ret = BaseProcessXfrmCommand( pXfrm, i, com );
            if ( ret >= 0 )
              i = ret;
            break;
          }
          break;
        }
      }
      if ( i >= CmdCount )
      {
        ret = BaseProcessXfrmCommand( pXfrm, -1, com );
      } else {
        ret = i;
      }
      if ( ret >= 0 )
      { 
        if ( pXfrm->xfrmParam->debug )
          logit("","%s: command '%s' accepted\n",pXfrm->mod_name, com);
        init[ret] = 1;
      } else if ( ret == -1 ) {
        logit ("e", "%s: <%s> Unknown command in <%s>.\n", 
               pXfrm->mod_name, com, configfile);
        continue;
      } else if ( ret < -1 ) {
          logit ("e", 
                   "%s: Bad <%s> command in <%s>; exiting!\n",
                   pXfrm->mod_name, com, configfile);
          return EW_FAILURE;
      }
      
   /* See if there were any errors processing the command */
      if (k_err ()) 
      {
        logit ("e", 
                 "%s: Bad <%s> command in <%s>; exiting!\n",
                 pXfrm->mod_name, com, configfile);
        return EW_FAILURE;
      }

    } /** end while k_rd() **/

    nfiles = k_close ();

  } /** end while nfiles **/
  
/* After all files are closed, check init flags for missed commands */
  nmiss = 0;
  for(i = 0; i < CmdCount; i++)  if(BaseCmd[i][0]=='R' && !init[i])  nmiss++;

  if (nmiss) 
  {
    logit ("e", "%s: ERROR, no ", pXfrm->mod_name);
    for(i = 0; i < CmdCount; i++) if (!init[i]) logit("e", "<%s> ", BaseCmd[i]+2 );

    logit ("e", "command%s in <%s>; exiting!\n", (nmiss==1 ? "" : "s"), configfile);
    return EW_FAILURE;
  }

  ret = ReadXfrmConfig( init );
  free( init );

  return ret;
}

static int BaseReadXfrmEWH( XFRMWORLD *pXfrm )
{
  int i;

  if ( GetLocalInst( &(pXfrm->xfrmEWH->myInstId)) != 0 )
  {
    logit("e", "%s: Error getting myInstId for %s.\n", 
          pXfrm->mod_name, pXfrm->xfrmEWH->myInstId);
    return EW_FAILURE;
  }
  
  if ( GetModId( pXfrm->xfrmParam->myModName, &(pXfrm->xfrmEWH->myModId)) != 0 )
  {
    logit("e", "%s: Error getting myModId for %s.\n",
           pXfrm->mod_name, pXfrm->xfrmParam->myModName );
    return EW_FAILURE;
  }

  for (i=0; i<pXfrm->xfrmParam->nlogo; i++ ) {
    if ( GetInst( pXfrm->xfrmParam->readInstName[i], &(pXfrm->xfrmEWH->readInstId[i])) != 0)
    {
      logit("e", "%s: Error getting readInstId for %s.\n",
             pXfrm->mod_name, pXfrm->xfrmParam->readInstName[i] );
      return EW_FAILURE;
    }
  
    if ( GetModId( pXfrm->xfrmParam->readModName[i], &(pXfrm->xfrmEWH->readModId[i])) != 0 )
    {
      logit("e", "%s: Error getting readModName for %s.\n",
             pXfrm->mod_name, pXfrm->xfrmParam->readModName[i] );
      return EW_FAILURE;
    }

    if ( GetType( pXfrm->xfrmParam->readTypeName[i], &(pXfrm->xfrmEWH->readMsgType[i])) != 0 )
    {
      logit("e", "%s: Error getting readMsgType for %s.\n",
             pXfrm->mod_name, pXfrm->xfrmParam->readTypeName[i] );
      return EW_FAILURE;
    }
  }

  /* Look up keys to shared memory regions */
  if ((pXfrm->xfrmEWH->ringInKey = GetKey (pXfrm->xfrmParam->ringIn)) == -1) 
  {
    logit("e", "%s:  Invalid input ring name <%s>; exiting!\n", 
           pXfrm->mod_name, pXfrm->xfrmParam->ringIn);
    return EW_FAILURE;
  }

  if ((pXfrm->xfrmEWH->ringOutKey = GetKey (pXfrm->xfrmParam->ringOut) ) == -1) 
  {
    logit("e", "%s:  Invalid output ring name <%s>; exiting!\n", 
          pXfrm->mod_name, pXfrm->xfrmParam->ringOut);
    return EW_FAILURE;
  }

  /* Look up message types of interest */
  if (GetType ("TYPE_HEARTBEAT", &(pXfrm->xfrmEWH->typeHeartbeat)) != 0) 
  {
    logit("e", "%s: Invalid message type <TYPE_HEARTBEAT>; exiting!\n", 
            pXfrm->mod_name);
    return EW_FAILURE;
  }

  if (GetType ("TYPE_ERROR", &(pXfrm->xfrmEWH->typeError)) != 0) 
  {
    logit("e", "%s: Invalid message type <TYPE_ERROR>; exiting!\n", 
            pXfrm->mod_name);
    return EW_FAILURE;
  } 

  if (GetType ("TYPE_TRACEBUF", &(pXfrm->xfrmEWH->typeTrace)) != 0) 
  {
    logit("e", "decimate: Invalid message type <TYPE_TRACEBUF>; exiting!\n");
    return EW_FAILURE;
  }

  if (GetType ("TYPE_TRACEBUF2", &(pXfrm->xfrmEWH->typeTrace2)) != 0) 
  {
    logit("e", "decimate: Invalid message type <TYPE_TRACEBUF2>; exiting!\n");
    return EW_FAILURE;
  }

  return ReadXfrmEWH();

} 

static int BaseConfigureXfrm(XFRMWORLD *pXfrm, char **argv)
{
  /* Set initial values of WORLD structure */
  InitializeParameters( pXfrm );

  /* Initialize name of log-file & open it  
   ***************************************/
  logit_init (argv[1], 0, MAXMESSAGELEN, 1);

  /* Read config file and configure the decimator */
  if (BaseReadXfrmConfig(argv[1], pXfrm) != EW_SUCCESS)
  {
    logit("e", "%s: failed reading config file <%s>\n", pXfrm->mod_name, argv[1]);
    return EW_FAILURE;
  }
  logit ("" , "Read command file <%s>\n", argv[1]);

  /* Look up important info from earthworm.h tables
   ************************************************/
  if (BaseReadXfrmEWH(pXfrm) != EW_SUCCESS)
  {
    logit("e", "%s: Call to ReadEWH failed \n", pXfrm->mod_name );
    return EW_FAILURE;
  }

  /* Reinitialize logit to the desired logging level 
   ***********************************************/
  logit_init (argv[1], 0, MAXMESSAGELEN, pXfrm->xfrmParam->logSwitch);
  
  /* Get our process ID
   **********************/
  if ((pXfrm->MyPid = getpid ()) == -1)
  {
    logit ("e", "%s: Call to getpid failed. Exiting.\n", pXfrm->mod_name);
    return (EW_FAILURE);
  }

  return ConfigureXfrm();
}

int matchXfrmSCNL( TracePacket* pPkt, unsigned char msgtype, XFRMWORLD* pXfrm )
{
  int   i;
  
/* Look for match in TYPE_TRACEBUF2 packet 
 *****************************************/
  if( msgtype == pXfrm->xfrmEWH->typeTrace2 ) 
  {
     if( (pPkt->trh2.sta  == NULL) || 
         (pPkt->trh2.chan == NULL) || 
         (pPkt->trh2.net  == NULL) || 
         (pPkt->trh2.loc  == NULL)    )
     {
       logit ("et",  "%s: invalid (NULL) parameters to matchSCNL\n", pXfrm->mod_name);
       return( -2 );
     }
  
     for(i = 0; i < pXfrm->nSCNL; i++ )
     {
     /* try to match explicitly */
       if ((strcmp(pPkt->trh2.sta,  pXfrm->scnls[i].inSta)  == 0) &&
           (strcmp(pPkt->trh2.chan, pXfrm->scnls[i].inChan) == 0) &&
           (strcmp(pPkt->trh2.net,  pXfrm->scnls[i].inNet)  == 0) &&
           (strcmp(pPkt->trh2.loc,  pXfrm->scnls[i].inLoc)  == 0)    )
         return( i );
     }
     return( -1 );  /* no match in SCNL for TYPE_TRACEBUF2 */
  }

/* Look for match in TYPE_TRACEBUF packet
 ****************************************/
  else if( msgtype == pXfrm->xfrmEWH->typeTrace )
  {
     if( (pPkt->trh.sta  == NULL) || 
         (pPkt->trh.chan == NULL) || 
         (pPkt->trh.net  == NULL)    )
     {
       logit ("et",  "%s: invalid (NULL) parameters to matchSCNL\n", pXfrm->mod_name);
       return( -2 );
     }
  
     for(i = 0; i < pXfrm->nSCNL; i++ )
     {
     /* try to match explicitly */
       if ((strcmp(pPkt->trh.sta,   pXfrm->scnls[i].inSta)  == 0) &&
           (strcmp(pPkt->trh.chan,  pXfrm->scnls[i].inChan) == 0) &&
           (strcmp(pPkt->trh.net,   pXfrm->scnls[i].inNet)  == 0) &&
           (strcmp(LOC_NULL_STRING, pXfrm->scnls[i].inLoc)  == 0)    )
         return( i );
     }
     return( -1 );  /* no match in SCN for TYPE_TRACEBUF */
  }

/* Unknown Message Type
 **********************/
  if ( pXfrm->xfrmParam->debug ) 
    logit("", "%s: Unknown message type %d\n", pXfrm->mod_name, msgtype);
  
  return( -2 );
}


/*      Function: SubtusReport                                          */
void StatusReport( XFRMWORLD *pXfrm, unsigned char type, short code, 
                   char* message )
{
  char          outMsg[MAXMESSAGELEN];  /* The outgoing message.        */
  time_t        msgTime;        /* Time of the message.                 */

  /*  Get the time of the message                                     */
  time( &msgTime );

  /*  Build & process the message based on the type                   */
    if ( pXfrm->xfrmEWH->typeHeartbeat == type )
    {
      sprintf( outMsg, "%ld %ld\n\0", (long) msgTime, (long) pXfrm->MyPid );
      
      /*Write the message to the output region                          */
      if ( tport_putmsg( &(pXfrm->regionOut), &(pXfrm->hrtLogo), 
                         (long) strlen( outMsg ), outMsg ) != PUT_OK )
      {
        /*     Log an error message                                    */
        logit( "et", "%s: Failed to send a heartbeat message (%d).\n",
               pXfrm->mod_name, code );
      }
    }
    else
    {
      if ( message ) {
        sprintf( outMsg, "%ld %hd %s\n\0", (long) msgTime, code, message );
        logit("t","Error:%d (%s)\n", code, message );
      }
      else {
        sprintf( outMsg, "%ld %hd\n\0", (long) msgTime, code );
        logit("t","Error:%d (No description)\n", code );
      }

      /*Write the message to the output region                         */
      if ( tport_putmsg( &(pXfrm->regionOut), &(pXfrm->errLogo), 
                         (long) strlen( outMsg ), outMsg ) != PUT_OK )
      {
        /*     Log an error message                                    */
        logit( "et", "%s: Failed to send an error message (%d).\n",
               pXfrm->mod_name, code );
      }
      
    }
}

/*      Function: XfrmThread                                            */
thr_ret BaseXfrmThread (void* args)
{
  XFRMWORLD     *World;
  int            ret;
  int            jSta;
  MSG_LOGO       reclogo;       /* logo of retrieved message     */
  char          *InBuf;         /* string to hold input message  */
  long           InBufLen;      /* length of InBuf               */
  TracePacket   *WaveBuf;       /* pointer to raw trace message  */
  TracePacket   *OutBuf;        /* pointer to debiased trace message  */

  World = ( XFRMWORLD *) args;
  
  /* Allocate the waveform buffer */
  InBufLen = (MAX_TRACEBUF_SIZ + sizeof (TP_PREFIX));
  InBuf    = (char *) malloc ((size_t) InBufLen);

  if (InBuf == NULL)
  {
    logit ("e", "%s: Cannot allocate input buffer\n", World->mod_name);
    World->XfrmStatus = -1;
    KillSelfThread();
  } else {
    WaveBuf  = (TracePacket *)(InBuf + sizeof(TP_PREFIX));
  }

  OutBuf    = (TracePacket *) malloc ((size_t) InBufLen);
  if (OutBuf == NULL)
  {
    logit ("e", "%s: Cannot allocate output buffer\n", World->mod_name);
    World->XfrmStatus = -1;
    KillSelfThread();
  }

  /* Tell the main thread we're feeling ok */
  World->XfrmStatus = 0;

  while (1)
  {
    /* Get top message from the MsgQueue */
    RequestMutex ();
    ret = dequeue (&(World->MsgQueue), InBuf, &InBufLen, &reclogo);
    ReleaseMutex_ew ();
    
    if (ret < 0)
    {                                 /* empty queue */
      sleep_ew (500);
      continue;
    }

    if ( World->completionFcn != NULL && InBufLen==0 )
    {
      logit("","%s: thread knows to terminate\n", World->mod_name);
      World->completionFcn( 1 );
      return;
    }
    
    /* Extract the SCNL number; recall, it was pasted as an int on the front 
     * of the message by the main thread */
    jSta = *((int*) InBuf);
    
    if (FilterXfrm( World, WaveBuf, jSta, reclogo.type, OutBuf ) ==
        EW_FAILURE)
    {
      logit("et", "%s: error from FilterXfrm; exiting\n", World->mod_name);
      World->XfrmStatus = -1;
      KillSelfThread();
    }

  } /* while (1) - message dequeuing process */

}

int BaseXfrmResetSCNL( XFRMSCNL *pSCNL, TracePacket *inBuf, XFRMWORLD* pWorld )
{  
  pSCNL->inEndtime = 0.0;    /* used to identify gaps */
  
  /* Save params that mustn't change over span of packets */
  strcpy( pSCNL->datatype, inBuf->trh2.datatype );
  pSCNL->samprate = inBuf->trh2.samprate;
  
  return XfrmResetSCNL( pSCNL, inBuf, pWorld );
}

int BaseFilterXfrm (XFRMWORLD* pDb, TracePacket *inBuf, int jSta, 
                    unsigned char msgtype, TracePacket *outBuf)
{
  XFRMSCNL *pSCNL;
  int ret, datasize;
  
  pSCNL = &(pDb->scnls[jSta]);
  
  /* Create SCNL info field if need be */
  if ( pSCNL->status < 0 ) 
  {
    ret = XfrmInitSCNL( pDb, pSCNL );  
    if ( ret != EW_SUCCESS )
    {
      return ret;
    }
    pSCNL->status = 0;
  }
  else if ( pSCNL->status )
  {
    return EW_WARNING;
  }

  /* If it's tracebuf, make it look like a tracebuf2 message */
  if( msgtype == pDb->xfrmEWH->typeTrace ) TrHeadConv( &(inBuf->trh) );
  
  if (pDb->xfrmParam->debug)
    logit("", "%s: enter filter with <%s.%s.%s.%s> start: %lf\n",
    	  pDb->mod_name,
          inBuf->trh2.sta, inBuf->trh2.chan, inBuf->trh2.net, inBuf->trh2.loc,
          inBuf->trh2.starttime );
  
  /* Check for useable data types: we only handle shorts and longs for now */
  if ( (inBuf->trh2.datatype[0] != 's' && inBuf->trh2.datatype[0] != 'i') ||
        (inBuf->trh2.datatype[1] != '2' && inBuf->trh2.datatype[1] != '4') )
  {
    logit("et","%s: unusable datatype <%s> from <%s.%s.%s.%s>; skipping\n",
          pDb->mod_name,
          inBuf->trh2.datatype, inBuf->trh2.sta, inBuf->trh2.chan, 
          inBuf->trh2.net, inBuf->trh2.loc);
    return EW_WARNING;
  }
  else
  {
    datasize = inBuf->trh2.datatype[1]-'0';
  }

  /* If we have previous data, check for data gap */
  if ( pSCNL->inEndtime != 0.0 )
  {
    if ( (inBuf->trh2.starttime - pSCNL->inEndtime) * inBuf->trh2.samprate > 
         pDb->xfrmParam->maxGap )
    {
      logit("et","%s: gap in data for <%s.%s.%s.%s>:\n"
            "\tlast end: %lf this start: %lf; resetting\n",
            pDb->mod_name,
            pSCNL->inSta, pSCNL->inChan, pSCNL->inNet, pSCNL->inLoc, 
            pSCNL->inEndtime, inBuf->trh2.starttime);
      if ( BaseXfrmResetSCNL( (XFRMSCNL*)pSCNL, inBuf, (XFRMWORLD*)pDb ) != EW_SUCCESS )
      {
        pSCNL->status = 1;
        return EW_FAILURE;
      }
    }
    else if (inBuf->trh2.starttime < pSCNL->inEndtime)
    {
      logit("et","%s: overlapping times for <%s.%s.%s.%s>:\n"
            "\tlast end: %lf this start: %lf; resetting\n",
            pDb->mod_name,
            pSCNL->inSta, pSCNL->inChan, pSCNL->inNet, pSCNL->inLoc, 
            pSCNL->inEndtime, inBuf->trh2.starttime);
      if ( BaseXfrmResetSCNL( (XFRMSCNL*)pSCNL, inBuf, (XFRMWORLD*)pDb ) != EW_SUCCESS )
      {
        pSCNL->status = 1;
        return EW_FAILURE;
      }
    }
    else if ( inBuf->trh2.samprate != pSCNL->samprate ) 
    {
      logit("et","%s: sample rate changed in <%s.%s.%s.%s> at %lf from %lf to %lf; exiting.\n",
            pDb->mod_name,
            pSCNL->inSta, pSCNL->inChan, pSCNL->inNet, pSCNL->inLoc, inBuf->trh2.starttime,
            pSCNL->samprate, inBuf->trh2.samprate);
      pSCNL->status = 1;
      return EW_FAILURE;
    }
    else if ( inBuf->trh2.datatype[1] != pSCNL->datatype[1] ) 
    {
      logit("et","%s: datatype changed in <%s.%s.%s.%s> at %lf from %s to %s; exiting.\n",
            pDb->mod_name,
            pSCNL->inSta, pSCNL->inChan, pSCNL->inNet, pSCNL->inLoc, inBuf->trh2.starttime,
            pSCNL->datatype, inBuf->trh2.datatype);
      pSCNL->status = 1;
      return EW_FAILURE;
    }

  }
  else
  {
    BaseXfrmResetSCNL( (XFRMSCNL*)pSCNL, inBuf, (XFRMWORLD*)pDb );
  }
  
  pSCNL->inEndtime = inBuf->trh2.endtime;
    
  return EW_SUCCESS;
}

int XfrmWritePacket( XFRMWORLD* pDb, XFRMSCNL* pSCNL, unsigned char msgtype, 
	TracePacket* outBuf, int outBufLen )
{
  pDb->trcLogo.type = msgtype;
  outBuf->trh2.version[0] = TRACE2_VERSION0;
  outBuf->trh2.version[1] = TRACE2_VERSION1;
  strcpy( outBuf->trh2.sta, pSCNL->outSta );
  strcpy( outBuf->trh2.net, pSCNL->outNet );
  strcpy( outBuf->trh2.chan, pSCNL->outChan );
  strcpy( outBuf->trh2.loc, pSCNL->outLoc );
  if (tport_putmsg (&(pDb->regionOut), &(pDb->trcLogo), outBufLen, 
                    outBuf->msg) != PUT_OK)
  {
    logit ("et","%s: Error sending type:%d message.\n", pDb->mod_name, msgtype );
    return EW_FAILURE;
  }
    
  return EW_SUCCESS;
}

int main (int argc, char **argv) 
{
  XFRMWORLD *World;                /* Our main data structure              */
  time_t    timeNow;               /* current time                         */ 
  time_t    timeLastBeat;          /* time last heartbeat was sent         */
  long      sizeMsg;               /* size of retrieved message            */
  SHM_INFO  regionIn;              /* Input shared memory region info.     */
  MSG_LOGO  logoWave[MAX_LOGO];    /* Logo(s) of requested waveforms.      */
  int       ilogo;                 /* working counter                      */
  MSG_LOGO  logoMsg;               /* logo of retrieved message            */
  unsigned  tidXfrmor;             /* Transformer thread id                */
  char     *inBuf;                 /* Pointer to the input message buffer. */
  int       inBufLen;              /* Size of input message buffer         */
  TracePacket  *TracePkt;
  int       ret;
  char      msgText[MAXMESSAGELEN];/* string for log/error messages    */
  int i;

  /* Allocate a WORLD
   *****************************************************/
  CmdCount = NUM_BASE_CMDS;
  SetupXfrm( &World, &ModCmds, &CmdCount, &ModParamTargets );
  World->scnls = NULL;
  World->nSCNL = 0;

  /* Check command line arguments 
   ******************************/
  if (argc != 2)
  {
    fprintf (stderr, "Usage: %s <configfile>\n", World->mod_name);
    FreeXfrmWorld( World );
    exit (EW_FAILURE);
  }
  
  /* Read config file and configure the decimator */
  if (BaseConfigureXfrm(World, argv) != EW_SUCCESS)
  {
    logit("e", "%s: configure() failed \n", World->mod_name);
    FreeXfrmWorld( World );
    exit (EW_FAILURE);
  }

  if ( World->xfrmParam->testMode )
  {
    logit("e", "%s terminating normally for test mode\n", World->mod_name);
    FreeXfrmWorld( World );
    exit (EW_SUCCESS);
  }
  
  /* We will put the SCNL index in front of the trace message, so we  *
   * don't have to look up the SCNL again at the other end of the queue. */
  inBufLen = MAX_TRACEBUF_SIZ + sizeof( double );
  if ( ! ( inBuf = (char *) malloc( (size_t) inBufLen ) ) )
  {
    logit( "e", "%s: Memory allocation failed - initial message buffer!\n",
        argv[0] );
    FreeXfrmWorld( World );
    exit( EW_FAILURE );
  }
  TracePkt = (TracePacket *) (inBuf + sizeof(TP_PREFIX));
  
  /* Attach to Input shared memory ring 
   ************************************/

  tport_attach (&regionIn, World->xfrmEWH->ringInKey);
  if (World->xfrmParam->debug) {
    logit ("", "%s: Attached to public memory region %s: %d\n", World->mod_name,
           World->xfrmParam->ringIn, World->xfrmEWH->ringInKey);
  }

  /* Attach to Output shared memory ring 
   *************************************/
  if (World->xfrmEWH->ringOutKey == World->xfrmEWH->ringInKey) {
    World->regionOut = regionIn;
  } else {
    tport_attach (&(World->regionOut), World->xfrmEWH->ringOutKey);
    if (World->xfrmParam->debug)
      logit ("", "%s: Attached to public memory region %s: %d\n", World->mod_name,
             World->xfrmParam->ringOut, World->xfrmEWH->ringOutKey);
  }

 /* Specify logos of incoming waveforms 
  *************************************/
  for( ilogo=0; ilogo<World->xfrmParam->nlogo; ilogo++ ) {
    logoWave[ilogo].instid = World->xfrmEWH->readInstId[ilogo];
    logoWave[ilogo].mod    = World->xfrmEWH->readModId[ilogo];
    logoWave[ilogo].type   = World->xfrmEWH->readMsgType[ilogo];
    if ( World->xfrmParam->debug ) 
      logit("", "%s: Logo[%d] = %d,%d,%d\n", World->mod_name, ilogo,
          logoWave[ilogo].instid, logoWave[ilogo].mod, logoWave[ilogo].type); 
  }

  /* Specify logos of outgoing messages 
   ************************************/
  World->hrtLogo.instid = World->xfrmEWH->myInstId;
  World->hrtLogo.mod    = World->xfrmEWH->myModId;
  World->hrtLogo.type   = World->xfrmEWH->typeHeartbeat;
  if ( World->xfrmParam->debug ) 
    logit("", "%s: hrtLogo = %d,%d,%d\n", World->mod_name,
        World->hrtLogo.instid, World->hrtLogo.mod, World->hrtLogo.type); 

  World->errLogo.instid = World->xfrmEWH->myInstId;
  World->errLogo.mod    = World->xfrmEWH->myModId;
  World->errLogo.type   = World->xfrmEWH->typeError;
  if ( World->xfrmParam->debug ) 
    logit("", "%s: errLogo = %d,%d,%d\n", World->mod_name,
        World->errLogo.instid, World->errLogo.mod, World->errLogo.type); 
  
  SpecifyXfrmLogos();

  /* Force a heartbeat to be issued in first pass thru main loop  */
  timeLastBeat = time (&timeNow) - World->xfrmParam->heartbeatInt - 1;

  /* Flush the incoming transport ring */
  while (tport_getmsg (&regionIn, logoWave, (short)World->xfrmParam->nlogo, &logoMsg,
                       &sizeMsg, inBuf, inBufLen) != GET_NONE);

  /* Create MsgQueue mutex */
  CreateMutex_ew();

  /* Allocate the message Queue */
  initqueue (&(World->MsgQueue), QUEUE_SIZE, inBufLen);
  

  /* Start decimator thread which will read messages from   *
   * the Queue, decimate them and write them to the OutRing */
  if (StartThreadWithArg (XfrmThread, (void *) World, (unsigned) 
                          THREAD_STACK, &tidXfrmor) == -1)
  {
    logit( "e", 
           "%s: Error starting thread.  Exiting.\n", World->mod_name);
    tport_detach (&regionIn);
    
    if (World->xfrmEWH->ringOutKey != World->xfrmEWH->ringInKey) {
       tport_detach (&(World->regionOut)); 
    }
    free( inBuf );
    FreeXfrmWorld();
    exit( EW_FAILURE );
  }

  World->XfrmStatus = 0; /*assume the best*/

  if ( World->completionFcn != NULL )
  {
    ret = World->completionFcn( -1 );
    if ( ret != EW_SUCCESS )
    {
      logit("e", "%s: Problem with completionFcn(-1)\n", World->mod_name, ret );
    }
  }

/*--------------------- setup done; start main loop -------------------------*/
  if ( World->xfrmParam->debug )
    logit("","%s: starting main loop\n", World->mod_name);
  while (tport_getflag (&regionIn) != TERMINATE  &&
         tport_getflag (&regionIn) != World->MyPid )
  {
    /* send module's heartbeat */
    if (time (&timeNow) - timeLastBeat >= World->xfrmParam->heartbeatInt) 
    {
      timeLastBeat = timeNow;
      StatusReport (World, World->xfrmEWH->typeHeartbeat, 0, ""); 
    }

    if (World->XfrmStatus < 0)
    {
      logit ("et", 
             "%s: thread died. Exiting\n", World->mod_name);
      exit (EW_FAILURE);
    }

    ret = tport_getmsg (&regionIn, logoWave, (short)World->xfrmParam->nlogo, 
                        &logoMsg, &sizeMsg, TracePkt->msg, MAX_TRACEBUF_SIZ);

    /* Check return code; report errors */
    if (ret != GET_OK)
    {
      if (ret == GET_TOOBIG)
      {
        sprintf (msgText, "msg[%ld] i%d m%d t%d too long for target",
                 sizeMsg, (int) logoMsg.instid,
                 (int) logoMsg.mod, (int)logoMsg.type);
        StatusReport (World, World->xfrmEWH->typeError, ERR_TOOBIG, msgText);
        continue;
      }
      else if (ret == GET_MISS)
      {
        sprintf (msgText, "missed msg(s) i%d m%d t%d in %s",
                 (int) logoMsg.instid, (int) logoMsg.mod, 
                 (int)logoMsg.type, World->xfrmParam->ringIn);
        StatusReport (World, World->xfrmEWH->typeError, ERR_MISSMSG, msgText);
      }
      else if (ret == GET_NOTRACK)
      {
        sprintf (msgText, "no tracking for logo i%d m%d t%d in %s",
                 (int) logoMsg.instid, (int) logoMsg.mod, 
                 (int)logoMsg.type, World->xfrmParam->ringIn);
        StatusReport (World, World->xfrmEWH->typeError, ERR_NOTRACK, msgText);
      }
      else if (ret == GET_NONE)
      {
        sleep_ew(500);
        continue;
      }
      else if (World->xfrmParam->debug)
        logit( "", "%s: Unexpected return from tport_getsg: %d\n", 
            World->mod_name, ret );
    }

    if ((ret = matchXfrmSCNL( TracePkt, logoMsg.type, World )) < -1 )
    {
      logit ("et", "%s: Call to matchSCNL failed with %d; exiting.\n", 
        World->mod_name, ret);
      FreeXfrmWorld();
      exit (EW_FAILURE);
    }
    else if ( ret == -1 )
    {
      /* Not an SCNL we want */
      if ( World->xfrmParam->processRejected ) 
      {
        if ( ProcessXfrmRejected( TracePkt, logoMsg, inBuf ) == EW_FAILURE )
        {
          logit ("et","%s: Error processing rejected packet; type:%d message.\n", 
          	World->mod_name, logoMsg.type );
          FreeXfrmWorld();
          exit (EW_FAILURE);
        }
      }
      continue;
    }
    
    /* stick the SCNL number as an int at the front of the message */
    *((int*)inBuf) = ret; 

    /* If necessary, swap bytes in the wave message */
    if( logoMsg.type == World->xfrmEWH->typeTrace2 )
        ret = WaveMsg2MakeLocal( &(TracePkt->trh2) );
    else if( logoMsg.type == World->xfrmEWH->typeTrace )  
        ret = WaveMsgMakeLocal( &(TracePkt->trh) );

    ret = PreprocessXfrmBuffer( TracePkt, logoMsg, inBuf );

    if ( ret < -1 )
    {
      FreeXfrmWorld();
      exit (EW_FAILURE);
    }
    if ( ret == -1 )
    {
      continue;
    }

    /* Queue retrieved msg */
    RequestMutex ();
    ret = enqueue (&(World->MsgQueue), inBuf, sizeMsg + sizeof(TP_PREFIX), logoMsg); 
    ReleaseMutex_ew ();

    if (ret != 0)
    {
      if (ret == -1)
      {
        sprintf (msgText, 
                 "Message too large for queue; Lost message.");
        StatusReport (World, World->xfrmEWH->typeError, ERR_QUEUE, msgText);
        continue;
      }
      if (ret == -3) 
      {
        sprintf (msgText, "Queue full. Old messages lost.");
        StatusReport (World, World->xfrmEWH->typeError, ERR_QUEUE, msgText);
        continue;
      }
      logit("", "%s: Problem w/ queue: %d\n", World->mod_name, ret );
    } /* problem from enqueue */

  } /* wait until TERMINATE is raised  */  

  if ( World->completionFcn != NULL ) /* send empty packet to signal termination */
  {
    RequestMutex (); 
    ret = enqueue (&(World->MsgQueue), inBuf, 0, logoMsg); 
    ReleaseMutex_ew ();
    ret = World->completionFcn( 0 );
    if ( ret != EW_SUCCESS )
    {
      logit("e", "%s: Problem with completionFcn(0)\n", World->mod_name, ret );
    }
  }

  /* Termination has been requested */
  tport_detach (&regionIn);
  if (World->xfrmEWH->ringOutKey != World->xfrmEWH->ringInKey) {
    tport_detach (&(World->regionOut));
  }
  free( inBuf );
  FreeXfrmWorld();
  logit ("t", "Termination requested; exiting!\n" );
  exit (EW_SUCCESS);

}

