/*
 *   THIS FILE IS UNDER RCS - DO NOT MODIFY UNLESS YOU HAVE
 *   CHECKED IT OUT USING THE COMMAND CHECKOUT.
 *
 *    $Id: ora_trace_req.c,v 1.17 2002/06/07 16:51:48 patton Exp $
 *
 *    Revision history:
 *     $Log: ora_trace_req.c,v $
 *     Revision 1.17  2002/06/07 16:51:48  patton
 *     Made logit changes.
 *
 *     Revision 1.16  2002/04/16 20:57:07  davidk
 *     concierge Mark 2.
 *
 *     Revision 1.15  2002/02/12 04:32:40  davidk
 *     Upped the default internal queue size to 30.  Made it a
 *     configable parameter.
 *
 *     Now call parseSnippet() outside of the main loop, so
 *     as to get Event information (included with snippet info
 *     from parseSnippet() ), to process event data before
 *     main loop.
 *
 *     Now call parseSnippet() at the bottom of the main
 *     loop instead of within the while clause.
 *
 *     Revision 1.14  2001/05/15 02:15:35  davidk
 *     Moved functions around between the apps, DB API, and DB API INTERNAL
 *     levels.  Renamed functions and files.  Added support for amplitude
 *     magnitude types.  Reformatted makefiles.
 *
 *     Revision 1.12  2001/05/10 20:16:48  dietz
 *     Changed to shut down gracefully if the transport flag is
 *     set to TERMINATE or MyPid.
 *
 *     Revision 1.11  2001/02/28 17:29:10  lucky
 *     Massive schema redesign and cleanup.
 *
 *     Revision 1.10  2001/01/18 17:05:04  davidk
 *     New file ora_trace_req.c replaces ora_trace_save.c.  ora_trace_req
 *     parses trigger messages and stores the resulting snippet requests
 *     in the datbase, where another program later comes along and
 *     gets the requests and attempts to retrieve trace for them.  ora_trace_save
 *     previously attempted to parse the trig message and retrieve data
 *     for the resulting requests, which it stuffed(the data) into the DB.
 *
 *     Revision 1.3  2000/04/14 00:13:38  alex
 *     1. Failure to stuff a snippet now generates error message via dbtsv_status.
 *     2. Removed (hopefully) redundand logit() calls, since dbtsv_status does a logit.
 *     3. Gave the SnippetMaker thread its own error text buffer, so as to not conflict with
 *     the global "Text" buffer.
 *     Alex
 *
 *     Revision 1.2  2000/03/31 21:32:44  davidk
 *     changed SnippetMaker thread so that it sets its status to -1 before
 *     killing itself.  This way the main thread should restart when the
 *     SnippetMaker dies, instead of continuing to run happily.
 *
 *     Revision 1.1  2000/03/31 17:32:54  lucky
 *     Initial revision
 *
 *     Revision 1.6  2000/03/07 19:04:02  lucky
 *     Increased MAX_WAVESERVERS to 100, and removed an annoying debug statement
 *     that printed out every single channel being compared to
 *
 *     Revision 1.5  2000/01/20 22:21:28  alex
 *     alex: changed snippet structure .eventId from integer to string,
 *     so eventId is now a string.
 *
 *     Revision 1.3  2000/01/04 20:05:29  davidk
 *     converted code from ora_trace_save.c to ora_trace_save.c to working
 *     with schema2 and later API's.
 *
 *     Revision 1.2  1999/11/09 16:42:55  lucky
 *     *** empty log message ***
 *
 *     Revision 1.1  1999/11/09 16:37:20  lucky
 *     Initial revision
 *
 *
 */

/*

   ora_trace_req.c:  adapted from v2 of ora_trace_save.c to store snippet
   requests in DB.  Does not deal with wave_servers at all, just stores
   the requests in DB in hopes that some other concierge like program
   will come along later and convert those requests to actual trace data.

   ora_trace_save.c(rev2): adapted from ora_trace_save.c to use new schema2
   and beyond API's.

   ora_trace_save.c : Cloned from dbtrace_save, which was cloned from 
     dbreport (which was written by Lynn Dietz). reads trigger messages and
     writes specified trace snippets to an Oracle db. Different from
     dbtrace_save in that it uses the API, written by Dave Kragness.

Story: we pick up TYPE_TRIG messages which are generated by various
detector algorithms. It's a verbose ascii message which we contracted
from exposure to the Menlo volcano crowd. It lists an event id, event
time, and a number of lines, each listing a station by name, trigger
start time, and duration of data to be saved for that station.
Wildcards are permitted in either the station, network or component
field (suggestion made my Steve Malone).

We loop over the lines of the message. For each line we construct an
array of implied trace requests. These had better be than the
configuration parameter MaxTrace. We then loop over this array,
requesting and disposing of each trace, on at a time.  As we do so, we
might be making premature requests: the waves might not have reached
the sensors yet. The WaveServers are smart enough to tell us if that's
the case, and how long we have to wait. So we wait for the longest wait
period, and then run over the requests again, retrieving any
'stragglers'. After that, we give up.

*/


