RotoServer.cxx

Go to the documentation of this file.

// $Id: RotoServer.cxx,v 1.36 2005/09/07 19:41:58 rhatcher Exp $
//
// Communications and Control of Roto-rooter for receiving raw blocks,
// turning them into objects and writing them to files
//
// rhatcher@fnal.gov

#include "Rotorooter/RotoServer.h"
#include "Rotorooter/RotoSocket.h"
#include "Rotorooter/RotoObjectifier.h"

#include "OnlineUtil/msgLogLib/msgLog.h"

#include "TSystem.h"
#include "TServerSocket.h"

// for function: char* Form(const char *fmt, ...)
#include "TString.h"
#include "TSystem.h"

#include "RawData/RawRecord.h"
#include "RawData/RawDaqHeader.h"
#include "RawData/RawBeamMonHeader.h"

#include "RawData/RawLIAdcSummaryBlock.h"
#include "RawData/RawLITimingSummaryBlock.h"
#include "RawData/RawSnarlHeaderBlock.h"
#include "RawData/RawDigitDataBlock.h"

#include "RawData/RawDcsAlarmBlock.h"
#include "RawData/RawDcsMonitorBlock.h"
#include "RawData/RawBeamMonHeaderBlock.h"

#include "Persistency/PerOutputStream.h"
#include "Persistency/PerOutputStreamManager.h"

const UInt_t dbg_ReportCmd    = 0x0001;
const UInt_t dbg_ReportReply  = 0x0002;
const UInt_t dbg_ReportHex    = 0x0004;
const UInt_t dbg_ReportNotTo  = 0x0008;

UInt_t RotoServer::fgDebugFlags = 0;
//  dbg_ReportCmd | dbg_ReportReply | dbg_ReportNotTo;

static string reportOpt = 
  (RotoServer::GetDebugFlags() & dbg_ReportHex) ? "b" : " ";

#include "MinosObjectMap/MomNavigator.h"
#include "MessageService/MsgService.h"

ClassImp(RotoServer)
CVSID("$Id: RotoServer.cxx,v 1.36 2005/09/07 19:41:58 rhatcher Exp $");

enum RotoFailureMode {
   kSuccess         = 0x00000000,
   kFailRecvCmd     = 0x00000001,
   kFailRecvByteCnt = 0x00000002,
   kFailRecvData    = 0x00000004,
   kFailBuffResize  = 0x00000008,
   kFailNotToMe     = 0x00000010,
   kFailCmdBadFrom  = 0x00000020,
   kFailCmdUnknown  = 0x00000040,
   kFailReqUnknown  = 0x00000080,
   kFailBadRcType   = 0x00000100,
   kFailOpenFile    = 0x00001000,
   kFailCloseFile   = 0x00010000,
   kFailRecBuffer   = 0x00100000,
   kFailInflateBuff = 0x00200000,
   kFailPartialRec  = 0x00400000,
   kFailWriteRec    = 0x01000000,

   kFailNoCodeYet   = 0x40000000,
   kFailWhoops      = 0x80000000
};

//_____________________________________________________________________________
RotoServer::RotoServer(Int_t port, Int_t nwords, Bool_t allow_overwrite, 
                       Int_t tcp_nodelay_flag)
  : fPort(port), fTCP_NODELAY_flag(tcp_nodelay_flag),
    fFlatBinaryName(""), fFlatBinaryOutputFile(0), fSymlink("")
{
//
//  Purpose:  Create a RotoServer
//            Open the server socket looking for connections on the port
//            Allocate buffer to receive data
//
//  Argument: port     TCP/IP port #
//            nwords   number of words for initial data recv buffer
//            allow_overwrite   all server to overwrite output files
//            tcp_nodelay_flag  set socket to use TCP_NODELAY
//
//  Return:   (none)
//
//  Contact:  R. Hatcher
//

   Init(nwords, allow_overwrite);

   fServerSocket = new TServerSocket(port, kTRUE);
   fServerSocket->SetOption(kNoDelay,fTCP_NODELAY_flag);
}

//_____________________________________________________________________________
RotoServer::RotoServer(const char* fileName, Int_t nwords, 
                       Bool_t allow_overwrite)
   : fFlatBinaryName(""), fFlatBinaryOutputFile(0), fSymlink("")
{
//
//  Purpose:  Create a RotoServer
//            Open the RotoSocket (file emulation of a TSocket)
//            Allocate buffer to receive data
//
//  Argument: fileName File name for RotoSocket.
//            nwords   number of words for initial data recv buffer
//            allow_overwrite kTRUE if allowed to overwite output file.
//
//  Return:   (none)
//
//  Contact:  R. Hatcher
//

   Init(nwords, allow_overwrite);

   fSocket = new RotoSocket(fileName);

}
//_____________________________________________________________________________
RotoServer::~RotoServer() 
{
//
//  Purpose:  Destroy a RotoServer
//            Close the server socket looking for connections on the port
//            Close any open sockets
//            Deallocate buffer to receive data
//
//  Argument: (none)
//
//  Return:   (none)
//
//  Contact:  R. Hatcher
//

   fCurrentState = MINOS_ROOTER_STATE_STOPPING;

   if (fServerSocket) {
      fServerSocket->Close();
      delete fServerSocket;
      fServerSocket = 0;
   }

   CloseAndDeleteSocket(fSocket);

   if (fOutputStreamManager) {
      // make sure *everything* has been flushed
      fOutputStreamManager->Write();
      fOutputStreamManager->CloseStream();
      //CloseStream() makes this redundant: fOutputStreamManager->CloseFile();
     
      delete fOutputStreamManager;
      fOutputStreamManager = 0;
   }   

   SetFlatBinaryOutputFile("");

   fCurrentState = MINOS_ROOTER_STATE_UNKNOWN;

   logNotice("~RotoServer");
}

//_____________________________________________________________________________
void RotoServer::CloseAndDeleteSocket(TSocket*& socket)
{
//
//  Purpose:  Close and delete socket; zero pointer
//
//  Argument: socket   reference to pointer to socket      
//
//  Return:   (none)
//
//  Contact:  R. Hatcher
//
   if (socket) {
      if (socket->IsValid()) socket->Close();
      delete socket;
      socket = 0;
   }
}

//_____________________________________________________________________________
void RotoServer::Init(Int_t nwords, Bool_t allow_overwrite)
{
//
//  Purpose:  Common initialisation of a RotoServer
//            Allocate buffer to receive data
//
//  Argument: nwords          - number of words for initial data recv buffer
//            allow_overwrite - allow any existing file to be overwritten
//
//  Return:   (none)
//
//  Contact:  R. Hatcher
//

   fServerSocket         = 0;
   fSocket               = 0;
   fCurrentState         = MINOS_ROOTER_STATE_UNKNOWN;
   fNackMessage          = ""; 
   fStateReport          = "";
   fAccessMode           = allow_overwrite ? Per::kRecreate : Per::kNew;
   fOutputStreamManager  = 0;
   fFakeDAQFileName      = "";

   ResizeBuffer(fRecvBuffer,nwords*sizeof(Int_t)/sizeof(Char_t),"Recv");
   ResizeBuffer(fReplyBuffer,4096*sizeof(Char_t),"Reply");

   fOutputStreamManager = new PerOutputStreamManager();

   fCurrentState = MINOS_ROOTER_STATE_UNCONNECTED;
   RotoObjectifier::SysLogRawBlockRegistry();

   logInfo("Default RotoServer configuration");

   // Set default AutoSave parameters
   // all streams: 1000 records or 10 seconds
   SetAutoSaveConfig("*",1000,10);
   // Set default Compression to use file defaults
   SetCompressConfig("*",-1);
   // Set default BasketSize to a uniform 64000
   SetBasketSizeConfig("*",64000);

}

