IoInputModule.cxx

Go to the documentation of this file.

// $Id: IoInputModule.cxx,v 1.83 2007/09/21 19:49:05 buckley Exp $
//
// Job control interface to input data streams
//
// messier@huhepl.harvard.edu
#include "TSystem.h"
#include "TRegexp.h"

#include "Dispatcher/DDS.h"
#include "IoModules/IoInputModule.h"
#include <cassert>
#include "MessageService/MsgService.h"
#include "MinosObjectMap/MomNavigator.h"
#include "IoModules/IoDataStreamItr.h"
#include "IoModules/IoDDSStreamItr.h"
#include "IoModules/IoDataStreamFactory.h"
#include "JobControl/JobCInputModule.h"
#include "JobControl/JobCModuleRegistry.h"
#include "JobControl/JobCEnv.h"
#include "RawData/RawRecord.h"
#include "RawData/RawDaqSnarlHeader.h"
#include "CandData/CandHeader.h"
#include "Record/RecRecord.h"
#include "Record/RecPhysicsHeader.h"
#include "Registry/Registry.h"
#include "Validity/VldContext.h"
#include "Validity/VldTimeStamp.h"
#include "Util/UtilString.h"
#include "TSystem.h"

#include <algorithm>
#include <cstring>
#include <string>
#include <map>

#ifdef SITE_HAS_SAM
#include "sam_cpp_api/SamConsumer.hpp"
#endif

CVSID("$Id: IoInputModule.cxx,v 1.83 2007/09/21 19:49:05 buckley Exp $");
JOBMODULE(IoInputModule,"INPUT","Read and configure input streams");

typedef std::map<std::string,std::string>::const_iterator mapStrStrItr_t;

// A result that says "AOK"
static const JobCResult gsAllClear = JobCResult::kAOK;

// Many of the input methods are recursive. Ie. try to load an event,
// if that fails, try the next file and try again, etc. To prevent
// initializations from being done twice we need a way to keep track
// of the recursion level. This little class helps with that.
class CallDepth {
public:
  CallDepth()  { ++fsDepth; }
  ~CallDepth() { --fsDepth; }
  static int fsDepth;
};
int CallDepth::fsDepth = 0;

//......................................................................

IoInputModule::IoInputModule() : 
  fDataStreamItr(0),
  fFormat(""),
  fStreamList(""),
  fServer(""),
  fPort(0),
  fTimeOut(0),
  fDataSource(0),
  fKeepUpMode(0),
  fMaxSyncDelay(0),
  fOffLine(false),
  fMaxRetry(0),
  fRetryDelay(1),
  fClientType(DDS::kUnknownClientType),
  fClientName(""),
  fStatus(JobCResult::kAOK),
  fLastRun(-1),
  fLastSnarl(-1),
  fCurrentRun(-1),
  fCurrentSnarl(-1),
  fLoadedCommandLineFiles(false)
#ifdef SITE_HAS_SAM
  ,fsamProject(0)
#endif
{ fStopwatch.Reset(); fStopwatch.Stop(); }

//......................................................................

IoInputModule::~IoInputModule() 
{ 
  if ( fDataStreamItr ) { delete fDataStreamItr; fDataStreamItr = 0; }
}

//......................................................................

void IoInputModule::BeginJob() 
{
  this->LoadFilesFromCommandLine();
  // Delay opening files until first action (Next,Prev,etc.) call
}


//......................................................................

void IoInputModule::EndJob()
{
  fStopwatch.Stop();
  MSG("Io",Msg::kDebug) << "IoInputModule::EndJob, Time(sec), Real " 
        << fStopwatch.RealTime() << ", CPU "
        << fStopwatch.CpuTime() << endl;
  
}

//......................................................................

const Registry& IoInputModule::DefaultConfig() const 
{
//======================================================================
// Get the default configuration for this module
//======================================================================
  static Registry r;
  r.SetName("INPUT.config");

  r.UnLockValues();

  MSG("Io",Msg::kDebug) << "Loading default config\n";

  // Stream config
  r.Set("Format" ,"input");
  r.Set("Streams","DaqMonitor,DaqSnarl,LightInjection");
  // r.Set("Streams","DaqMonitor,DaqSnarl,LightInjection,DcsMonitor,DcsAlarm");
  
  // DDS config
  r.Set("DDSServer", "daqdds.minos-soudan.org");
  r.Set("DDSPort",    DDS::kPort);
  r.Set("DDSTimeOut",  120);
  r.Set("DDSDataSource","Daq");
  r.Set("DDSKeepUpMode", "FileKeepUp");
  r.Set("DDSMaxSyncDelay",15);
  r.Set("DDSOffLine",false);
  r.Set("DDSMaxRetry",0);
  r.Set("DDSRetryDelay",1);
  r.Set("DDSClientType","Unknown");
  r.Set("DDSClientName","");

#ifdef SITE_HAS_SAM

  // SAM config
  r.Set("Station","minos");
  r.Set("SnapShotVers",0);
  r.Set("WorkGroupName","minos");
  r.Set("ApplicationName","loon");
  r.Set("ApplicationVers","dev");
  r.Set("MaxNumberOfFiles",0);
  r.Set("StartNewProject",1);

  // Create default project name
  // Get $USER
  const char* username = gSystem->Getenv("USER");
  if (!username) username = "unknown";
  r.Set("ProjectName",username);

#endif

  r.LockValues();
  return r;
}

//......................................................................