#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <earthworm.h>
#include <kom.h>
#include <transport.h>
#include <ws_clientII.h>
#include <mem_circ_queue.h>
#include <parse_trig.h>
#include <ewdb_ora_api.h>
#include <ewdb_apps_utils.h>
#include "ora_trace_req.h"

/* Defines 
 *********/
#define TRUE		1
#define FALSE		0

#define MAX_STR		 255
#define MAXLINE		1000
#define MAX_NAME_LEN	  50
#define MAXCOLS		 100
#define MAXTXT           150
#define MAX_WAVESERVERS   100
#define MAX_ADRLEN        20

/* Functions in this source file 
 *******************************/
void  dbtsv_config ( char * );
void  dbtsv_lookup ( void );
void  dbtsv_status ( unsigned char, short, char * );
int   dbtsv_writefile( TRACE_REQ *, long );
int match( SNIPPET* pSnppt, WS_PSCN pscn);
int duplicate( WS_PSCN pscn, TRACE_REQ* pTrReq, int nTraceReq);
thr_ret MessageStacker( void * );
thr_ret SnippetMaker( void * );


/* The message queue
 *******************/
QUEUE OutQueue;                 /* from queue.h, queue.c; sets up linked */

WS_MENU_QUEUE_REC MenuList;	/* the list of menues; the menues will be malloced! DANGER */

/* Thread things
 ***************/
#define THREAD_STACK 8192
static unsigned tidSnipper;      /* Message processing thread id */
static unsigned tidStacker;      /* Thread moving messages from transport */
                                 /*   to queue */
int MessageStackerStatus=0;      /* 0=> Stacker thread ok. <0 => dead */
int SnippetMakerStatus=0;        /* 0=> Snipper thread ok. <0 => dead */

/* Transport global variables
 ****************************/
static  SHM_INFO  Region;      /* shared memory region to use for i/o    */

#define   MAXLOGO   2
MSG_LOGO  GetLogo[MAXLOGO];    /* array for requesting module,type,instid */
short 	  nLogo;

        
/* Globals to set from configuration file
 ****************************************/
static int     RingSize;            /* max messages in output circular buffer       */
static char    RingName[20];        /* name of transport ring for i/o      */
static char    MyModName[20];       /* speak as this module name/id        */
static int     LogSwitch;           /* 0 if no logfile should be written   */
static int     HeartBeatInt;        /* seconds between heartbeats          */
int 	         MaxTraces;	    /* max traces per message we'll ever deal with  */
int   	       iDefaultRequestGroup = 0;/* the default "Request Group" used by this program  */
int   	       iNumAttempts = 0;/* the number of attempts this program will schedule a request for.  */

/* Things to look up in the earthworm.h tables with getutil.c functions
 **********************************************************************/
static long          RingKey;       /* key of transport ring for i/o      */
static unsigned char InstId;        /* local installation id              */
static unsigned char MyModId;       /* Module Id for this program         */
static unsigned char TypeHeartBeat; 
static unsigned char TypeError;
static unsigned char TypeTrig;
static unsigned char TypeH71Sum;

/* Error messages used by ora_trace_req  
 ***************************************/
#define  ERR_MISSMSG       0   /* message missed in transport ring       */
#define  ERR_TOOBIG        1   /* retreived msg too large for buffer     */
#define  ERR_NOTRACK       2   /* msg retreived; tracking limit exceeded */
#define  ERR_API_INIT_FAILED 3 /* ORA_API_INIT failed */
#define  ERR_XALLOC_ERROR  4   /* unable to dynamically allocate memory  */
#define  ERR_UNKNOWN       5   /* msg retreived; unkown error */
#define  ERR_QUEUE         6   /* error queueing message for sending     */
#define  ERR_MSG_STACKER   7   /* error with message stacker thread      */
#define  ERR_SNPT_MAKER    8   /* error with snippet maker thread        */
#define  ERR_WAVE_SERVER   9   /* error getting data from a wave_server  */
#define  ERR_SNIPPET2TRREQ 10  /* error in call to snippet2trReq         */
#define  ERR_ORA_STSNEVENT 11  /* error in call to DB ewdb_apps_putaway_StartSnippetEvent()*/
#define  ERR_STUFF_SNPT    13  /* can't stuff snippet into dbms          */