//_____________________________________________________________________________
Int_t RotoServer::SetAutoSaveConfig(const string& stream, UInt_t nrec, UInt_t nsec) 
{
//
//  Purpose:  Set AutoSave configuration values in <map>s
//
//  Argument: stream    - name of stream ("*" = all)
//            nrec      - record frequency
//            nsec      - max time duration
//
//  Return:   number of stream configs modified
//
//  Contact:  R. Hatcher
//

   int nmod = 0;

   int ibeg = Per::kDaqSnarl;
   int iend = Per::kBeamMon;
   for (int i = ibeg; i<=iend; i++ ) {
      Per::EStreamType stype = (Per::EStreamType) i;
      string treeName = Per::AsString(stype);
      if (treeName == stream || stream == "*") {
         logInfo("AutoSave is %d records or %d seconds for %s stream",
                 nrec,nsec,treeName.c_str());
         fAutoSaveIntMap[stype]  = nrec;
         fAutoSaveTimeMap[stype] = nsec;
         nmod++;
      }
   }

   return nmod;
}

//_____________________________________________________________________________
Int_t RotoServer::SetCompressConfig(const string& stream, Int_t level)
{
//
//  Purpose:  Set Compression configuration values in <map>s
//
//  Argument: stream    - name of stream ("*" = all)
//            level     - compression level
//                          -1 : use file default
//                           0 : no compression
//                           1 : default ROOT compression
//                        ...9 : maximum compression
//
//  Return:   number of stream configs modified
//
//  Contact:  R. Hatcher
//

   int nmod = 0;

   int ibeg = Per::kDaqSnarl;
   int iend = Per::kBeamMon;
   for (int i = ibeg; i<=iend; i++ ) {
      Per::EStreamType stype = (Per::EStreamType) i;
      string treeName = Per::AsString(stype);
      if (treeName == stream || stream == "*") {
         logInfo("Compression level is %d for %s stream",
                 level,treeName.c_str());
         fCompressionMap[stype]  = level;
         nmod++;
      }
   }

   return nmod;
}

//_____________________________________________________________________________
Int_t RotoServer::SetBasketSizeConfig(const string& stream, UInt_t basketsize)
{
//
//  Purpose:  Set BasketSize configuration values in <map>s
//
//  Argument: stream     - name of stream ("*" = all)
//            basketsize - size of basket
//
//  Return:   number of stream configs modified
//
//  Contact:  R. Hatcher
//

   int nmod = 0;

   int ibeg = Per::kDaqSnarl;
   int iend = Per::kBeamMon;
   for (int i = ibeg; i<=iend; i++ ) {
      Per::EStreamType stype = (Per::EStreamType) i;
      string treeName = Per::AsString(stype);
      if (treeName == stream || stream == "*") {
         logInfo("BasketSize is %d for %s stream",
                 basketsize,treeName.c_str());
         fBasketSizeMap[stype]  = basketsize;
         nmod++;
      }
   }

   return nmod;
}

//_____________________________________________________________________________
Int_t RotoServer::SetFlatBinaryOutputFile(const string& filename)
{
//
//  Purpose:  Set name of flat binary output file
//            Open file if necessary
//
//  Argument: filename   - name of file
//
//  Return:   success
//
//  Contact:  R. Hatcher
//

   // if the filename is unchanged do nothing
   if (filename == fFlatBinaryName) return 1;

   if (fFlatBinaryOutputFile) {
     // close currently open file
     logNotice("close flat binary output file: %s",fFlatBinaryName.c_str());
     fclose(fFlatBinaryOutputFile);
     fFlatBinaryOutputFile = 0;
     fFlatBinaryName = "";
   }

   if (filename != "") {
     // open the requested file
     logNotice("open flat binary output file: %s",filename.c_str());
     fFlatBinaryOutputFile = fopen(filename.c_str(),"wb");
     fFlatBinaryName = filename;
   }

   return 1;
}

//_____________________________________________________________________________
Int_t RotoServer::SetSymlink(const char* symlink_name)
{
    fSymlink = "";
    if (symlink_name) fSymlink = symlink_name;
    return 1;
}


//_____________________________________________________________________________
void RotoServer::Run() 
{
//
//  Purpose:  Main loop
//            Test validity of server socket, wait for connection attempt
//            accept commands on open socket, process and reply
//
//  Argument: (none)
//
//  Return:   (none)
//
//  Contact:  R. Hatcher
//

   //
   // is this server socket okay? If missing the socket must be 
   // a valid RotoSocket.
   if (!fServerSocket ) { 
     if ( ! fSocket->IsValid() ) {
        logCritical("Unable to open file as RotoSocket");
        return;
     }
   }
   else if (!fServerSocket->IsValid()) {
      string why;
      switch (fServerSocket->GetErrorCode()) {
      case  0: why = "no error, socket should be valid"; break;
      case -1: why = "low level socket() call failed";   break;
      case -2: why = "low level bind() call failed - ";
               why += "port may already be in use";
               break;
      case -3: why = "low level listen() call failed";   break;
      default: why = "unknown error";                    break;
      }
      // MSG("Roto",Msg::kFatal) 
      //   << endl
      //   << "  server socket failed on port "
      //   << fServerSocket->GetLocalPort()
      //   << " was invalid \n  "
      //   << why
      //   << endl;
      logCritical("server socket on port %d failed, why=%s",
                  fServerSocket->GetLocalPort(),why.c_str());
      return;
   }

   fCurrentState = MINOS_ROOTER_STATE_UNCONNECTED;
   Int_t ncmd = 0;

   RotoRcCmd cmd;

   //
   // main loop, while shutdown hasn't been requested
   //
   while (fCurrentState != MINOS_ROOTER_STATE_SHUTDOWN_REQ) {

      // Accept a connection and return a full-duplex communication socket.
      if (!fSocket) {
         fSocket = fServerSocket->Accept();
         fSocket->SetOption(kNoDelay,fTCP_NODELAY_flag);
         //fSocket->SetOption(kKeepAlive,1);
         if (fCurrentState == MINOS_ROOTER_STATE_UNCONNECTED)
             fCurrentState = MINOS_ROOTER_STATE_CONNECTED;
      }

      Int_t status = kSuccess;
      fNackMessage = "";

      // accept commands on open socket
      
      status |= RecvCommand(fSocket,cmd);
      // failure to get command+data correctly is indicative
      // of a broken socket
      if (status & (kFailRecvCmd|kFailRecvByteCnt|kFailRecvData) ) {
         logError("socket from '%s' abruptly broken",
                  fSocket->GetInetAddress().GetHostName());

         // if (fgDebugFlags & dbg_ReportCmd) 
         //    MSG("Roto", Msg::kInfo) 
         //      << "RotoServer::Run socket appears broken"  << endl;

         // should here cleanly close all files associated with
         // this socket, for now (since we're single socketed)
         // just close *everything*
         fOutputStreamManager->Write();
         fOutputStreamManager->CloseStream();
         //CloseStream() makes this redundant: fOutputStreamManager->CloseFile();
         
         CloseAndDeleteSocket(fSocket);
         fCurrentState = MINOS_ROOTER_STATE_UNCONNECTED;
         continue;
      }
      ncmd++;

      if (fgDebugFlags & dbg_ReportCmd) 
         logDebug(3,"msg %d status %d",ncmd,status);

      //if (fgDebugFlags & dbg_ReportCmd) 
      //   MSG("Roto", Msg::kInfo) 
      //      << "RotoServer::Run msg#" 
      //      << ncmd << " status=" << status
      //      << " " << cmd.AsStlString(reportOpt)
      //      << endl;

      status |= CheckTo(cmd);
      if (status & kFailNotToMe) continue;

      status |= ProcessCommand(fSocket,cmd);

      ReplyToCommand(fSocket,cmd,status);

      if (fCurrentState == MINOS_ROOTER_STATE_DISCONNECTING) {
         CloseAndDeleteSocket(fSocket);
         fCurrentState = MINOS_ROOTER_STATE_UNCONNECTED;
      }
   }
}