void IoInputModule::Config(const Registry& r) 
{
//======================================================================
// Configure the module based on the contents of the registry r
//======================================================================
  const char* tmps;
  int         tmpi;
  int         tmpb;  // bools

  MSG("Io",Msg::kDebug) << "Config IoInputModule with r=" << r << "\n";

  // Input data stream configuration
  bool doFormatConfig = false;
  if (r.Get("Format", tmps)) { fFormat     = tmps; doFormatConfig = true; }
  if (doFormatConfig) this->UpdateFormatConfig();
  bool doStreamConfig = false;
  if (r.Get("Streams",tmps)) { fStreamList = tmps; doStreamConfig = true; }
  if (doStreamConfig) this->UpdateStreamConfig();
    
  // DDS options
  bool doDDSConfig = false;
  if (r.Get("DDSServer", tmps)) { fServer  = tmps; doDDSConfig = true; }
  if (r.Get("DDSPort",   tmpi)) { fPort    = tmpi; doDDSConfig = true; }
  if (r.Get("DDSTimeOut",tmpi)) { fTimeOut = tmpi; doDDSConfig = true; }
  if (r.Get("DDSClientType",tmps)) { fClientType = DDS::GetClientType(tmps); 
                                     doDDSConfig = true; }
  if (r.Get("DDSClientName",tmps)) { fClientName = tmps; 
                                     doDDSConfig = true; }
  if (r.Get("DDSDataSource",tmps)){fDataSource = DDS::GetDataSourceCode(tmps);
                                   doDDSConfig = true;}
  if (r.Get("DDSKeepUpMode",tmps)) { fKeepUpMode = DDS::GetKeepUpCode(tmps); 
                                     doDDSConfig = true; }
  if (r.Get("DDSMaxSyncDelay",tmpi)){fMaxSyncDelay = tmpi; doDDSConfig = true;}
  if (r.Get("DDSOffLine",tmpb)) {fOffLine = tmpb; doDDSConfig = true;}
  if (r.Get("DDSMaxRetry",tmpi)) {fMaxRetry = tmpi; doDDSConfig = true; }
  if (r.Get("DDSRetryDelay",tmpi)) {fRetryDelay = tmpi; doDDSConfig = true; }
  if (doDDSConfig) this->UpdateDDSConfig();


}

//......................................................................  

JobCResult IoInputModule::Get()
{
//======================================================================
// Load the data records at the current position in the input stream
//======================================================================


  if ( fDataStreamItr==0 ) {
    if ( this->OpenStreamItr()==0 ) {
      MSG("Io",Msg::kWarning) << "Failed to open input stream itr." << endl;
      return fStatus;
    }
    return this->Get();
  }

  MSG("Io",Msg::kVerbose) << "IoInputModule::Get " << endl;

  fStopwatch.Start(false);
  MomNavigator* mom = this->GetMom();
  assert(mom);
  mom -> Clear(); // Moving on so clear contents of Mom

  int nrecord = fDataStreamItr->LoadRecords(mom);
  bool isDDS = (UtilString::ToUpper(fFormat) == "DDS");
  // special treatment required because dds doesn't separate advance from load
  if ( isDDS && !nrecord ) fStatus.SetEndOfInputStream();

  this->ReadHeader(); // sets beginrun/endrun beginfile/endfile fStatus bits
  if ( fStatus.EndOfInputStream() ) 
     { fStatus.SetEndFile(); fStatus.SetEndRun(); }
  MSG("Io",Msg::kVerbose) 
   << "IoInputModule::Get returning status " << fStatus << endl;
  fStopwatch.Stop();
  return fStatus;
}

//......................................................................

JobCResult IoInputModule::Next(int n)
{
//======================================================================
// Advance the position in the stream n record sets. Load the records
// at the last position
//======================================================================
  CallDepth d; // Keep track of the call depth
  
  // Set the input status to "all clear" since advancing
  if (d.fsDepth==1) fStatus = gsAllClear;

  MSG("Io",Msg::kVerbose) << "IoInputModule::Next " << n << endl;

  if ( fDataStreamItr==0 ) {
    if (this->OpenStreamItr()==0) {
      MSG("Io",Msg::kWarning) << "Failed to open input stream itr." << endl;
      return fStatus;
    }
    return this->Next(n);
  }

  fStopwatch.Start(false);
  MomNavigator* mom = this->GetMom();
  assert(mom);
  mom -> Clear(); // Moving on so clear contents of Mom

  // Advance the position in the input stream until we run out of
  // records and files
  int nstep = 0;
  int ndone = 0;
  int ntry  = 0;
  while ( ndone < n ) {
    ntry  = n - ndone;
    nstep = fDataStreamItr->Increment(ntry);

    if ( nstep < ntry ) {
      // Reached end of file, load next one
      fStatus |= this->NextFile(); 
      
      // If this is the end of the input stream, we're done.
      if ( fStatus.EndOfInputStream() ) {
        fStopwatch.Stop();
        return this->Get();
      }
    }
    ndone += nstep;
  }

  // Load the current event
  fStopwatch.Stop();
  return this->Get();
}

//......................................................................

JobCResult IoInputModule::Prev(int n) 
{
//======================================================================
// Back up n positions in the input data stream. Load the records at
// the current position
//======================================================================
  CallDepth d;

  // Set the input status to "all clear"
  if (d.fsDepth == 1) fStatus = gsAllClear;

  MSG("Io",Msg::kVerbose) << "IoInputModule::Prev " << n << endl;
  
  if (fDataStreamItr==0) {
    if (this->OpenStreamItr()==0) {
      MSG("Io",Msg::kWarning) << "Failed to open input stream itr." << endl;
      return fStatus;
    }
    return this->Prev(n);
  }

  // Back up the the position in the input stream until we run out of
  // records and files
  fStopwatch.Start(false);
  MomNavigator* mom = this->GetMom();
  assert(mom);
  mom -> Clear(); // Moving on so clear contents of Mom

  int nstep = 0;
  int ndone = 0;
  int ntry  = 0;
  while (ndone < n) {
    ntry = n - ndone;
    nstep = fDataStreamItr->Decrement(ntry);

    if (nstep < ntry) {
      // Reached start of file, load previous file.
      fStatus |= this->PrevFile();

      // If there is no previous file, then we're done.
      if ( fStatus.BeginOfInputStream() || fStatus.EndOfInputStream() ) {
        return this->Get(); // may be end if failed to open stream itr
        fStopwatch.Stop();
      }

      // Move the position to the end of the current file so we can
      // walk backwards over it
      fDataStreamItr->GoToEOF();
    }
    ndone += nstep;
  }

  // Load the current event
  fStopwatch.Stop();
  return this->Get();

}