static char Text[MAXTXT];      /* string for log/error messages          */


/* Things to be malloc'd
 ***********************/
/* story: Some of these things don't have to be global; they are only
so that we can malloc them from main at startup time. Having them 
malloc'd by the routine which uses them presents the danger that the 
module would start ok, but die (if unable to malloc) only when the 
first event hits. Which could be much later, at a bad time...*/


/* Debug debug DEBUG */
 int Debug_tr_sv=0;	/* setting this to one causes all sorsts of log output */

pid_t MyPid;            /* Our own pid, sent with heartbeat for restart 
                           purposes */

main( int argc, char **argv )
{
  long         timeNow;          		/* current time                   */       
  long         timeLastBeat;     		/* time last heartbeat was sent   */
  int          dbconnected=0;                  /* =1 if connected to database, =0 if not */
  char         LogFileName[128];
  
  /* Check command line arguments 
  ******************************/
  if ( argc != 2 )
  {
    fprintf( stderr, "Usage: ora_trace_req  <configfile>\n" );
    exit( 0 );
  }

  /* Initialize name of log-file & open it 
  ***************************************/
  fprintf(stderr,"ora_trace_req :  Initializing logit.\n");
  /* Created LogFileName, to be obtained from config file name
     instead of using the fixed "ora_trace_req " for logname. */
  strcpy(LogFileName,argv[1]);
  LogFileName[strlen(argv[1])-2/*.d*/]=0;

  logit_init( LogFileName, 0, 1024, 1 );

  /* Read the configuration file(s)
  ********************************/
  fprintf(stderr,"ora_trace_req :  Reading Config files.\n");
  dbtsv_config( argv[1] );
  logit( "" , "ora_trace_req : Read command file <%s>\n", argv[1] );

  /* Get our own Pid for restart purposes
  ***************************************/
  MyPid = getpid();
  if(MyPid == -1)
  {
		/* DK 19990715 changed from logit to fprintf since logit_init
		   not yet called **************************/
    fprintf(stderr,"ora_trace_req : Cannot get pid. Exiting.\n");
    exit(0);
  }

  /* Look up important info from earthworm.h tables
  ************************************************/
  fprintf(stderr,"ora_trace_req :  Performing Earthworm config lookup.\n");
  dbtsv_lookup();
  
  /* Reinitialize logit to desired logging level
  **********************************************/
  logit_init( LogFileName, 0, 1024, LogSwitch );

	/* 19990715 DK  I relocated (he he!!!) the tport_attach() call,
	   so that status messages could be issued for errors earlier on
		 in the program *****************************/

  /* Attach to Input/Output shared memory ring 
  *******************************************/
  tport_attach( &Region, RingKey );

	/* Indicate whether the debugger flag is turned on. */
  logit("","Debug_tr_sv=%d\n",Debug_tr_sv);

	/* If debug mode, log our logo */
  if(Debug_tr_sv)
  {
    logit("", "ora_trace_req : Attached to public memory region %s: %d\n", 
          RingName, RingKey );
    logit("","logo:%d %u %u %u\n", nLogo,
          GetLogo[0].type,GetLogo[0].mod,GetLogo[0].instid);
  }

  

  /* Initialize the disposal system
  ********************************/
  if( ewdb_api_Init(DBuser,DBpassword,DBservice ) != EWDB_RETURN_SUCCESS )
  {
    sprintf(Text, "OraAPIInit() failure; exiting!\n" );
    dbtsv_status( TypeError, ERR_API_INIT_FAILED, Text ); 
    exit( -1 );
  }

  /* Force a heartbeat to be issued in first pass thru main loop
  *************************************************************/
  timeLastBeat = time(&timeNow) - HeartBeatInt - 1;
  
  /* Create a Mutex to control access to queue
  ********************************************/
  CreateMutex_ew();  /* a single unnamed mutex? */

  /* Initialize the message queue
  *******************************/
  RingSize = 30;  /* that is, we can stack up to 30 trigger messages */

  if(initqueue(&OutQueue,(unsigned long)RingSize,
		           (unsigned long)MAX_BYTES_PER_EQ))
	{
    logit( "", "Initqueue failure; exiting!\n" );
    sprintf(Text, "Initqueue failure; exiting!\n" );
    dbtsv_status( TypeError, ERR_QUEUE, Text ); 
    exit( -1 );
	}
     
  /* Start the  message stacking thread
  *************************************/
  if ( StartThread(MessageStacker, (unsigned)THREAD_STACK, &tidStacker ) == -1)
  {
    logit( "e", "ora_trace_req : Error starting  MessageStacker thread. "
           "Exiting.\n" );
    sprintf(Text, "ora_trace_req : Error starting  MessageStacker thread. "
           "Exiting.\n" );
    dbtsv_status( TypeError, ERR_MSG_STACKER, Text ); 
    tport_detach( &Region );
    exit (-1);
  }

  MessageStackerStatus=0; /*assume the best*/

  /* Start the  snippet maker thread
  **********************************/
  if ( StartThread(  SnippetMaker, (unsigned)THREAD_STACK, &tidSnipper ) == -1 )
  {
    logit( "e", "ora_trace_req : Error starting  MessageStacker thread. "
                "Exiting.\n" );
    sprintf(Text, "ora_trace_req : Error starting  MessageStacker thread. "
                "Exiting.\n" );
    dbtsv_status( TypeError, ERR_SNPT_MAKER, Text ); 
    tport_detach( &Region );
    exit (-1);
    /* shouldn't there be a status issued here at exit!?!! DK11/1998*/
  }
  SnippetMakerStatus=0; /*assume the best*/

  /* Setup done; start main loop: 
  ******************************/
  /* Having delegated message collecting, and message processing, there's
  not much for us left to do: watch thread status, and beat the heart */

  while( tport_getflag(&Region) != TERMINATE  &&
         tport_getflag(&Region) != MyPid )
    /* begin loop till Earthworm shutdown (level 1) */
  {
    /* send ora_trace_req 's heartbeat
    **********************************/
    if  ( time(&timeNow) - timeLastBeat  >=  HeartBeatInt ) 
    {
      timeLastBeat = timeNow;
      dbtsv_status( TypeHeartBeat, 0, "" ); 
    }

    /* see how our threads are feeling
    **********************************/
    if ( SnippetMakerStatus < 0)
    {
      logit("et","ERROR Snippet making thread died. Exiting\n");
      sprintf(Text,"ERROR Snippet making thread died. Exiting\n");
      dbtsv_status( TypeError, ERR_SNPT_MAKER, Text ); 
      exit (-1);
    }
    if ( MessageStackerStatus < 0)
    {
      logit("et","ERROR Message stacking thread died. Exiting\n");
      sprintf(Text,"ERROR Message stacking thread died. Exiting\n");
      dbtsv_status( TypeError, ERR_MSG_STACKER, Text ); 
      exit (-1);
    }

    sleep_ew( 3000 );  	
  }					/* end of while!(shutdown requested)  (level 1) */  

  /* Termination has been requested 
  ********************************/
  tport_detach( &Region ); /* detach from shared memory */
  logit( "t", "ora_trace_req : Termination requested; exiting!\n" );
  return(0);
}
/*--------------------------------- end of main() ---------------------------------------*/


