ReadDispatcherModule.cxx

Go to the documentation of this file.

// 
// ReadDispatcherModule
//
// This module largely duplicates the functionality 
// doing dispatcher reads in the standard Input module.
//
// The point of this module is to allow read-aheads of live
// data so you can check if there is new data before calling
// JobCPath::Next().  This is good for viewers.
// 
//
// n.tagg1@physics.ox.ac.uk
#include "TriD/stat/ReadDispatcherModule.h"
#include "MessageService/MsgService.h"
#include "MinosObjectMap/MomNavigator.h"
#include "JobControl/JobCModuleRegistry.h" // For JOBMODULE macro

#include "TIterator.h"
#include "TROOT.h"
#include "TSystem.h"
#include "RawData/RawRecord.h"
#include "Persistency/Per.h"
#include "Dispatcher/DDS.h"
#include "Dispatcher/DDSClient.h"
#include "Dispatcher/DDSSubscription.h"
#include "DataUtil/DumpMom.h"
#include "RawData/RawRecord.h"
#include "RawData/RawDaqSnarlHeader.h"
#include "RawData/RawDigitDataBlock.h"
#include "RawData/RawBlockRegistry.h"
#include "RawData/RawBlockId.h"
#include "RawData/RawDigit.h"


JOBMODULE(ReadDispatcherModule, "ReadDispatcherModule",
          "ReadDispatcherModule");
CVSID("$Id: ReadDispatcherModule.cxx,v 1.3 2007/03/01 16:59:43 rhatcher Exp $");
//......................................................................

ReadDispatcherModule::ReadDispatcherModule()
{
  fClient = 0;
  fConnected = false;
  fSurrogateMom = 0;
}



//......................................................................
ReadDispatcherModule::~ReadDispatcherModule()
{
  if(fClient) delete fClient;
}


Int_t ReadDispatcherModule::ConnectToServer( void )
{
  // Connect.

  if(fClient) delete fClient;
  fConnected = false;

  // Create a new DDSClient object connected to ther server running on
  // node "serverhostname" and port 9090.
  fClient = new DDSClient(fDDSServer.c_str(),fDDSPort);
  
  // Check validity of connected socket before using it
  if (! fClient -> IsValid() ) {
    MSG("ReadDisp",Msg::kInfo) << "Error in creation of socket connected to server." << endl;
    delete fClient; fClient = 0;
    return 1;  // end of session
  }
  else {
    MSG("ReadDisp",Msg::kInfo) << "Successfully connected to dispatcher server.\n" 
       << fClient << endl;
  }
    
  // Subscribe to data source of interest (kDaq or kDcs, default is kDaq).
  fClient->GetSubscription()
    ->SetDataSource((DDS::EDataSource)DDS::GetDataSourceCode(fDDSDataSource.c_str()));
  
  // Subscribe to streams of interest
  fClient->GetSubscription()
    ->SetStreams(fStreams.c_str());
  // Keep up mode.
  fClient->GetSubscription()->
    SetKeepUpMode((DDS::EKeepUpMode)DDS::GetKeepUpCode(fDDSKeepUpMode.c_str()));
  
  // Selection, assumes DaqSnarl
  fClient->GetSubscription()->
    SetSelection(fStreams.c_str(),fSelectRule.c_str());
  
  // Submit the subscription to the ddschildserver and check to make sure that
  // it was received okay
  DDS::EMessageType msgrc = fClient->Subscribe();
  if (msgrc != DDS::kOk) {
    MSG("ReadDisp",Msg::kInfo) << "An error message " << DDS::AsString(msgrc) 
       << " was received from DDS::Subscribe." << endl;    
    return 1;
  }

  fConnected = true;
  return 0;
}