//......................................................................

JobCResult IoInputModule::GoTo(int run, int snarl, int searchDir) 
{
  CallDepth d;
  
  if (d.fsDepth==1) fStatus = gsAllClear;

  if (fDataStreamItr==0) {
    if (this->OpenStreamItr()==0) {
      MSG("Io",Msg::kWarning) << "Failed to open input stream itr." << endl;
      return fStatus;
    }
    return this->GoTo(run,snarl,searchDir);
  }

  if ( run == fCurrentRun && snarl == fCurrentSnarl ) return this -> Get();

  int dir = searchDir;
  if (dir==0) {
    if      (run>fLastRun) { dir =  1; }
    else if (run<fLastRun || (run==fLastRun && fCurrentRun < 0)) { dir = -1; }
    else {
      if    (snarl>fLastSnarl) { dir =  1; }
      else                     { dir = -1; }
    }
  }

  // Move position in the stream looking for run/event number
  while ( 1 ) {
    if ( dir > 0 ) {
      this->Next();
      if (fCurrentRun>run) {
        MSG("Io",Msg::kWarning) << 
          "Went to run "<<fCurrentRun<<
          " without finding run="<<run<<" snarl="<<snarl<<"\n";
        return fStatus;
      }
      if (fCurrentRun==run && fCurrentSnarl>snarl) {
        MSG("Io",Msg::kWarning) << 
          "Went to run="<<fCurrentRun<<" snarl="<<fCurrentSnarl<<
          " without finding run="<<run<<" snarl="<<snarl<<"\n";
        return fStatus;
      }
    }
    if ( dir <0 ) {
      this->Prev();
      if (fCurrentRun<run) {
        MSG("Io",Msg::kWarning) << 
          "Went to run "<<fCurrentRun<<
          " without finding run="<<run<<" snarl="<<snarl<<"\n";
        return fStatus;
      }
      if (fCurrentRun==run && fCurrentSnarl<snarl) {
        MSG("Io",Msg::kWarning) << 
          "Went to run="<<fCurrentRun<<" snarl="<<fCurrentSnarl<<
          " without finding run="<<run<<" snarl="<<snarl<<"\n";
        return fStatus;
      }
    }
    // Check if we're done
    if (fCurrentRun == run && fCurrentSnarl == snarl) return fStatus;
    if (dir>0 && fStatus.EndOfInputStream())          return fStatus;
    if (dir<0 && fStatus.BeginOfInputStream())        return fStatus;
  }
  return fStatus;
}

//......................................................................

JobCResult IoInputModule::GoTo(const VldContext& vld)
{
//======================================================================
// Go to records that match validity context. If vld is not found, will
// GoTo record set one beyond requested validity.
//======================================================================
  CallDepth d;

  // Set the input status to "all clear"
  if (d.fsDepth==1) fStatus = gsAllClear;

  if (fDataStreamItr==0) {
    if (this->OpenStreamItr()==0) {
      MSG("Io",Msg::kWarning) << "Failed to open input stream itr." << endl;
      return fStatus;
    }
    return this->GoTo(vld);
  }

  fStatus |= fDataStreamItr -> GoTo(vld);

  // Load the current event
  return this->Get();

}

//......................................................................

void IoInputModule::List(const char* streamlist) const
{
//======================================================================
// Print list of files loaded
//======================================================================

  MsgStream& m = MSGSTREAM("Io",Msg::kInfo);

  m << "IoInputModule using data format " << fFormat << endl;

  if ( fDataStreamItr ) {
    fDataStreamItr -> ListFile(std::cout,streamlist);
    return;
  }

  // data stream not yet open, list files from fFileList
  m << "File Name\tStream List " << endl;
  m << "=========\t=========== " << endl;
  std::list<IoFileListItem>::const_iterator itr = fFileList.begin();
  std::list<IoFileListItem>::const_iterator itrEnd = fFileList.end();

  for ( ; itr != itrEnd; itr++ ) {
    m << *itr;
  }
 
}

//......................................................................