/***********************************************************************
 *  dbtsv_config() processes command file(s) using kom.c functions;    *
 *                  exits if any errors are encountered.	       *
 ***********************************************************************/
void dbtsv_config( char *configfile )
{
   int      ncommand;     /* # of required commands you expect to process   */ 
   char     init[20];     /* init flags, one byte for each required command */
   int      nmiss;        /* number of required commands that were missed   */
   char    *com;
   char    *str;
   int      nfiles;
   int      success;
   int      i;

/* Set to zero one init flag for each required command 
 *****************************************************/   
   ncommand = 8;   /* Total number of mandatory config file commands expected */
   for( i=0; i<ncommand; i++ )  init[i] = 0;
   nLogo = 0;

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

/* 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", 
                          "ora_trace_req : Error opening command file <%s>; exiting!\n",
                           &com[1] );
                  exit( -1 );
               }
               continue;
            }

        /* Process anything else as a command 
         ************************************/
  /*0*/     if( k_its("LogFile") ) {
                LogSwitch = k_int();
                init[0] = 1;
            }
  /*1*/     else if( k_its("MyModuleId") ) {
                str = k_str();
                if(str) strcpy( MyModName, str );
                init[1] = 1;
            }
  /*2*/     else if( k_its("RingName") ) {
                str = k_str();
                if(str) strcpy( RingName, str );
                init[2] = 1;
            }
  /*3*/     else if( k_its("HeartBeatInt") ) {
                HeartBeatInt = k_long();
                init[3] = 1;
            }

         /* Enter installation & module to get event messages from
          ********************************************************/
  /*4*/     else if( k_its("GetEventsFrom") ) {
                if ( nLogo >= MAXLOGO ) {
                    logit( "e", 
                            "ora_trace_req : Too many <GetEventsFrom> commands in <%s>", 
                             configfile );
                    logit( "e", "; max=%d; exiting!\n", (int) MAXLOGO );
                    exit( -1 );
                }
                if( ( str=k_str() ) ) {
                   if( GetInst( str, &GetLogo[nLogo].instid ) != 0 ) {
                       logit( "e", 
                               "ora_trace_req : Invalid installation name <%s>", str ); 
                       logit( "e", " in <GetEventsFrom> cmd; exiting!\n" );
                       exit( -1 );
                   }
                   GetLogo[nLogo+1].instid = GetLogo[nLogo].instid;
                }
                if( ( str=k_str() ) ) {
                   if( GetModId( str, &GetLogo[nLogo].mod ) != 0 ) {
                       logit( "e", 
                               "ora_trace_req : Invalid module name <%s>", str ); 
                       logit( "e", " in <GetEventsFrom> cmd; exiting!\n" );
                       exit( -1 );
                   }
                   GetLogo[nLogo+1].mod = GetLogo[nLogo].mod;
                }
                if( GetType( "TYPE_TRIGLIST2K", &GetLogo[nLogo].type ) != 0 ) {
                    logit( "e", 
                               "ora_trace_req : Invalid message type <TYPE_TRIGLIST2K>" ); 
                    logit( "e", "; exiting!\n" );
                    exit( -1 );
                }
                nLogo++;
                init[4] = 1;
            }
  /*5*/     else if( k_its("DBservice") ) {
                str = k_str();
                if(str) strcpy( DBservice, str );
                init[5] = 1;
            }
  /*6*/     else if( k_its("DBpassword") ) {
                str = k_str();
                if(str) strcpy( DBpassword, str );
                init[6] = 1;
            }
  /*7*/     else if( k_its("DBuser") ) {
                str = k_str();
                if(str) strcpy( DBuser, str );
                init[7] = 1;
            }
            else if( k_its("Debug") ) {  /*optional command*/
                Debug_tr_sv=1;
            }

            else if( k_its("InternalQueue") ) {
                RingSize= k_long();
            }

            else if( k_its("RequestGroup") ) {
                iDefaultRequestGroup = k_long();
            }

            else if( k_its("NumAttempts") ) {
                iNumAttempts = k_long();
            }


        /* Unknown command
         *****************/ 
	    else {
                logit( "e", "ora_trace_req : <%s> Unknown command in <%s>.\n", 
                         com, configfile );
                continue;
            }

        /* See if there were any errors processing the command 
         *****************************************************/
            if( k_err() ) {
               logit( "e", 
                       "ora_trace_req : Bad <%s> command in <%s>; exiting!\n",
                        com, configfile );
               exit( -1 );
            }
	}
	nfiles = k_close();
   }