//_____________________________________________________________________________
void RotoServer::ResizeBuffer(TArray& buffer, Int_t size, const Char_t* name) 
{
//
//  Purpose:  Attempt to resize receive/reply buffer size
//
//  Argument: buffer   TArray to be resized
//            size     desire size in nominal units
//            name     name for log messages
//
//  Return:   (none)
//
//  Contact:  R. Hatcher
//

   int oldsize = buffer.GetSize();
   buffer.Set(0);  // toss old array so we don't copy contents

   Int_t request = size;
   Int_t failures = 0;
   while ( buffer.GetSize() != request ) {
      buffer.Set(request);
      if (buffer.GetSize() == request) break; // success!
      failures++;
      request -= 1024*sizeof(Int_t);  // failed, try smaller size
   }

   if (failures) {
         logError("resize %s buffer from %d to %d only got %d",
                  name,oldsize,size,request);
   } else {
         logNotice("resize %s buffer from %d to %d",
                   name,oldsize,size);
   }
}

//_____________________________________________________________________________
Int_t RotoServer::RecvCommand(TSocket* socket, RotoRcCmd& command)
{
//
//  Purpose:  Receive RunControl-protocol-like commands from socket
//
//  Argument: socket   ptr to socket on which command is to be read
//            command  RunControl encoded command
//
//  Return:   kSuccess (0)     = success
//            kFailRecvCmd     = error retrieving 4-byte command
//            kFailRecvByteCnt = error retrieving followup byte count
//            kFailRecvData    = error retrieving followup data
//            kFailBuffResize  = could not resize buffer to hold entire record
//                               RecvBuffer holds only partial record
//
//  Contact:  R. Hatcher
//
   Int_t  lCommand, ltouse, lBuffer;

   lCommand = fRecvBufferUsed = ltouse = lBuffer = 0;

   //
   // Retrieve "command" (4bytes of packed info)
   //
   lCommand = socket->RecvRaw(&command.fEncoded,sizeof(Int_t));
   if (lCommand != sizeof(Int_t)) {
      logError("RecvRaw command got %d expect %d",
               lCommand,sizeof(Int_t));
      return kFailRecvCmd;
   }
   //
   // if DATATOFOLLOW bit is set get the size and the data
   //
   if (command.HasDataToFollow()) {
      // byte count includes its own 4 bytes
      Int_t extra_bytes; 
      ltouse = socket->RecvRaw(&extra_bytes,sizeof(Int_t));
      fRecvBufferUsed = extra_bytes - 4;
      if (ltouse != sizeof(Int_t)) {
      logError("RecvRaw nbytes got %d expect %d",
               ltouse,sizeof(Int_t));
         return kFailRecvByteCnt;
      }
      if (fRecvBufferUsed>0) {
         // there be data in dat' d'er socket

         // if the buffer isn't big enough resize it (up only)
         if (fRecvBufferUsed>fRecvBuffer.GetSize()) 
            ResizeBuffer(fRecvBuffer,fRecvBufferUsed,"Recv");

         Bool_t partial = false;
         if (fRecvBufferUsed>fRecvBuffer.GetSize()) {
            // couldn't resize buffer large enough
            fNackMessage += 
               Form("Buffer could only hold %d of %d;",
                    fRecvBuffer.GetSize(),fRecvBufferUsed);
            partial = true;
            fRecvBufferUsed = fRecvBuffer.GetSize();
         }

         lBuffer = socket->RecvRaw(fRecvBuffer.GetArray(),fRecvBufferUsed);
         if (lBuffer != fRecvBufferUsed) {
            logError("RecvRaw buffer got %d expect %d",
                     lBuffer,fRecvBufferUsed);
            fNackMessage += 
               Form("Buffer expected %d, saw %d;",
                    fRecvBufferUsed,lBuffer);
            return kFailRecvData;
         }
         if (partial) return kFailBuffResize;

      }
   }

   return kSuccess;
}
   
//_____________________________________________________________________________
Int_t RotoServer::CheckTo(RotoRcCmd& command) 
{
//
//  Purpose:  Check if this command was directed to the right place
//
//  Argument: command   Run Control command
//
//  Return:   kSuccess      if okay
//            kFailNotToMe  if not intended for RotoRooter
//
//  Contact:  R. Hatcher
//

   int to = command.GetTo();
   if (to == MINOS_ROOTER_ROOTER) return kSuccess;

   logError("I am 0x%2.2x, got message for 0x%2.2x (cmd 0x%8.8x)",
            MINOS_ROOTER_ROOTER,to,command.fEncoded);

   // if (fgDebugFlags & dbg_ReportNotTo) 
   //   MSG("Roto", Msg::kInfo) 
   //      << "RotoServer::Run received message meant for 0x" 
   //      << hex << to << dec
   //      << ",  I am 0x" 
   //      << hex << MINOS_ROOTER_ROOTER << dec 
   //      << endl;

   return kFailNotToMe;
}
//_____________________________________________________________________________
Int_t RotoServer::ProcessCommand(TSocket* socket, RotoRcCmd command)
{
//
//  Purpose:  Process the command received on socket
//
//  Argument: socket      ptr to socket on which command is to be read
//            command     RunControl encoded command
//
//  Return:   kSuccess (0)  if successful (yields ACK)
//            various       if there was a problem (yields UNABLE_TO)
//
//  Contact:  R. Hatcher
//
   int from  = command.GetFrom();
   int type  = command.GetType();
   int instr = command.GetInstr();

   switch (type) {
   case MINOS_ROOTER_COMMAND:
      if (MINOS_ROOTER_DCP     != from && 
          MINOS_ROOTER_DCS     != from && 
          MINOS_ROOTER_BEAMMON != from ) {
         logError("MinosEntity 0x%2.2x on %s attempted COMMAND",
                  from,socket->GetInetAddress().GetHostName());

         fNackMessage += 
            Form("Not from DCP, DCS or BEAMMON but %s (0x%x);",
                 RotoRcCmd::ElementAsStlString(from).c_str(),from);
         return kFailCmdBadFrom;
         break;
      }         
      switch (instr) {
      case MINOS_ROOTER_OPENSOCKET: 
         return kSuccess;
         break;
      case MINOS_ROOTER_CLOSESOCKET: 
         // we can't do this immediately here because we need to reply first
         fCurrentState = MINOS_ROOTER_STATE_DISCONNECTING;
         return kSuccess;
         break;
      case MINOS_ROOTER_OPENFILE: 
         return OpenFile();
         break;
      case MINOS_ROOTER_CLOSEFILE: 
         return CloseFile();
         break;
      case MINOS_ROOTER_RECBUFFER: 
         return ProcessBuffer();
         break;
      case MINOS_ROOTER_SHUTDOWN: 
         logInfo("MinosEntity 0x%2.2x on %s requested shutdown",
                 from,socket->GetInetAddress().GetHostName());
         // MSG("Roto",Msg::kInfo)
         //    << " Roto SHUTDOWN command " << endl;
         fCurrentState = MINOS_ROOTER_STATE_SHUTDOWN_REQ;
         return kSuccess;
         break;
      case MINOS_ROOTER_CONFIG:
         return ProcessConfig();
         break;
      default:
         logError("MinosEntity 0x%2.2x on %s bad cmd 0x%8.8x",
                  from,socket->GetInetAddress().GetHostName(),
                  command.fEncoded);

         fNackMessage += 
            Form("Unmodelled %s %s;",
                 RotoRcCmd::TypeAsStlString(type).c_str(),
                 RotoRcCmd::InstrAsStlString(type,instr).c_str());
         return kFailCmdUnknown;
         break;
      }
      break;
   case MINOS_ROOTER_REQUEST:
      switch (instr) {
      case MINOS_ROOTER_REQ_CURRENT_STATE: 
         return kSuccess;
         break;
      case MINOS_ROOTER_REQ_STATUS_REPORT: 
         BuildStateReport();
         return kSuccess;
         break;
      default:
         logError("MinosEntity 0x%2.2x on %s bad cmd 0x%8.8x",
                from,socket->GetInetAddress().GetHostName(),
                command.fEncoded);

         fNackMessage += 
            Form("Unmodelled %s %s;",
                 RotoRcCmd::TypeAsStlString(type).c_str(),
                 RotoRcCmd::InstrAsStlString(type,instr).c_str());
         return kFailReqUnknown;
         break;
      }
      break;
   default:
         logError("MinosEntity 0x%2.2x on %s bad cmd 0x%8.8x",
                  from,socket->GetInetAddress().GetHostName(),
                  command.fEncoded);

         fNackMessage += 
            Form("Unmodelled %s %s;",
                 RotoRcCmd::TypeAsStlString(type).c_str(),
                 RotoRcCmd::InstrAsStlString(type,instr).c_str());
      return kFailBadRcType;
      break;
   }
   return kFailWhoops;
}