void IoInputModule::AddFile(const char *filepath, const char* streamlist, 
                                                                 int at) {
//======================================================================
// Add to the list of attached streams at the position "at". -1 = end
// of list
//======================================================================


  // Find out by checking the format of the filepath whether this
  // is a SAM job. Format will be SAM:samdataset

  const char *s1 = "SAM:";
  if ( strstr(filepath,s1) != NULL ) {

#ifdef SITE_HAS_SAM

  // SAM options

    const char* tmps;
    int tmpi;

    Registry& r = GetConfig();
    if (r.Get("Station",tmps)) {fStation = tmps;}
    if (r.Get("SnapShotVers",tmpi)){fSnapShotVers = tmpi;}
    if (r.Get("WorkGroupName",tmps)) {fWorkGroupName = tmps;}
    if (r.Get("ApplicationName",tmps)) {fApplicationName = tmps;}
    if (r.Get("ApplicationVers",tmps)) {fApplicationVers = tmps;}
    if (r.Get("ProjectName",tmps)) {fProjectName = tmps;}
    if (r.Get("MaxNumberOfFiles",tmpi)) {fMaxNumberOfFiles = tmpi;}
    if (r.Get("StartNewProject",tmpi)) {fStartNewProject = tmpi;}

    std::string projectname;

    // If this is a new project then append time-stamp
    if (fStartNewProject == 1) {
      // Construct full project name including timestamp
      // Get timestamp as a string
      VldTimeStamp ts;
      std::string timestamp = ts.AsString("lc");
      // Replace blank with "-"
      size_t pos = timestamp.find(" ");
      timestamp.replace(pos,1,"-");
      // Replace : with - as DbServer does not like : in project names
      pos = timestamp.find(":");
      while ( pos != string::npos ) {
        timestamp.replace(pos,1,"-");
        pos = timestamp.find(":",pos+1);
      }
      // Append timestamp to fProjectName taken from registry
      fProjectName.append("-");
      projectname = fProjectName+timestamp;
    }
    // Otherwise use project name that is supplied
    else if(fStartNewProject == 0) {
      projectname = fProjectName;
    }
        

    MSG("Io",Msg::kDebug) << "Sam Station " << fStation << " Snap Shot " <<
      fSnapShotVers << " Work Group Name " << fWorkGroupName 
                         << " Application Name "
      << fApplicationName << " Application Version " << fApplicationVers <<
      " Project Name " << projectname << endl;

    // Get the samdataset from filepath

    std::string temp = filepath;
    size_t pos = temp.find(":")+1;
    std::string samdataset = temp.substr(pos,temp.length()-pos);

    // Define snapshot version

    long snapshot;
    // Snapshot version = 0 means create New Snapshot
    if ( fSnapShotVers == 0 ) {
      snapshot = sam::SamProject::NewSnapshotVersion;
    }
    // Snapshot version < 0 means use last one created
    else if (fSnapShotVers < 0) {
      snapshot = sam::SamProject::LatestSnapshotVersion;
    }
    else if (fSnapShotVers > 0 ) {
      // Use specified SnapShot version 
      snapshot = fSnapShotVers;
    }
    
    MSG("Io",Msg::kDebug) << "SnapShot Version " << snapshot << endl; 

    if (fStartNewProject == 1) {

      // Create SAM project

      fsamProject = new sam::SamProject(projectname,fStation);

      // Start SAM project

      std::list<std::string> projectMasterArgList;

      try {
        MSG("Io",Msg::kInfo) << "Starting SAM Project " << projectname << 
          " on station " << fStation << endl;
        fsamProject->startProject(fWorkGroupName,samdataset,snapshot,
                                  projectMasterArgList);
      }
      catch(const sam::SamProject::StartProjectRequestRejected& ex) {
        MSG("Io",Msg::kInfo) << "Rejected start SAM project request " 
                             << ex << endl;
      }
    }

    // Start SAM consumer to deliver files

    const int projectMasterTimeout(60);
    const std::string processDescription("Loon Analysis Process");

    try{

      MSG("Io",Msg::kInfo) << "Starting SAM Consumer" << endl;

      sam::SamConsumer fsamConsumer(projectname,fStation,fWorkGroupName,
                                   fApplicationName,fApplicationVers,
                                   processDescription,
                                   fMaxNumberOfFiles,
                                   projectMasterTimeout);

      // Now get files. Format of returned files depends on whether
      // SAM cache is local disk, dcache disk or AFS file space
      
      std::map<std::string,std::string> filelist;
      map<std::string,std::string>::iterator fitr;
      
      int location;
      int length;
      int comp;
      std::string fileonly;
      std::string restOfPath;
      std::string afsroot("afsroot:");
      try {
        while(true) {
          std::string filename = fsamConsumer.getFile().getFullFileName();
          // The files need to be sorted as they come back in an undefined
          // order. Split them into a filename and the rest of the path. 
          // Put them in a map and then iterate over key which is filename 
          // - guarantees correct order.
          
          MSG("Io",Msg::kDebug) << "Filename " << filename << endl;
          location = filename.find_last_of("/");
          length = filename.length();
          fileonly = filename.substr(location+1,length-1);
          // Need to look for afsroot: at start of path. If it is there the
          // remove it and rest of path is AFS path.
          comp = filename.compare(0,8,afsroot);
          if (comp == 0 ) {
            restOfPath = filename.substr(8,location-8);
          }
          else {
            restOfPath = filename.substr(0,location);
          }
          
          restOfPath.append("/");
          filelist.insert(make_pair(fileonly,restOfPath));
          
          MSG("Io",Msg::kDebug) << "File Only " << fileonly << " Rest Of Path " 
                                << restOfPath << endl;
          
          // Release file
          
          fsamConsumer.releaseFile();
        }
      }
      catch(const sam::SamConsumer::EndOfFileStreamReached& ex) {
        MSG("Io",Msg::kDebug) << "End of File Stream reached" << endl;
      }

      // Got all files. Now need to add them to file list. Iterate over map
      
      std::string sfile;
      const char *samfile = 0;
      for (fitr = filelist.begin(); fitr != filelist.end(); fitr++) {
        sfile = (fitr->second+fitr->first);
        samfile = sfile.data();
        MSG("Io",Msg::kInfo) << "Adding File " << samfile << endl;
        // Add file to file list
        IoFileListItem iofile(samfile,at,streamlist);
        fFileList.push_back(iofile);
      }
    }
    catch(const sam::SamConsumer::InitializationError& ex) {
        MSG("Io",Msg::kInfo) << "Rejected start SAM Consumer request " 
                             << ex << endl;
    }


    if (fsamProject) {
      try {
        MSG("Io",Msg::kInfo) << "Requesting end of SAM project " << endl;
        fsamProject->endProject();
      }
      catch(const sam::SamProject::EndProjectRequestRejected& ex) {
        MSG("Io",Msg::kInfo) << "SAM Project end request rejected "<< ex << endl;
        
      }
      catch(const sam::SamProject::EndProjectRequestFailed& ex) {
        MSG("Io",Msg::kInfo) << "SAM Project end request failed "<< ex << endl;
      }
    }


#endif     // End of ifdef SITE_HAS_SAM

  }
  else {

    // Add file to file list
    IoFileListItem iofile(filepath,at,streamlist);
    fFileList.push_back(iofile);
    
    if ( !fDataStreamItr ) return;
    
    // Add files to stream managed lists
    const IoFileListItem::FileStreamMap& filemap = iofile.GetFileStreamMap();
    
    if ( at < 0 ) {
      IoFileListItem::FileStreamMapConstItr itr = filemap.begin();
      for ( ; itr != filemap.end(); itr++ ) {
        std::string filename = itr -> first;
        std::string streamlist = itr -> second;
        fDataStreamItr -> AddFile(filename.c_str(),at,streamlist.c_str());
      }
    }
    else {
      // Apply files in reverse to have first file in list inserted at pos At
      IoFileListItem::FileStreamMap::const_reverse_iterator itr=filemap.rbegin();
      for ( ; itr != filemap.rend(); itr++ ) {
        std::string filename = itr -> first;
        std::string streamlist = itr -> second;
        fDataStreamItr -> AddFile(filename.c_str(),at,streamlist.c_str());
      }      
    }
  }
}