/* After all files are closed, check init flags for missed commands
 ******************************************************************/
   nmiss = 0;
   for ( i=0; i<ncommand; i++ )  if( !init[i] ) nmiss++;
   if ( nmiss ) {
       logit( "e", "ora_trace_req : ERROR, no " );
       if ( !init[0] )  logit( "e", "<LogFile> "       );
       if ( !init[1] )  logit( "e", "<MyModuleId> "    );
       if ( !init[2] )  logit( "e", "<RingName> "      );
       if ( !init[3] )  logit( "e", "<HeartBeatInt> "  );
       if ( !init[4] )  logit( "e", "<GetEventsFrom> " );
       if ( !init[5] )  logit( "e", "<DBservice> "     );
       if ( !init[6] )  logit( "e", "<DBpassword> "    );
       if ( !init[7] )  logit( "e", "<DBuser> "        );
       logit( "e", "command(s) in <%s>; exiting!\n", configfile );
       exit( -1 );
   }

   return;
}

/************************************************************************
 *  dbtsv_lookup( )   Look up important info from earthworm.h tables   *
 ************************************************************************/
void dbtsv_lookup( void )
{
/* Look up keys to shared memory regions
   *************************************/
   if( ( RingKey = GetKey(RingName) ) == -1 ) {
	fprintf( stderr,
 	        "ora_trace_req :  Invalid ring name <%s>; exiting!\n", RingName);
	exit( -1 );
   }

/* Look up installations of interest
   *********************************/
   if ( GetLocalInst( &InstId ) != 0 ) {
      fprintf( stderr, 
              "ora_trace_req : error getting local installation id; exiting!\n" );
      exit( -1 );
   }

/* Look up modules of interest
   ***************************/
   if ( GetModId( MyModName, &MyModId ) != 0 ) {
      fprintf( stderr, 
              "ora_trace_req : Invalid module name <%s>; exiting!\n", MyModName );
      exit( -1 );
   }

/* Look up message types of interest
   *********************************/
   if ( GetType( "TYPE_HEARTBEAT", &TypeHeartBeat ) != 0 ) {
      fprintf( stderr, 
              "ora_trace_req : Invalid message type <TYPE_HEARTBEAT>; exiting!\n" );
      exit( -1 );
   }
   if ( GetType( "TYPE_ERROR", &TypeError ) != 0 ) {
      fprintf( stderr, 
              "ora_trace_req : Invalid message type <TYPE_ERROR>; exiting!\n" );
      exit( -1 );
   }
   if ( GetType( "TYPE_TRIGLIST2K", &TypeTrig ) != 0 ) {
      fprintf( stderr, 
              "ora_trace_req : Invalid message type <TYPE_TRIGLIST2K>; exiting!\n" );
      exit( -1 );
   }
   return;
} 