//_____________________________________________________________________________
void RotoServer::ReplyToCommand(TSocket* socket, RotoRcCmd command,
                                Int_t status)
{
//
//  Purpose:  Reply to the command received on socket
//            Special commands send more than just ack/nack
//
//  Argument: socket   ptr to socket on which command is to be read
//            command  RunControl encoded command
//            ack      whether to send ACKNOWLEDGE or UNABLE_TO
//
//  Return:   (none)
//
//  Contact:  R. Hatcher
//
   static Int_t nreply = 0;
   nreply++;

   Int_t type  = command.GetType();
   Int_t instr = command.GetInstr();
   Bool_t ack  = ( kSuccess == status );

   RotoRcCmd reply = command;
   reply.SwapFromTo();

   switch (type) {
   case MINOS_ROOTER_COMMAND: {
      //
      // send ACK/UNABLE_TO for COMMAND
      //

      // construct the entirety of what we're sending first
      // rather than sending the individual bits
      // combine ACK with STATE_REPORT into single socket->SendRaw

      Int_t  ackfield   = 
         (ack) ? MINOS_ROOTER_ACKNOWLEDGE : MINOS_ROOTER_UNABLE_TO;
      Bool_t nackmsg    = ( ! ack && fNackMessage.length()>0);

      int totmsgsize = sizeof(Int_t);
      if (nackmsg) totmsgsize += sizeof(Int_t) + fNackMessage.length();
      totmsgsize += sizeof(Int_t);
      if (totmsgsize>fReplyBuffer.GetSize())
         ResizeBuffer(fReplyBuffer,totmsgsize,"Reply");
      Char_t* p = fReplyBuffer.GetArray();

      reply.SetType(ackfield);
      reply.SetDataToFollow(nackmsg);
//      socket->SendRaw(&reply.fEncoded,sizeof(Int_t));
      memcpy(p,&reply.fEncoded,sizeof(Int_t));
      p += sizeof(Int_t);

      if (!ack) {
         logError("sent NACK 0x%8.8x to MinosEntity 0x%2.2x on %s",
                  reply.fEncoded,reply.GetTo(),
                  socket->GetInetAddress().GetHostName());
         logError(reply.AsStlString().c_str());
         if (fNackMessage.length()>0) 
            logError("  %s",fNackMessage.c_str());
      }
#ifdef VERBOSE_SYSLOG
      else {
//rwh: very verbose
         logInfo("ACK to %s",command.AsStlString().c_str());
      }
#endif

      if (nackmsg) {
         // byte count includes its own 4 bytes
         Int_t nackmsg_length = fNackMessage.length() + 4;
//         socket->SendRaw(&nackmsg_length,sizeof(Int_t));
//         socket->SendRaw(fNackMessage.c_str(),nackmsg_length-4);
         memcpy(p,&nackmsg_length,sizeof(Int_t));
         p += sizeof(Int_t);
         memcpy(p,fNackMessage.c_str(),nackmsg_length-4);
         p += nackmsg_length-4;
      }
      //
      // protocol says COMMAND also gets a STATE_REPORT reply
      //
      reply.SetType(MINOS_ROOTER_STATE_REPORT);
      reply.SetInstr(fCurrentState);
      reply.SetDataToFollow(false);
//      socket->SendRaw(&reply.fEncoded,sizeof(Int_t));
      memcpy(p,&reply.fEncoded,sizeof(Int_t));
      p += sizeof(Int_t);
      socket->SendRaw(fReplyBuffer.GetArray(),totmsgsize);                 
        
      //if (fgDebugFlags & dbg_ReportReply) {
      //   string status_msg = "";
      //   if (status != 0) status_msg += Form("\n  status 0x%x",status);
      //   MSG("Roto", Msg::kInfo) 
      //      << "RotoServer::ReplyToCommand reply#" 
      //      << nreply 
      //      << " " << reply.AsStlString(reportOpt)
      //      << status_msg
      //      << endl;

      break;
   }
   case MINOS_ROOTER_REQUEST:
      switch (instr) {
      case MINOS_ROOTER_REQ_CURRENT_STATE:
         reply.SetType(MINOS_ROOTER_STATE_REPORT);
         reply.SetInstr(fCurrentState);
         reply.SetDataToFollow(false);
         socket->SendRaw(&reply.fEncoded,sizeof(Int_t));
         break;
      case MINOS_ROOTER_REQ_STATUS_REPORT: {
         reply.SetType(MINOS_ROOTER_REPORT);
         reply.SetInstr(MINOS_ROOTER_REQ_STATUS_REPORT);
         reply.SetDataToFollow(true);
         socket->SendRaw(&reply.fEncoded,sizeof(Int_t));
         // byte count includes its own 4 bytes
         Int_t report_length = fStateReport.length() + 4;
         socket->SendRaw(&report_length,sizeof(Int_t));
         socket->SendRaw(fStateReport.c_str(),report_length-4);

         //if (fgDebugFlags & dbg_ReportReply) 
         //   MSG("Roto", Msg::kInfo) 
         //      << "RotoServer::ReplyToCommand reply#" 
         //      << nreply 
         //      << " " << reply.AsStlString(reportOpt)
         //      << endl
         //      << "  \"" << fStateReport << "\" (" << report_length << ")"
         //      << endl;
         break;
      }

      default:
         break;
      }
      break;
   default:
      break;
   }      
          
}