//......................................................................

void IoInputModule::RemoveFile(const char* filename, const char* streamlist) {
//======================================================================
// Remove the file "filename" from the list of input data files. 
//======================================================================

  if ( fDataStreamItr ) fDataStreamItr -> RemoveFile(filename,streamlist);

  std::string f(filename);
  std::list<IoFileListItem>::iterator  itr = fFileList.end();
  while ( !fFileList.empty() && itr != fFileList.begin() ) {
    itr--;
    IoFileListItem& iofile = *itr;
    iofile.RemoveFile(filename,streamlist);
    if ( iofile.GetNumFile() == 0 ) fFileList.erase(itr);
  }

  return;

}

//......................................................................

JobCResult IoInputModule::NextFile(int n, const char* streamlist)
{
//======================================================================
// Move to the next file in the list (move by n positions)
//======================================================================
  CallDepth d;

  if (d.fsDepth==1) fStatus = gsAllClear;

  if (fDataStreamItr==0) {
    if (this->OpenStreamItr()==0) {
      MSG("Io",Msg::kWarning) << "Failed to open input stream itr." << endl;
      return fStatus;
    }
    return this->NextFile(n,streamlist);
  }    

  fStatus |= fDataStreamItr -> NextFile(n,streamlist);

  MSG("Io",Msg::kDebug)
      << "status is " << fStatus
      << " current file is " << fDataStreamItr->GetCurrentFile() << endl;

  return fStatus;

}

//......................................................................

JobCResult IoInputModule::PrevFile(int n, const char* streamlist) 
{
//======================================================================
// Move to the previous list in the file (move back by n files)
//======================================================================
  CallDepth d;
  
  if (d.fsDepth==1) fStatus = gsAllClear;

  if (fDataStreamItr==0) {
    if (this->OpenStreamItr()==0) {
      MSG("Io",Msg::kWarning) << "Failed to open input stream itr." << endl;
      return fStatus;
    }
    return this->PrevFile(n,streamlist);
  }    

  fStatus |= fDataStreamItr -> PrevFile(n,streamlist);

  return fStatus;

}

//......................................................................

JobCResult IoInputModule::GoToFile(int n, const char* streamlist) 
{
//======================================================================
// Move the stream to the nth file in the list (n=0 is first)
//======================================================================
  CallDepth d;
  
  if (d.fsDepth==1) fStatus = gsAllClear;

  if (fDataStreamItr==0) {
    if (this->OpenStreamItr()==0) {
      MSG("Io",Msg::kWarning) << "Failed to open input stream itr." << endl;
      return fStatus;
    }
    return this->GoToFile(n,streamlist);
  }    

  fStatus |= fDataStreamItr -> GoToFile(n,streamlist);

  return fStatus;

}

//......................................................................

JobCResult IoInputModule::GoToFile(const char* filename, const char*streamlist){
//======================================================================
// Move the stream to a named file
//======================================================================
  CallDepth d;
  
  if (d.fsDepth==1) fStatus = gsAllClear;

  if (fDataStreamItr==0) {
    if (this->OpenStreamItr()==0) {
      MSG("Io",Msg::kWarning) << "Failed to open input stream itr." << endl;
      return fStatus;
    }
    return this->GoToFile(filename,streamlist);
  }    

  fStatus |= fDataStreamItr -> GoToFile(filename,streamlist);

  return fStatus;
}

//......................................................................  

void IoInputModule::Select(const char* stream, const char* select, 
                           bool isRequired)
{
//======================================================================
// Add/Change the selection cuts for a stream
//======================================================================
  // Insert the selection into the map
  fStreamSelectionMap[stream] = select;
  fStreamRequiredMap[stream] = isRequired;
  
  // Pass the info on to the data stream
  if (fDataStreamItr) {
    fDataStreamItr->Select(stream, select,isRequired);
  }

}

//......................................................................  

void IoInputModule::DefineStream(const char* stream, const char* tree) {
//======================================================================
// Define stream to serve specified tree
//======================================================================
  // Insert the definition into the map
  fStreamDefMap[stream] = tree;

  // Pass the info on to the data stream
  if ( fDataStreamItr ) {
    fDataStreamItr->DefineStream(stream, tree);
  }

}

//......................................................................  

void IoInputModule::SetSequenceMode(const char* stream,
                                    Per::ESequenceMode sequenceMode) {
//======================================================================
// Define stream sequence mode
//======================================================================
  // Insert the sequence mode into the map
  fStreamSeqModeMap[stream] = sequenceMode;

  // Pass the info on to the data stream
  if ( fDataStreamItr ) {
    fDataStreamItr->SetSequenceMode(stream, sequenceMode);
  }

}

//......................................................................  

void IoInputModule::SetTestMode(const char* stream,
                                bool testMode) {
//======================================================================
// Define stream test mode
//======================================================================
  // Insert the test mode into the map
  fStreamTestModeMap[stream] = testMode;

  // Pass the info on to the data stream
  if ( fDataStreamItr ) {
    fDataStreamItr->SetTestMode(stream, testMode);
  }

}

//......................................................................

void IoInputModule::SetWindow(const char* stream, double lower, double upper)
{
//======================================================================
// Define stream window if kWindow sequence mode is used
//======================================================================
  fStreamWindowMap[stream] = std::pair<double,double>(lower,upper);

  // Pass the info on to the data stream
  if ( fDataStreamItr ) {
      fDataStreamItr->SetWindow(stream, lower, upper);
  }

}

//......................................................................