/*************************************************************************
 * GetRequestGroupForSnippet() function used to calculate which          *
 * request group a snippet-request should be placed in.                  *
 *************************************************************************/
int GetRequestGroupForSnippet(const SNIPPET * pSnippet)
{
  /* for now (at the very least) this function just returns the
     request group specified in the config file. 
     DK 04/16/2002
  ****************************************************************/
  return(iDefaultRequestGroup);
}


/*************************************************************************
 * dbtsv_status() builds a heartbeat or error message & puts it into    *
 *                 shared memory.  Writes errors to log file & screen.   *
 *************************************************************************/
void dbtsv_status( unsigned char type, short ierr, char *note )
{
   MSG_LOGO    logo;
   char	       msg[256];
   long	       size;
   long        t;
 
/* Build the message
 *******************/ 
   logo.instid = InstId;
   logo.mod    = MyModId;
   logo.type   = type;

   time( &t );

   if( type == TypeHeartBeat )
   {
     sprintf( msg, "%ld %ld\n\0", t,MyPid);
   }
   else if( type == TypeError )
   {
     sprintf( msg, "%ld %hd %s\n\0", t, ierr, note);
     logit( "et", "ora_trace_req: Issuing Status Message: (%s)\n", note );
   }

   size = strlen( msg );   /* don't include the null byte in the message */ 	

/* Write the message to shared memory
 ************************************/
   if( tport_putmsg( &Region, &logo, size, msg ) != PUT_OK )
   {
        if( type == TypeHeartBeat ) {
           logit("et","ora_trace_req :  Error sending heartbeat.\n" );
	}
	else if( type == TypeError ) {
           logit("et","ora_trace_req :  Error sending error:%d.\n", ierr );
	}
   }

   return;
}


/********************** Message Stacking Thread *******************
 *           Move messages from transport to memory queue         *
 ******************************************************************/
thr_ret MessageStacker( void *dummy )
{
  char         *msg;           /* "raw" retrieved message               */
  int          res;
  long         recsize;        /* size of retrieved message             */
  MSG_LOGO     reclogo;        /* logo of retrieved message             */
  int          ret;

  /* Allocate space for input/output messages
  *******************************************/
  if ( ( msg = (char *) malloc(MAX_BYTES_PER_EQ) ) == (char *) NULL )
  {
    logit( "e", "export_generic: error allocating msg; exiting!\n" );
    goto error;
  }

  /* Tell the main thread we're ok
  ********************************/
  MessageStackerStatus=0;

  /* Flush the ring, something a married man should never get
     caught saying.  */
  while ( tport_getmsg( &Region, GetLogo, (short)nLogo, &reclogo,
            &recsize, msg, MAX_BYTES_PER_EQ-1 ) != GET_NONE );


  if(Debug_tr_sv==1) 
    logit("e",
	        "ora_trace_req :MessageStacker:starting to watch for "
	  			"trigger messages\n");

  /* Start service loop, picking up trigger messages
  **************************************************/
  while( 1 )
  {
    /* Get a message from transport ring
    ************************************/
    res = tport_getmsg( &Region, GetLogo, nLogo, &reclogo, &recsize, 
                       msg, MAX_BYTES_PER_EQ-1 );
    if(Debug_tr_sv==1  && res==GET_OK) 
      logit("et","Got message from transport of %d bytes,res=%d\n",recsize,res); 

    if( res == GET_NONE ) 
    {
      /*wait if no messages for us */
      sleep_ew(100); 
      continue;
    }

    /* Check return code; report errors
    ***********************************/
    if( res != GET_OK )
    {
      if( res==GET_TOOBIG )
      {
        sprintf(Text, "msg[%ld] i%d m%d t%d too long for target",
                recsize, (int) reclogo.instid,
                (int) reclogo.mod, (int)reclogo.type );
        dbtsv_status( TypeError, ERR_TOOBIG, Text );
        continue;
      }
      else if( res==GET_MISS )
      {
        sprintf(Text, "missed msg(s) i%d m%d t%d in %s",(int) reclogo.instid,
                (int) reclogo.mod, (int)reclogo.type, RingName );
        dbtsv_status( TypeError, ERR_MISSMSG, Text );
      }
      else if( res==GET_NOTRACK )
      {
        sprintf(Text, "no tracking for logo i%d m%d t%d in %s",
                (int) reclogo.instid, (int) reclogo.mod, (int)reclogo.type,
                RingName );
        dbtsv_status( TypeError, ERR_NOTRACK, Text );
      }
      else
      {
        logit("et","ERROR:ERROR:ERROR: MsgStacker():"
                   "Unrecognized return code from tport_getmsg() %d.\n",res);
        sprintf(Text,"ERROR:ERROR:ERROR: MsgStacker():"
                   "Unrecognized return code from tport_getmsg() %d.\n",res);
        dbtsv_status( TypeError, ERR_UNKNOWN, Text );
      }
    }  /* tport_getmsg() returned !GET_OK */

    /* Queue retrieved msg (res==GET_OK,GET_MISS,GET_NOTRACK)
    *********************************************************/
    RequestMutex();
    ret=enqueue( &OutQueue, msg, recsize, reclogo ); /* put it into the queue */
    /* the SnippetMaker thread is in the biz of de-queueng and processing */
    ReleaseMutex_ew();

    if(Debug_tr_sv==1)
      logit("","Queued a message\n");
    if ( ret!= 0 )
    {
      if (ret==-2)  /* Serious: quit */
      {
        sprintf(Text,"internal queue error. Terminating.");
        dbtsv_status( TypeError, ERR_QUEUE, Text );
        goto error;
      }
      if (ret==-1)
      {
        sprintf(Text,"queue cannot allocate memory. Lost message.");
        dbtsv_status( TypeError, ERR_QUEUE, Text );
        continue;
      }
      if (ret==-3) 
      {
        sprintf(Text,"Queue full. Message lost.");
        dbtsv_status( TypeError, ERR_QUEUE, Text );
        continue;
      }
    }  /* if enqueue() failed */
    if(Debug_tr_sv==1) 
      logit("t", "stacker thread: queued msg len:%ld\n",recsize);
  } /* while(1) */

  /* we're quitting
  *****************/
  error:
    MessageStackerStatus=-1; /* file a complaint to the main thread */
    KillSelfThread(); /* main thread will restart us */
}