//_____________________________________________________________________________
Int_t RotoServer::OpenFile()
{
//
//  Purpose:  Open a file, use buffer to determine parameters
//
//  Argument: (none)
//
//  Return:   if 0 then file creation was okay
//
//  Contact:  R. Hatcher
//   
   int    nwords = fRecvBufferUsed/sizeof(Int_t);
   if (nwords<1) {
      logError("OpenFile() too few words (%d)",nwords);
      return kFailOpenFile;
   }
   Int_t* ibuffer = (Int_t*) fRecvBuffer.GetArray();
   Int_t from = ibuffer[0];

   if (fFakeDAQFileName != "") {
      fNackMessage += "OpenFile not possible, already in FakeDaqFile mode ";
      fNackMessage += ";";
      logNotice("OpenFile() already in fake mode");
      return kFailCmdBadFrom;
   }

   switch (from) {
   case MINOS_ROOTER_DCP:
      if (nwords<4) {
         logError("OpenFile() from DCP too few words (%d)",nwords);
         return kFailOpenFile;
      }
      return OpenDaqFile(ibuffer[1],ibuffer[2],ibuffer[3]);
      break;
   case MINOS_ROOTER_DCS:
      if (nwords<4) {
         logError("OpenFile() from DCS too few words (%d)",nwords);
         return kFailOpenFile;
      }
      return OpenDcsFile(ibuffer[1],ibuffer[2],ibuffer[3]);
      break;
   case MINOS_ROOTER_BEAMMON:
      if (nwords<4) {
         logError("OpenFile() from BEAMMON too few words (%d)",nwords);
         return kFailOpenFile;
      }
      return OpenBeamMonFile(ibuffer[1],ibuffer[2],ibuffer[3]);
      break;
   default: {
      // a fake "from" in the buffer is a hack to support
      // binary files without proper header blocks
      // buffer should be 4bytes of "fake from" + the name
      EFileType ftype = kDaqFile;
      const char* fname = fRecvBuffer.GetArray()+sizeof(Int_t);
      Int_t err = OpenFile(fname,ftype);
      if (!err) fFakeDAQFileName = fname;
      return err;
      break;
      }
   }
   fNackMessage += "OpenFile not possible from ";
   fNackMessage += RotoRcCmd::ElementAsStlString(from);
   fNackMessage += ";";
   return kFailCmdBadFrom;
}

//_____________________________________________________________________________

Int_t RotoServer::CloseFile()
{
//
//  Purpose:  Close a file, use buffer to determine parameters
//
//  Argument: (none)
//
//  Return:   if 0 then file closure was okay
//
//  Contact:  R. Hatcher
//   
   int    nwords = fRecvBufferUsed/sizeof(Int_t);
   if (nwords<1) {
      logError("CloseFile() too few words (%d)",nwords);
      return kFailOpenFile;
   }
   Int_t* ibuffer = (Int_t*) fRecvBuffer.GetArray();
   Int_t from = ibuffer[0];

   switch (from) {
   case MINOS_ROOTER_DCP:
      if (fFakeDAQFileName != "") {
         fNackMessage += "CloseFile not possible, in FakeDaqFile mode ";
         fNackMessage += ";";
         return kFailCmdBadFrom;
      }
      if (nwords<4) {
         logError("CloseFile() from DCP too few words (%d)",nwords);
         return kFailOpenFile;
      }
      return CloseDaqFile(ibuffer[1],ibuffer[2],ibuffer[3]);
      break;
   case MINOS_ROOTER_DCS:
      if (fFakeDAQFileName != "") {
         fNackMessage += "CloseFile not possible, in FakeDaqFile mode ";
         fNackMessage += ";";
         return kFailCmdBadFrom;
      }
      if (nwords<4) {
         logError("CloseFile() from DCS too few words (%d)",nwords);
         return kFailOpenFile;
      }
      return CloseDcsFile(ibuffer[1],ibuffer[2],ibuffer[3]);
      break;
   case MINOS_ROOTER_BEAMMON:
      if (fFakeDAQFileName != "") {
         fNackMessage += "CloseFile not possible, in FakeDaqFile mode ";
         fNackMessage += ";";
         return kFailCmdBadFrom;
      }
      if (nwords<4) {
         logError("CloseFile() from BEAMMON too few words (%d)",nwords);
         return kFailOpenFile;
      }
      return CloseBeamMonFile(ibuffer[1],ibuffer[2],ibuffer[3]);
      break;
   default: {
      // a fake "from" in the buffer is a hack to support
      // binary files without proper header blocks
      // buffer should be 4bytes of "fake from" + the name
      EFileType ftype = kDaqFile;
      const char* fname = fRecvBuffer.GetArray()+sizeof(Int_t);
      Int_t err = CloseFile(fname,ftype);
      if (!err) fFakeDAQFileName = "";
      return err;
      break;
      }
   }
   fNackMessage += "CloseFile not possible from ";
   fNackMessage += RotoRcCmd::ElementAsStlString(from);
   fNackMessage += ";";
   return kFailCmdBadFrom;
}

//_____________________________________________________________________________
string RotoServer::FileExtName(EFileType ftype)
{
//
//  Purpose:  return standardized file extension names
//
//  Argument: ftype      enum of file types
//
//  Return:   string version of standard name
//
//  Contact:  R. Hatcher
//   
   switch (ftype) {
   case kDaqFile:        return string("mdaq.root");    break;
   case kDcsFile:        return string("mdcs.root");    break;
   case kBeamMonFile:    return string("mbeam.root");   break;
   case kBogusFile:      return string("mall.root");    break;
   default:
      logError("bad FileExtName request %d",(int)ftype);
      //MSG("Roto", Msg::kInfo) 
      //   << "RotoServer::StreamName illegal enum value: " 
      //   << (int)stype << endl;      
      return string("generic.root");
      break;
   }

}

//_____________________________________________________________________________
string RotoServer::BogusFileName()
{
//
//  Purpose:  return bogus file name
//
//  Argument: (none)
//
//  Return:   string of a file name in which to put records
//            who's real destination couldn't be determined
//            tag it with the machine, port, and current date/time
//
//  Contact:  R. Hatcher
//   
   Int_t date = fStartTime.GetDate();
   if (date>19700000) date %= 1000000;
   Int_t time = fStartTime.GetTime();
   string fname = Form("BOGUS_%s_p%d_%6.6d_%6.6d",
                       gSystem->HostName(),fPort,date,time);
   // squeeze out any blanks with "_"
   string::size_type where;
   while (string::npos != (where = fname.find(" "))) {
      fname.replace(where,1,"_");
   }
   return fname;
}

//_____________________________________________________________________________
Int_t RotoServer::OpenDaqFile(Int_t detector, Int_t run, Int_t subrun)
{
//
//  Purpose:  Open a DAQ file
//
//  Argument: detector   detector type (near,far,caldet)
//            run        run # 
//            subrun     subrun #
//
//  Return:   if 0 then file creation was okay
//
//  Contact:  R. Hatcher
//   

   string filebase = BuildDaqBaseName(detector,run,subrun);
   return OpenFile(filebase.c_str(),kDaqFile);
}

//_____________________________________________________________________________
Int_t RotoServer::OpenDcsFile(Int_t detector, Int_t sec, Int_t nanosec)
{
//
//  Purpose:  Open the DCS file
//
//  Argument: detector   detector type (near,far,caldet)
//            sec        seconds when file was started
//            nanosec    nano when file was started
//
//  Return:   if 0 then file creation was okay
//
//  Contact:  R. Hatcher
//   

   string filebase = BuildDcsBaseName(detector,sec,nanosec);
   return OpenFile(filebase.c_str(),kDcsFile);
}

//_____________________________________________________________________________
Int_t RotoServer::OpenBeamMonFile(Int_t detector, Int_t sec, Int_t nanosec)
{
//
//  Purpose:  Open the BEAMMON file
//
//  Argument: detector   detector type (near,far,caldet)
//            sec        seconds when file was started
//            nanosec    nano when file was started
//
//  Return:   if 0 then file creation was okay
//
//  Contact:  R. Hatcher
//   

   string filebase = BuildBeamMonBaseName(detector,sec,nanosec);
   return OpenFile(filebase.c_str(),kBeamMonFile);
}