void IoInputModule::SetMaxFileRepeat(const char* stream, int numRepeat)
{
//======================================================================
// Define maximum number of times to reuse a file in the stream before
// loading the next one; for kSequential and kRandom sequence modes
//======================================================================
  fStreamMaxRepeatMap[stream] = numRepeat;

  // Pass it on to the data stream
  if ( fDataStreamItr ) fDataStreamItr->SetMaxFileRepeat(stream,numRepeat);
}

//......................................................................

void IoInputModule::SetMeanMom(const char* stream, double mean)
{
//======================================================================
// Define mean number of events to push to mom for this stream
// for kSequential and kRandom sequence modes
//======================================================================
  fStreamMeanMap[stream] = mean;

  // Pass it on to the data stream
  if ( fDataStreamItr ) fDataStreamItr->SetMeanMom(stream,mean);
}

//......................................................................

void IoInputModule::SetPushRandom(const char* stream, bool setRandom)
{
//======================================================================
// Define whether to push a random or constant number of events to mom
// for this stream for kSequential and kRandom sequence modes
//======================================================================
  fStreamPushRandomMap[stream] = setRandom;

  // Pass it on to the data stream
  if ( fDataStreamItr ) fDataStreamItr->SetPushRandom(stream,setRandom);
}

//......................................................................

void IoInputModule::SetRandomSeed(int rSeed)
{
//======================================================================
// Set the random seed for SetPushRandom(stream,true) case
// for kSequential and kRandom sequence modes
//======================================================================
  fRandomSeed = rSeed;

  // Pass it on to the data stream
  if ( fDataStreamItr ) fDataStreamItr->SetRandomSeed(rSeed);

}

//......................................................................

const char* IoInputModule::GetCurrentFile(const char* streamname) const
{
    MSG("Io",Msg::kDebug) << "IoInputModule::GetCurrentFile()" << endl;
    mapStrStrItr_t it, done=fCurrentFileMap.end();
    std::string strmstring = streamname;

    for (it = fCurrentFileMap.begin(); it !=done; ++it) {
        MSG("Io",Msg::kVerbose)
          << "stream: " << setw(16) << it->first
          << " file: " << it->second
          <<endl;
    }

    for (it = fCurrentFileMap.begin(); it!=done; ++it) {
      if ( strmstring == it->first ) return it->second.c_str();
    }
    // Sue's original approach
    if (!fDataStreamItr) return 0;
    return fDataStreamItr->GetCurrentFile(streamname);
}

const char* IoInputModule::GetLastFile(const char* streamname) const
{
    MSG("Io",Msg::kInfo) << "IoInputModule::GetLastFile()" << endl;
    mapStrStrItr_t it, done = fLastFileMap.end();
    std::string strmstring = streamname;

    for (it = fLastFileMap.begin(); it!=done; ++it) {
        MSG("Io",Msg::kVerbose)
          << "stream: " << setw(16) << it->first
          << " file: " << it->second
          <<endl;
    }

    for (it = fLastFileMap.begin(); it!=done; ++it) {
      if ( strmstring == it->first ) return it->second.c_str();
    }
    return 0;

}

//......................................................................

void IoInputModule::LoadFilesFromCommandLine()
{
//======================================================================
// Load the files listed on the program command line
//======================================================================
  JobCEnv& jce = JobCEnv::Instance();
  if (!fLoadedCommandLineFiles) {
    for (int i=0; i<jce.GetNfile(); ++i) {
      this->AddFile(jce.GetFileName(i));
    }
    fLoadedCommandLineFiles = true;
  }
}

//......................................................................