/******************** Message Processing Thread *******************
 *           Take messages from memory queue, create trace        *
 *	snippets, and call the disposal (PutAway) routines        *
 ******************************************************************/
/* story: the trigger message contains lines specifying 'snippets' of trace data
   (EventId, SCN, start time, and duration to save). The S,C, and N specification can
   be a *, meaning wildcard. What we do below is to loop over the lines in the message,
   creating an array of trace request structures (TRACE_REQ). We then loop over those,
   trying to extract the trace from the WaveServers. 
	As we collect, we may be making premature requests. In that case, we're told
   by the WaveServer that we're premature. We accummulate worst case wait time for any
   such stragglers.We wait this amount of time, and then request the stragglers again.	   
	This version stuffs the trace data into the Oracle server via oci. Another version
   (or maybe an option on this one), is to write various format files.
*/

thr_ret SnippetMaker( void *dummy )
{
  static char errText[MAXTXT];      /* string for log/error messages          */
  
  int          rc;
  long         msgSize;        /* size of retrieved message             */
  int          ret;
  SNIPPET 	    Snppt;          /* holds params for trace snippet. From parse_trig.h */
  
  MSG_LOGO     reclogo;        /* logo of retrieved message             */
  char*	      nxtSnippetPtr;  /* pointer to next line in trigger message */
  
  EWDBid       DBidEvent;
  int          RetCode;
  char *       pEndString;
  int          iRequestGroup;


/* Buffer to store a single trigger message retrieved from dequeue() */  
  static char  trgMsg[DB_MAX_TRIG_BYTES];  

  
  /* Tell the main thread we're ok
  ********************************/
  SnippetMakerStatus=0;
  
  /* Get message from queue
  *************************/
  while(1)
  {
    /* Grab the next message if it exists */
    RequestMutex();
    ret=dequeue( &OutQueue, trgMsg, &msgSize, &reclogo);
    ReleaseMutex_ew();
    
    if(ret == -1 /* No messages */ )
    { 
      sleep_ew(500); /* wait a bit (from 1000 to 500 on 970624:ldd) */
    }
    else
    {
      /* Make sure it is a trig message */
      if( reclogo.type != TypeTrig ) /*TYPE_TRIGLIST2K*/
      {
        logit("et","ERROR: illegal message in queue\n");
        /* We presume something is hosed and commit suicide */
        SnippetMakerStatus = -1;
        KillSelfThread();
      }
      trgMsg[msgSize] = '\0';   /*null terminate the message*/
      
      if(Debug_tr_sv==1)
        logit( "","got a message\n" );   
      
      /* Initialize stuff for loop over snippets
      *****************************************/
      nxtSnippetPtr = trgMsg;  /* set next snippet pointer to start of message */
      
      /* Call the trigger message parser, so that we can get the event
         information (and the first line)
      ********************************/
      rc = parseSnippet(trgMsg, &Snppt, &nxtSnippetPtr);

      if(rc!= EW_SUCCESS /* as used by parseSnippet DK 990108*/ )
      {
        logit("et","Received a bogus trig message. Continuing\n");
        continue;
      }

      /* Call the Put Away initializer
      ********************************/
      /* This routine is responsible for initializing the scheme for disposing
      of the trace snippet requests we're hopefully about to produce */
      
      
      /* We need to parse the author, so that we only pass in up to ':',
      which is the source module in earthworm, but is for our purposes,
      the end of the usable author string in general  */
      /* DK 2000/12/07  I don't think the above assumption is correct,
      I think the database parses out everything after the first ':'
      on its own.  Oh well, not gonna mess with it */
      pEndString = strchr(Snppt.author,':');
      
      if(Debug_tr_sv)
        logit("","pEndString is %d for Author:%s\n",pEndString,Snppt.author);
      
      if(pEndString)
        *pEndString = 0x00;  /* NULL terminate right on top of the ':' */
      
      /* BEFORE WE CAN EVALUATE ANYTHING FROM Snppt, we have to do the conversion */
      if(Debug_tr_sv)
        logit("","calling ewdb_apps_putaway_StartSnippetEvent("
        "Snppt.eventID = %s,  Snppt.author = %s, Debug_tr_sv = %u\n",
        Snppt.eventId, Snppt.author, Debug_tr_sv );
      
      
      RetCode = ewdb_apps_putaway_StartSnippetEvent(&DBidEvent,Snppt.eventId,
        Snppt.author, Debug_tr_sv );
      if(Debug_tr_sv)
        logit("","returning from ewdb_apps_putaway_StartSnippetEvent\n");
      if(RetCode == EWDB_RETURN_FAILURE)
      {
        /* Something is hosed with the database.  Maybe it is our connection. */
        sprintf(errText,"ora_trace_req :  error in ewdb_apps_putaway_StartSnippetEvent()\n");
        dbtsv_status( TypeError, ERR_ORA_STSNEVENT, errText );
        SnippetMakerStatus=-1;
        KillSelfThread();
      }
      else
      {
        logit("t","Creating snippet requests for DB event %d\n",DBidEvent);
      }

      
      if(Debug_tr_sv)
        logit("t","Entering parseSnippet loop()\n");
      
      while (rc == EW_SUCCESS /* as used by parseSnippet DK 990108*/ )
      {	
        /* get next snippet params from msg (level 1)*/
        
        if (Debug_tr_sv ==1)
        {
          logit("t","After parseSnippet() in while loop\n");
          logit("","\nreturn from parseSnippet: %d\n",rc);
          logit("","eventId=%s, author=%s, sta=%s, chan=%s, net=%s, "
            "startYYYYMMDD=%s, startHHMMSS=%s, starttime=%lf, duration=%d\n",
            Snppt.eventId,Snppt.author,Snppt.sta,Snppt.chan,Snppt.net,
            Snppt.startYYYYMMDD,Snppt.startHHMMSS,Snppt.starttime,
            Snppt.duration);
        }
        
        /* create requests implied by this snippet - 
        ( it might include wildcards) 
        ****************************************************************/
        /* routine below will create the requests, and insert them into the
        database.*/
        
        iRequestGroup = GetRequestGroupForSnippet(&Snppt);

        rc = ewdb_api_ProcessSnippetReqs(Snppt.sta,Snppt.chan, Snppt.net,"",
                                     Snppt.starttime,
                                     Snppt.starttime+Snppt.duration, 
                                     DBidEvent, 
                                     iNumAttempts, iRequestGroup); 
        
        if (Debug_tr_sv==1)  
          logit("","return from ewdb_api_ProcessSnippetReqs\n");
        
        if (rc == EWDB_RETURN_FAILURE)  /* then there was some error */
        {  
          logit("","ora_trace_req : Error from ewdb_api_ProcessSnippetReqs().\n");
        }

        rc = parseSnippet(trgMsg, &Snppt, &nxtSnippetPtr);
      }  /* end while (rc =EW_SUCCESS (from parseSnippet() call)) */

      ewdb_apps_putaway_CloseSnippetEvent();  /* call the event closer */

    }  /* end else  (if no messages to dequeue) */
  }    /* end while(1)  dequeue messages */
  
}  /* end SnippetMaker() thread */