//_____________________________________________________________________________
Int_t RotoServer::OpenFile(const Char_t* fnamebase, EFileType ftype)
{
//
//  Purpose:  Open a DAQ/DCS/BEAMMON file
//
//  Argument: fnamebase  file name without extension
//            ftype      kDaqFile, kDcsFile, kBeamMonFile or kBogusFile
//
//  Return:   if 0 then file creation was okay
//
//  Contact:  R. Hatcher
// 

   const char* pbasedir = gSystem->Getenv("DAQ_DATA_DIR");
   if (kDcsFile == ftype) {
     // if DCS file and DCS_DATA_DIR is defined use that
     const char* pdcsdir = gSystem->Getenv("DCS_DATA_DIR");
     if (pdcsdir) pbasedir = pdcsdir;
   }
   if (kBeamMonFile == ftype) {
     // if Beam Monitor file and BEAM_DATA_DIR is defined use that
     const char* dir = gSystem->Getenv("BEAM_DATA_DIR");
     if (dir) pbasedir = dir;
   }
   // if no base dir use current working directory
   if (!pbasedir) pbasedir = ".";
   string filename = string(pbasedir)  + "/" + 
                     string(fnamebase) + "." + 
                     FileExtName(ftype);
   logInfo("OpenFile '%s'",filename.c_str());

   bool openok = true;

   // create possible streams, attach each to the file
   // !!!!!! needs more protection against failures

   int ibeg = Per::kDaqSnarl;
   int iend = Per::kDcsMonitor;
   if (kDaqFile == ftype) {
      ibeg = Per::kDaqSnarl;
      iend = Per::kLightInjection;
   }
   else
   if (kDcsFile == ftype) {
      ibeg = Per::kDcsAlarm;
      iend = Per::kDcsMonitor;
   }
   else
   if (kBeamMonFile == ftype) {
      ibeg = Per::kBeamMon;
      iend = Per::kBeamMon;
   }

   string fbase = fnamebase;
   // if the file is a dcs, beammon or bogus one, then the stream names
   // only have "dcs" or "bogus" and not the full file name
   // as there can be only one of each
   switch (ftype) {
   case kDcsFile:     fbase = "dcs";   break;
   case kBeamMonFile: fbase = "beam";  break;
   case kBogusFile:   fbase = "bogus"; break;
   default:
     // do nothing
     break;
   }
   int fversion = 0;

   // if running from RotoSocket (ie. connected directly to input
   // binary file) then don't bother with AutoSave   
   bool fromFile = fSocket->InheritsFrom("RotoSocket");
   if (fromFile) MSG("Roto",Msg::kInfo)
         << "Reading from RotoSocket - turn off autosave"
         << endl;

   for (int i = ibeg; i<=iend; i++ ) {
      Per::EStreamType stype = (Per::EStreamType) i;
      string treeName = Per::AsString(stype);
      string rotoStrmName = fbase + "." + treeName;

      Int_t  compressLevel = fCompressionMap[stype];
      UInt_t basketSize    = fBasketSizeMap[stype];

      PerOutputStream* astream =
         fOutputStreamManager->OpenStream(rotoStrmName,treeName,
                                          "RawRecord","","",Per::kRecSplit,
                                          basketSize,compressLevel);
      if (astream) {
         // possible tweaks
         //   autosave (interval: 0, setbasket = true)
         //   basketsaveint (interval: 0)
         if (fromFile) astream->SetAutoSave(0,0);
         else {
            UInt_t autoSaveInt  = fAutoSaveIntMap[stype];
            UInt_t autoSaveTime = fAutoSaveTimeMap[stype];
            astream->SetAutoSave(autoSaveInt,autoSaveTime);
         }
         bool openok_astream = 
            fOutputStreamManager->SetFile(rotoStrmName,filename,fAccessMode);

         // handle potential case where file already exists
         // and the rotorooter is in kNew not kRecreate mode
         Per::EErrorCode ecode = astream->GetErrorCode();
         if (Per::kErrFileExists == ecode) {
            string basefilename = filename;
            string::size_type where = filename.rfind(".m");
            if (string::npos == where) where = filename.rfind(".root");
            if (string::npos == where) where = filename.length();
            bool keep_trying = true;
            // start adding version numbers
            do {
               ++fversion;
               filename = basefilename;  // without any attempt at versioning
               filename.insert(where,Form(".%d",fversion));
               openok_astream = 
                  fOutputStreamManager->SetFile(rotoStrmName,filename,fAccessMode);
               if (Per::kErrSuccess != ecode) {
                  logWarn("open versioned file saw Per error: (%d) %s",
                          (int)ecode,Per::AsString(ecode));
               }

               ecode = astream->GetErrorCode();
               keep_trying = (fversion<100) &&
                  (Per::kErrFileExists == ecode);
            } while (keep_trying);
            // report if there was success
            if (Per::kErrSuccess == ecode) 
               logInfo("open stream using file %s",filename.c_str());

         } // file exists error

         if (Per::kErrSuccess != ecode) {
            logWarn("open stream saw Per error: (%d) %s",
                    (int)ecode,Per::AsString(ecode));
         }

         openok = openok && openok_astream;
         
         if (openok_astream) 
            logInfo("open stream %s as %s",
                    treeName.c_str(),rotoStrmName.c_str());

         //MSG("Roto", Msg::kDebug) 
         //   << "RotoServer::OpenDaqFile stream " << strmName
         //   << " opened " << ( (openok_astream) ? "ok" : "failed" )
         //   << endl;
      }
      else {
         openok = false;
      }
   }

   if (openok && fSymlink.size()) {
       gSystem->Unlink(fSymlink.c_str());
       gSystem->Symlink(filename.c_str(),fSymlink.c_str());
   }

   return (openok) ? kSuccess : kFailWhoops;
}

//_____________________________________________________________________________
Int_t RotoServer::CloseDaqFile(Int_t detector, Int_t run, Int_t subrun)
{
//
//  Purpose:  Close a DAQ file
//
//  Argument: detector   detector type (near,far,caldet)
//            run        run # 
//            subrun     subrun #
//
//  Return:   if >0 then file closure was okay
//
//  Contact:  R. Hatcher
//   

   string filebase = BuildDaqBaseName(detector,run,subrun);
   return CloseFile(filebase.c_str(),kDaqFile);
}

//_____________________________________________________________________________
Int_t RotoServer::CloseDcsFile(Int_t detector, Int_t sec, Int_t nanosec)
{
//
//  Purpose:  Close the DCS file
//
//  Argument: detector   detector type (near,far,caldet)
//            sec        seconds when file was started
//            nanosec    nano when file was started
//
//  Return:   if 0 then file closure as okay
//
//  Contact:  R. Hatcher
//   

   string filebase = BuildDcsBaseName(detector,sec,nanosec);
   return CloseFile(filebase.c_str(),kDcsFile);
}

//_____________________________________________________________________________
Int_t RotoServer::CloseBeamMonFile(Int_t detector, Int_t sec, Int_t nanosec)
{
//
//  Purpose:  Close the BEAMMON file
//
//  Argument: detector   detector type (near,far,caldet)
//            sec        seconds when file was started
//            nanosec    nano when file was started
//
//  Return:   if 0 then file closure as okay
//
//  Contact:  R. Hatcher
//   

   string filebase = BuildBeamMonBaseName(detector,sec,nanosec);
   return CloseFile(filebase.c_str(),kBeamMonFile);
}