int IoInputModule::ReadHeader()
{
//======================================================================
// Read temptags to get file name
// Read header information to get run/snarl info. 
//   if found Run and Snarl info return 2, 
//   if only Run info return 1,
//   else return 0
//======================================================================
  const MomNavigator* mom = this->GetMom();
  assert(mom);

  // BeginFile/EndFile boundaries may not be in synch across the different
  // data streams.  The definition used here is to set file boundary true if 
  // the file has changed for any of the managed streams.

  const TObjArray* momarray = mom->GetFragmentArray();
  for (int i=0; i < momarray->GetEntriesFast(); ++i) {
    TObject* obj = momarray->At(i);
    if (!obj) continue;
    Registry* temptags = 0;
    if ( RecMinos* record = dynamic_cast<RecMinos*>(obj) ) {
      temptags = &(record->GetTempTags());
    }
    else if ( RecRecord* record = dynamic_cast<RecRecord*>(obj) ) {
      temptags = &(record->GetTempTags());
    }
    if ( ! temptags ) continue;

    // on a named stream?
    const char* tagstream = 0;
    if ( ! temptags->Get("stream",tagstream) ) continue;

    // stream managed by i/o?
    std::string streamname(tagstream);
    const char* tagnewfile = 0;
    if ( ! temptags->Get("file",tagnewfile) ) continue; 

    std::string lstfilename = fLastFileMap[streamname];
    std::string curfilename = fCurrentFileMap[streamname];
    std::string newfilename(tagnewfile);

    if ( newfilename != curfilename ) {
            
      std::string starcur  = fCurrentFileMap.begin()->first;
      std::string starlast = fLastFileMap.begin()->first;
      
      MSG("Io",Msg::kDebug) 
        << "SetBeginFile on streamname '" << streamname << "'" << endl
        << "   current      '" << fCurrentFileMap[streamname] << "'" << endl
        << "   last         '" << fLastFileMap[streamname] << "'" << endl
        << "   current['" << starcur << "'] '" 
        << fCurrentFileMap[starcur] << "'" << endl
        << "      last['" << starlast << "'] '" 
        << fLastFileMap[starlast] << "'" << endl
        << "   new '" << newfilename << "' != "
        << " cur '" << curfilename << "'" << endl
        << "   update \"*\" ? " 
        << (( newfilename != fCurrentFileMap["*"] ) ? "yes":"no")
        << endl;
      
      // update "*" stream first
      if ( newfilename != fCurrentFileMap["*"] ) {
        fStatus.SetBeginFile();
        //if ( lstfilename != "" ) fStatus.SetEndFile();
        if ( fCurrentFileMap["*"] != "" ) fStatus.SetEndFile();
        
        fLastFileMap["*"]           = fCurrentFileMap["*"];
        fCurrentFileMap["*"]        = newfilename;

        MSG("Io",Msg::kDebug) 
          << "SetBeginFile on '*'" << endl
          << "   current['" << starcur << "'] '" 
          << fCurrentFileMap[starcur] << "'" << endl
          << "      last['" << starlast << "'] '" 
          << fLastFileMap[starlast] << "'" << endl;
      }
      // update this named stream
      fLastFileMap[streamname]    = curfilename;
      fCurrentFileMap[streamname] = newfilename;
      
      // if a stream on a file moved on then presumably all the
      // other streams on the same file have also been exhausted
      // and are going to move on -- help them along so we don't
      // have to wait for that stream to be the next record on
      // that stream is the next VldContext
      mapStrStrItr_t it, done = fCurrentFileMap.end();
      for (it = fCurrentFileMap.begin(); it != done; ++it) {
        if ( it->second == curfilename ) {
          std::string altstream = it->first;
          fLastFileMap[altstream]    = curfilename;
          fCurrentFileMap[altstream] = newfilename;
        }
      }
      
    } // new != cur filename

    MSG("Io",Msg::kVerbose) 
      << " stream '" << streamname << "' set fLastFileMap to '"
      << curfilename << "', fCurrentFileMap to '"
      << newfilename << "'" 
      << " * '" << fLastFileMap["*"] << "'  '" << fCurrentFileMap["*"] << "'"
      << endl;
        
  } // loop over records

  // update for EOF/EOJ condition (which doesn't come throught this
  // function) if it isn't a file change
  // this won't work is the file has just one record
  // ...the whole procedure is fundamentally flawed -- it shouldn't be
  // based on what we find in "mom" but rather the stream/file 
  // management classes...
  if ( ! fStatus.BeginFile() ) fLastFileMap["*"] = fCurrentFileMap["*"];

  // BeginRun/EndRun
  int run   = -1;  // default and flag value
  int snarl = -1;  // default and flag value
  for (int i=0; i < momarray->GetEntriesFast(); ++i) {
    TObject* obj = momarray->At(i);
    if (!obj) continue;
    if ( RecMinos* record = dynamic_cast<RecMinos*>(obj) ) {
      // old style
      
        // all DAQ generated records can supply run #
      const RawDaqHeader* rdh 
             = dynamic_cast<const RawDaqHeader*>(record->GetHeader());
      if (rdh) run = rdh->GetRun();

      // but only DaqSnarl records can supply snarl #
      const RawDaqSnarlHeader* rdsh 
             = dynamic_cast<const RawDaqSnarlHeader*>(record->GetHeader());
      if (rdsh) snarl = rdsh->GetSnarl();

      if (!rdh) {
        // not a DAQ record, perhaps it's a CandRecord
        const CandHeader* candhdr
          = dynamic_cast<const CandHeader*>(record->GetHeader());
        if (candhdr) {
          run   = candhdr->GetRun();
          snarl = candhdr->GetSnarl();
        }
      }
    }
    else if ( RecRecord* record = dynamic_cast<RecRecord*>(obj) ) {
      // New style
      const RecPhysicsHeader* rph 
          = dynamic_cast<const RecPhysicsHeader*>(&(record->GetHeader()));
      if ( rph ) {
        run   = rph->GetRun();
        snarl = rph->GetSnarl();
      }
    }
    // break early only if determined snarl in case it one of those
    // crazy record sets with a DaqMonitor and a DaqSnarl record.
    if ( snarl >= 0 ) break;
  }

  // set the status flags based on what was extracted
  fCurrentSnarl = snarl;
  if ( run < 0 ) {
    fCurrentRun = -1;
    return 0;
  }
  if ( run != fCurrentRun ) {
    fStatus.SetBeginRun();
    if ( fLastRun >= 0 ) fStatus.SetEndRun();
  }
  fLastRun      = fCurrentRun;
  fCurrentRun   = run;
  if ( snarl >= 0 ) {
    fLastSnarl    = fCurrentSnarl;
    fCurrentSnarl = snarl;
    return 2;
  }
  return 1;
}

//......................................................................

void IoInputModule::UpdateDDSConfig() {
//======================================================================
// Update the dispatcher configuration
//======================================================================
  if ( fDataStreamItr == 0 ) return;

  IoDDSStreamItr* ddsItr = dynamic_cast<IoDDSStreamItr*>(fDataStreamItr);
  if ( ! ddsItr ) return;

  ddsItr->SetTimeOut(fTimeOut);  

  // Need to reinitialize dispatcher if server hostname, port, clienttype
  // or clientname have changed
  bool reinit = (fServer != ddsItr->GetSourceName() 
                || fPort != ddsItr->GetPort()
                || fClientType != ddsItr->GetClientType() 
                || fClientName != ddsItr->GetClientName() );
  
  if ( reinit ) {
    this -> CloseStreamItr();  // wait for next action to reopen
  }
  else {
    ddsItr -> SetKeepUpMode(fKeepUpMode);
    ddsItr -> SetMaxSyncDelay(fMaxSyncDelay);
    ddsItr -> SetDataSource(fDataSource);
    ddsItr -> SetOffLine(fOffLine);
  }

}


//......................................................................

void IoInputModule::UpdateFormatConfig() {
//======================================================================
// Update the stream itr to match requested format
//======================================================================

  if ( fDataStreamItr == 0 ) return;

  bool reopen = ( fFormat != fDataStreamItr->GetFormat() );
  if ( reopen ) {
    this -> CloseStreamItr();  // wait for next action to reopen
  }

  return;

}

//......................................................................