Bool_t ReadDispatcherModule::IsNewEventReady( void ) 
{
  // Query the server and get a new mom if available.
  // Hold this mom until the next read cycle.
  // If we are already holding a mom, throw it out
  // in favour of a new one.

  if(!fConnected) ConnectToServer();

  if(!fConnected) return false;

  DDS::EMessageType msgrc = DDS::kOk;

  MomNavigator* newMom = 0;
  msgrc = fClient->Next(newMom,fDDSTimeOut);
  if (msgrc == DDS::kOk ) {
    //cout << "Got Next reply." << endl;

    if( newMom ) {
      // Got something!
      //cout << "Got a mom." << endl;

      if(newMom->GetFragmentArray()->IsEmpty()){
        cout << "Empty mom." << endl;
        delete newMom;
      } else {
        //cout << "Good mom." << endl;
        // Got a real, filled mom.
        // Do we already have something in the queue?
        // If so, get rid of it.
        if( fSurrogateMom ) delete fSurrogateMom;
        fSurrogateMom = newMom;
        
        if(MsgService::Instance()->GetStream("TriD")->IsActive(Msg::kDebug))
          DataUtil::dump_mom(newMom,std::cout);
      
        return true;
      }
    }
    
  } else {
    if(msgrc != DDS::kTimeoutNewRecord) {
      // Failed to get something from server.
      MSG("ReadDisp",Msg::kInfo) << "Dispatcher Error " 
         << DDS::AsString(msgrc)
         << "  Will try reconnect." 
         << endl;
      
      fClient->Shutdown();
      delete fClient;
      fClient = 0;
      fConnected = false;
    }
  }
 
  if(fSurrogateMom) return true;
  return false;
}


//......................................................................
JobCResult ReadDispatcherModule::Get()
{
  //
  // Get next event, default fMom
  //
  
  // If this were to be a JobCInputModule...
  //return Get(GetMom());
  return JobCResult::kAOK;
}

//......................................................................
JobCResult ReadDispatcherModule::Get(MomNavigator* mom)
{
  //
  // Get next event.
  //
  // Queries dispatcher once before failing.
  // Uses a previously cached-but-unused event if available.
  //
  if(! fSurrogateMom) {
    // Dont have one ready.. go look again.
    IsNewEventReady();
  }

  if(! fSurrogateMom) return JobCResult::kFailed; // Give up.


  // Copy the surrogate.
  // This bit of code ripped out of DDSClient mercilessly.
  mom->Clear();
  TObjArray* objArray =const_cast<TObjArray*>
    (fSurrogateMom -> GetFragmentArray());
  Int_t nrecord = objArray -> GetEntries();
  for ( Int_t irec=0; irec < nrecord; irec++ ) {
    //record is removed fr  om fSurrogateMom to pass to usr's mom for ownership
    RecMinos* record = dynamic_cast<RecMinos*>(objArray->RemoveAt(irec));
    mom -> AdoptFragment(record); // new owner of record
  }
  delete fSurrogateMom; fSurrogateMom=0; // clears array and deletes temptags


  return JobCResult::kPassed;
}

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

const Registry& ReadDispatcherModule::DefaultConfig() const
{
//======================================================================
// Supply the default configuration for the module
//======================================================================
  static Registry r; // Default configuration for module

  // Set name of config
  std::string name = this->GetName();
  name += ".config.default";
  r.SetName(name.c_str());

  // Set values in configuration
  r.UnLockValues();
  r.Set("DDSServer",      "localhost");
  r.Set("DDSPort",         9090);
  r.Set("Streams",        "DaqSnarl");
  r.Set("DDSKeepUpMode",  "RecordKeepUp");
  r.Set("DDSDataSource",  "Daq");
  r.Set("DDSTimeOut",      1);
  r.Set("SelectRule",     "true");
  r.LockValues();

  return r;
}

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

void ReadDispatcherModule::Config(const Registry& r)
{
//======================================================================
// Configure the module given the Registry r
//======================================================================
  int    tmpi;
  const char* tmps;

  if (r.Get("DDSServer",tmps))       { fDDSServer = tmps; }
  if (r.Get("DDSPort",tmpi))         { fDDSPort   = tmpi; }
  if (r.Get("Streams",tmps))         { fStreams = tmps; }
  if (r.Get("DDSKeepUpMode",tmps))   { fDDSKeepUpMode = tmps; }
  if (r.Get("DDSDataSource",tmps))   { fDDSDataSource = tmps; }
  if (r.Get("DDSTimeOut",tmpi))      { fDDSTimeOut = tmpi; }
  if (r.Get("SelectRule",tmps))      { fSelectRule = tmps; }
}

---