//_____________________________________________________________________________
Int_t RotoServer::CloseFile(const Char_t* fnamebase, EFileType ftype)
{
//
//  Purpose:  Close a DAQ/DCS/BEAMMON file
//
//  Argument: fnamebase  file name without extension
//            ftype     kDaqFile, kDcsFile, kBeamMonFile or kBogusFile
//
//  Return:   if >0 then file closure was okay
//
//  Contact:  R. Hatcher
//   
   const char* pbasedir = gSystem->Getenv("DAQ_DATA_DIR");
   if (kDcsFile == ftype) {
     // if DCS file and DCS_DATA_DIR is defined use that
     const char* pdcsdir = gSystem->Getenv("DCS_DATA_DIR");
     if (pdcsdir) pbasedir = pdcsdir;
   }
   // if no base dir use current working directory
   if (!pbasedir) pbasedir = ".";
   string filename = string(pbasedir)  + "/" + 
                     string(fnamebase) + "." + 
                     FileExtName(ftype);
   logInfo("CloseFile '%s'",filename.c_str());

   bool closeok = true;

   // close possible streams
   // !!!!!! needs more protection against failures

   int ibeg = Per::kDaqSnarl;
   int iend = Per::kDcsMonitor;
   if (kDaqFile == ftype) {
      ibeg = Per::kDaqSnarl;
      iend = Per::kLightInjection;
   }
   else
   if (kDcsFile == ftype) {
      ibeg = Per::kDcsAlarm;
      iend = Per::kDcsMonitor;
   }
   else
   if (kBeamMonFile == ftype) {
      ibeg = Per::kBeamMon;
      iend = Per::kBeamMon;
   }

   string fbase = fnamebase;
   // if the file is a dcs or bogus one, then the stream names
   // only have "dcs" or "bogus" and not the full file name
   // as there can be only one of each
   switch (ftype) {
   case kDcsFile:     fbase = "dcs";   break;
   case kBeamMonFile: fbase = "beam";  break;
   case kBogusFile:   fbase = "bogus"; break;
   default:
     // do nothing
     break;
   }

   for (int i = ibeg; i<=iend; i++ ) {
      Per::EStreamType stype = (Per::EStreamType) i;
      string treeName = Per::AsString(stype);
      string rotoStrmName = fbase + "." + treeName;

      fOutputStreamManager->Write(rotoStrmName);
      fOutputStreamManager->CloseStream(rotoStrmName);
   }
   // CloseStream does this for us (besides this method takes a stream name)
   // fOutputStreamManager->CloseFile(filename);
                                                  
   return (closeok) ? kSuccess : kFailWhoops;
}

//_____________________________________________________________________________
string RotoServer::BuildDaqBaseName(Int_t detector, Int_t run, Int_t subrun)
{
//
//  Purpose:  Construct a standardized DAQ file/stream name base
//
//  Argument: detector   detector type (near,far,caldet)
//            run        run # 
//            subrun     subrun #
//
//  Return:   the base file/stream name
//
//  Contact:  R. Hatcher
//   

   Detector::Detector_t det = (Detector::Detector_t)(detector);
   Char_t detchar = Detector::AsString(det)[0];

   string fname = Form("%c%8.8d_%4.4d",detchar,run,subrun);
   return fname;
}

//_____________________________________________________________________________
string RotoServer::BuildDcsBaseName(Int_t detector, Int_t sec, Int_t nanosec)
{
//
//  Purpose:  Construct a standardized DCS file/stream name
//
//  Argument: detector   detector type (near,far,caldet)
//            sec        seconds of timestamp
//            nanosec    nanoseconds
//
//  Return:   the base file/stream name
//
//  Contact:  R. Hatcher
//   

   Detector::Detector_t det = (Detector::Detector_t)(detector);
   Char_t detchar = Detector::AsString(det)[0];
   VldTimeStamp starttime(sec,nanosec);
   Int_t date = starttime.GetDate();
   if (date>19700000) date %= 1000000;
   Int_t time = starttime.GetTime();

   string fname = Form("%c%6.6d_%6.6d",detchar,date,time);
   return fname;
}

//_____________________________________________________________________________
string RotoServer::BuildBeamMonBaseName(Int_t /* detector */, 
                                        Int_t sec, Int_t nanosec)
{
//
//  Purpose:  Construct a standardized BEAMMON file/stream name
//
//  Argument: detector   detector type (near,far,caldet)
//            sec        seconds of timestamp
//            nanosec    nanoseconds
//
//  Return:   the base file/stream name
//
//  Contact:  R. Hatcher
//   

   // Detector::Detector_t det = (Detector::Detector_t)(detector);
   // Char_t detchar = Detector::AsString(det)[0];
   VldTimeStamp starttime(sec,nanosec);
   Int_t date = starttime.GetDate();
   if (date>19700000) date %= 1000000;
   Int_t time = starttime.GetTime();

   string fname = Form("B%6.6d_%6.6d",date,time);
   return fname;
}

//_____________________________________________________________________________
Int_t RotoServer::ProcessBuffer() 
{
//
//  Purpose:  Process the RawRecord that is in the buffer
//            Build it into objects, write it out
//
//  Argument: (none)
//
//  Return:   if 0 then conversion and write were successful
//
//  Contact:  R. Hatcher
//   
   static Int_t nrec = 0;

   logDebug(5,"process buffer %d",nrec);
   // MSG("Roto", Msg::kDebug) << "RotoServer::ProcessBuffer " << nrec << endl;

   if (fFlatBinaryOutputFile) {
     Int_t recLength = fRecvBufferUsed + sizeof(recLength);
     fwrite(&recLength,sizeof(recLength),1,fFlatBinaryOutputFile);
     fwrite(fRecvBuffer.GetArray(),fRecvBufferUsed,1,fFlatBinaryOutputFile);
   }

   Int_t status = kSuccess;
   Int_t obj_status = 0;

   // turn buffer into record(s)
   MomNavigator* mom = 
      RotoObjectifier::BufferInflate(fRecvBuffer.GetArray(),
                                     fRecvBufferUsed,obj_status);

   if (!mom) {
      logError("buffer inflate returned no MOM, REC %d ERR %d",
               nrec,obj_status);
      return kFailInflateBuff;
   }

   if (4 == obj_status || 8 == obj_status) {

      logError("partial buffer failure rec %d err %d",nrec,obj_status);

      status |= kFailPartialRec;
      fNackMessage += "Portion of record could not be converted to RawDataBlocks;";
   }      
                     
   // write records out

   const RawRecord* rawrec = 0;
   TIter reciter = mom->FragmentIter();
   while ( ( rawrec = dynamic_cast<const RawRecord*>(reciter()) ) ) {
     status |= WriteRawRecord(rawrec);
     nrec++;
   }

   delete mom; // clean up

   return status;

}

//_____________________________________________________________________________
Int_t RotoServer::WriteRawRecord(const RawRecord* rawrec) 
{
//
//  Purpose:  Write the inflated RawRecord
//            Build it into objects, write it out
//
//  Argument: (none)
//
//  Return:   if 0 then conversion and write were successful
//
//  Contact:  R. Hatcher
//   

   Int_t status = kSuccess;

   string rotoStrmName = ChooseStreamName(rawrec);

   if (!fOutputStreamManager) {
      logError("process buffer had no output stream manager");
      status |= kFailWriteRec;
      return status;
   }

   PerOutputStream* stream = 0;
   stream = dynamic_cast<PerOutputStream*>
      (fOutputStreamManager->GetOpenedStream(rotoStrmName));

   if (!stream) {
      // no open stream with the right name ...
      // openfile wasn't called ?? or perhaps the data is
      // corrupted.  we want to protect this data so
      // put it in a file of bogus data
      logError("process buffer had no stream '%s'",
               rotoStrmName.c_str());
      string streamType = // last part of rotoStrmName after final .
         rotoStrmName.substr(rotoStrmName.find_last_of(".")+1,string::npos); 
      rotoStrmName =  "bogus." + streamType;
      stream = dynamic_cast<PerOutputStream*>
         (fOutputStreamManager->GetOpenedStream(rotoStrmName));
      if (!stream) {
         // perhaps the bogus file isn't open ...
         EFileType ftype = kBogusFile;
         string fname = BogusFileName();
         Int_t err = OpenFile(fname.c_str(),ftype);
         if (err) {
            logError("process buffer could not open 'bogus' file '%s'",
                     fname.c_str());
         }
         // last try at getting the stream
         stream = dynamic_cast<PerOutputStream*>
            (fOutputStreamManager->GetOpenedStream(rotoStrmName));
         
      }
   }
   
   if (stream) {
//         MSG("Roto", Msg::kVerbose) 
//            << "RotoServer::ProcessBuffer put on stream " 
//            << strmName << endl;
         // remove const for PerOutputStream::SetObject
      RawRecord *rawrec_nonconst =
         const_cast<RawRecord *>(rawrec);
      stream->SetObject(rawrec_nonconst);
      stream->Store();
   } 
   else {
      logError("process buffer still had no stream '%s'",
               rotoStrmName.c_str());
      //MSG("Roto", Msg::kWarning) 
      //   << "RotoServer::ProcessBuffer no stream " << rotoStrmName;
      status |= kFailWriteRec;
   }

   return status;
}