void IoInputModule::UpdateFileList() 
{
//======================================================================
// Update the file list 
//======================================================================

  if ( fDataStreamItr == 0 ) return;

  // Fresh start. This should typically only be called when data stream 
  // iterator is newly opened (i.e. when the format changes)
  fDataStreamItr -> RemoveFile("*");

  std::list<IoFileListItem>::iterator itr = fFileList.begin();
  for ( ; itr != fFileList.end(); itr++ ) {
    IoFileListItem& iofile = *itr;
    const IoFileListItem::FileStreamMap& filemap = iofile.GetFileStreamMap();
    int at = iofile.GetAt();
  
    if ( at < 0 ) {
      IoFileListItem::FileStreamMapConstItr itr = filemap.begin();
      for ( ; itr != filemap.end(); itr++ ) {
        std::string filename = itr -> first;
        std::string streamlist = itr -> second;
        fDataStreamItr -> AddFile(filename.c_str(),at,streamlist.c_str());
      }
    }
    else {
      // Apply files in reverse to have first file in list inserted at pos At
      IoFileListItem::FileStreamMap::const_reverse_iterator itr 
                                                         = filemap.rbegin();
      for ( ; itr != filemap.rend(); itr++ ) {
        std::string filename = itr -> first;
        std::string streamlist = itr -> second;
        fDataStreamItr -> AddFile(filename.c_str(),at,streamlist.c_str());
      }      
    }
  }

  return;
  
}

//......................................................................

void IoInputModule::UpdateStreamConfig()
{
//======================================================================
// Set the stream and selection cuts for the open streams
//======================================================================
  if (fDataStreamItr==0) return;

  // Define streams
  mapStrStrItr_t itr;
  for ( itr = fStreamDefMap.begin(); itr != fStreamDefMap.end(); itr++ ) {
    fDataStreamItr->DefineStream((itr->first).c_str(),  // Stream
                                 (itr->second).c_str());  // Definition
  }

  // Set the streams to be activated
  fDataStreamItr->Streams(fStreamList.c_str());
  // Set the selection cuts for each stream
  for (itr=fStreamSelectionMap.begin();itr!=fStreamSelectionMap.end();++itr) {
    std::map<std::string,bool>::const_iterator reqitr 
                           = fStreamRequiredMap.find(itr->first);
    bool isrequired = false;
    if ( reqitr != fStreamRequiredMap.end() ) isrequired = reqitr -> second;
    fDataStreamItr->Select((itr->first).c_str(),   // Stream
                           (itr->second).c_str(), // Selection
                           isrequired);  // IsRequired
  }

  // Set the test mode for each stream
  std::map<std::string,bool>::const_iterator testitr;
  for ( testitr=fStreamTestModeMap.begin(); testitr !=fStreamTestModeMap.end();
                                          ++testitr ) {
    fDataStreamItr->SetTestMode((testitr->first).c_str(),   // Stream
                                testitr->second);  // TestMode
  }

  // Set the sequence mode for each stream
  bool setRandom = false;
  std::map<std::string,Per::ESequenceMode>::const_iterator seqitr;
  for ( seqitr=fStreamSeqModeMap.begin(); seqitr!=fStreamSeqModeMap.end();
                                        ++seqitr ) {
    fDataStreamItr->SetSequenceMode((seqitr->first).c_str(),   // Stream
                                     seqitr->second); // Sequence Mode
    pair<double,double> window = fStreamWindowMap[seqitr->first];
    fDataStreamItr->SetWindow((seqitr->first).c_str(),   // Stream
                              window.first,window.second);
    if (seqitr->second == Per::kSequential ||
        seqitr->second == Per::kRandom      ) {
      if (!setRandom) {
        setRandom = true;
        fDataStreamItr->SetRandomSeed(fRandomSeed);
      }
      int    repeat = fStreamMaxRepeatMap[seqitr->first];
      fDataStreamItr->SetMaxFileRepeat( (seqitr->first).c_str(), repeat );
      double mean = fStreamMeanMap[seqitr->first];
      fDataStreamItr->SetMeanMom( (seqitr->first).c_str(), mean );
      bool   pushRand = fStreamPushRandomMap[seqitr->first];
      fDataStreamItr->SetPushRandom( (seqitr->first).c_str(), pushRand );
    } // end if kSeq or kRand
  }
}

//......................................................................

int IoInputModule::OpenStreamItr()
{
//======================================================================
// Open a new stream iterator
//======================================================================
  if (fDataStreamItr) this->CloseStreamItr();

  bool isDDS = (UtilString::ToUpper(fFormat) == "DDS");
  std::string src;
  if ( isDDS ) src = fServer;
  
  fDataStreamItr = IoDataStreamFactory::CreateDataStreamItr(src.c_str(),
                     fFormat.c_str(),fPort,fMaxRetry,fRetryDelay,
                     fClientType,fClientName);


  if (fDataStreamItr == 0) {
    MSG("Io",Msg::kWarning) << "Failed to open stream '" << src << "'" <<
      " using format '" << fFormat << "'" << endl; 
    fStatus.SetEndRun();
    fStatus.SetEndFile();
    fStatus.SetEndOfInputStream();
    return 0;
  }

  fStatus.SetBeginOfInputStream();
  fStatus.SetBeginFile();
  fStatus.SetBeginRun();

  // Configure the file and module
  this->UpdateStreamConfig(); // this should be called before filelist
  this->UpdateFileList();
  this->UpdateDDSConfig();
  fFormat = fDataStreamItr->GetFormat();
  
  MSG("Io",Msg::kDebug) 
    << "Opened stream itr of format " << fDataStreamItr->GetFormat() << endl;

  return 1;
}

//......................................................................

void IoInputModule::CloseStreamItr() 
{
//======================================================================
// Close the currently openned stream
//======================================================================
  if (fDataStreamItr) {
    MSG("Io",Msg::kDebug) 
      << "Close stream itr of format " << fDataStreamItr->GetFormat() << endl;
    delete fDataStreamItr;
    fDataStreamItr = 0;
    fStatus.SetEndRun();
    fStatus.SetEndFile();
    fStatus.SetEndOfInputStream();
  }
}

---