//_____________________________________________________________________________
string RotoServer::ChooseStreamName(const RawRecord* rawrec) 
{
//
//  Purpose:  Pick which stream to put this record into
//
//  Argument: (none)
//
//  Return:   name of stream
//
//  Contact:  R. Hatcher
//   
   string rotoStrmName;

   VldContext vldc = rawrec->GetRawHeader()->GetVldContext(); // always okay
   
   const RawDaqHeader *rawdaqhead = 
      dynamic_cast<const RawDaqHeader *>(rawrec->GetRawHeader());
   
   const RawBeamMonHeader *rawbeammonhead =
      dynamic_cast<const RawBeamMonHeader *>(rawrec->GetRawHeader());

   if (rawdaqhead) {
      // this came from the DAQ
      Int_t run    = rawdaqhead->GetRun();
      Int_t subrun = rawdaqhead->GetSubRun();
      
      Per::EStreamType stype = Per::kDaqMonitor;
      
      TIter blkiter =
         (const_cast<RawRecord*>(rawrec))->GetRawBlockIter();
      TObject*            tobj = 0;
      const RawDataBlock* rb   = 0;

      Bool_t sawLISummary = false;
      Bool_t sawSnarlHeader = false;
      Bool_t sawDigitBlock = false;

      while ( (tobj = blkiter()) ) {
         rb = dynamic_cast<RawDataBlock*>(tobj);

         if ((dynamic_cast<const RawLIAdcSummaryBlock*>(rb))) 
            sawLISummary = true;

         if ((dynamic_cast<const RawLITimingSummaryBlock*>(rb)))
            sawLISummary = true;

         if ((dynamic_cast<const RawSnarlHeaderBlock*>(rb)))
            sawSnarlHeader = true;

         if ((dynamic_cast<const RawDigitDataBlock*>(rb)))
            sawDigitBlock = true;
      }
      // decide on which stream
      // if has LI blocks then --> LightInjection
      // else if had Snarl then --> DaqSnarl
      // otherwise --> DaqMonitor
      
      if (sawLISummary) {
         stype = Per::kLightInjection;
      }
      else if (sawSnarlHeader || sawDigitBlock ) {
         stype = Per::kDaqSnarl;
         if (!sawSnarlHeader || !sawDigitBlock ) {
            logWarn("XOR RawSnarlHeaderBlock %d RawDgitiDataBlock %d",
                   sawSnarlHeader,sawDigitBlock);
            
            //MSG("Roto", Msg::kWarning) 
            //   << "RotoServer::ProcessBuffer " 
            //   << "  RawSnarlHeaderBlock " << snarl_hdr 
            //   << "  RawDigitDataBlock " << digits
            //   << endl;
         }
      }
      
      // if we're in "one file" mode then the stream naming
      // doesn't depend on run/subrun
      if (fFakeDAQFileName != "") {
         rotoStrmName = fFakeDAQFileName;
      } else {
         rotoStrmName = BuildDaqBaseName(vldc.GetDetector(),run,subrun);
      }
      string streamType = Per::AsString(stype);
      rotoStrmName +=  "." + streamType;

   }
   else if (rawbeammonhead) {
      // this came from the BEAMMON
     Per::EStreamType stype = Per::kBeamMon;

     string streamType = Per::AsString(stype);
     rotoStrmName = "beam." + streamType;

   }
   else {
     // assume it came from the DCS
     Per::EStreamType stype = Per::kDcsMonitor;
     
     TIter blkiter =
       (const_cast<RawRecord*>(rawrec))->GetRawBlockIter();
     TObject*            tobj = 0;
     const RawDataBlock* rb   = 0;
     
     Bool_t sawAlarm = false;
     
     while ( (tobj = blkiter()) ) {
       rb = dynamic_cast<RawDataBlock*>(tobj);
       
       if ((dynamic_cast<const RawDcsAlarmBlock*>(rb))) 
         sawAlarm = true;
       
     }
     if (sawAlarm) stype = Per::kDcsAlarm;
     
     string streamType = Per::AsString(stype);
     rotoStrmName = "dcs." + streamType;

   }

   return rotoStrmName;
}

//_____________________________________________________________________________
Int_t RotoServer::ProcessConfig() 
{
//
//  Purpose:  Process the config that is in the buffer
//            
//  Argument: (none)
//
//  Return:   if 0 then config processing was uneventful
//
//  Contact:  R. Hatcher
//   

   char config[1024];
   char stream[1024];
   int  nItems = 0;
   bool ok = true;

   //   nItems = sscanf(fRecvBuffer.GetArray(),
   //                   "config autosave: %s = %d rec %d sec",
   //                   stream,&nrec,&nsec);

   nItems = sscanf(fRecvBuffer.GetArray(),
                   "config %s:",config);

   if (!strncmp(config,"autosave",8)) {
     unsigned int nrec, nsec;
     nItems = sscanf(fRecvBuffer.GetArray(),
                     "config autosave: %s = %u rec %u sec",
                     stream,&nrec,&nsec);

     if (nItems != 3) ok = false;
     else {
       int nmod = SetAutoSaveConfig(stream,nrec,nsec);
       if (nmod < 1) {
         logWarn("ProcessConfig stream \"%s\" unknown\n",stream);
       }
     }

   }
   else if (!strncmp(config,"compress",8)) {
     int level;
     nItems = sscanf(fRecvBuffer.GetArray(),
                     "config compress: %s = %d",
                     stream,&level);

     if (nItems != 2) ok = false;
     else {
       int nmod = SetCompressConfig(stream,level);
       if (nmod < 1) {
         logWarn("ProcessConfig stream \"%s\" unknown\n",stream);
       }
     }
   }
   else if (!strncmp(config,"basketsize",10)) {
     int basketsize;
     nItems = sscanf(fRecvBuffer.GetArray(),
                     "config basketsize: %s = %d",
                     stream,&basketsize);

     if (nItems != 2) ok = false;
     else {
       int nmod = SetBasketSizeConfig(stream,basketsize);
       if (nmod < 1) {
         logWarn("ProcessConfig stream \"%s\" unknown\n",stream);
       }
     }

   }
   else ok = false;

   if (!ok) 
     logWarn("ProcessConfig failed to interpret string: \"%s\"\n",
             fRecvBuffer.GetArray());

   return 0;
}

//_____________________________________________________________________________
void RotoServer::BuildStateReport() 
{
//
//  Purpose:  Update state report string
//
//  Argument: (none)
//
//  Return:   (none) [fStateReport is modified]
//
//  Contact:  R. Hatcher
//   
   fStateReport  = RotoRcCmd::ElementAsStlString(MINOS_ROOTER_ROOTER);
   fStateReport += "; ";
   fStateReport += 
      RotoRcCmd::InstrAsStlString(MINOS_ROOTER_STATE_REPORT,fCurrentState);
   fStateReport += "; ";
}

//_____________________________________________________________________________

---