DemoClient.cc

Go to the documentation of this file.

#include "TIterator.h"
#include "TROOT.h"
#include "TSystem.h"
#include "MessageService/MsgService.h"
#include "MinosObjectMap/MomNavigator.h"
#include "RawData/RawRecord.h"
#include "Persistency/Per.h"
#include "Dispatcher/DDS.h"
#include "Dispatcher/DDSClient.h"
#include "Dispatcher/DDSSubscription.h"

TROOT root("DemoClient","MINOS Data Dispatcher System Demo Client");
CVSID("$Id: DemoClient.cc,v 1.21 2007/08/24 05:39:07 schubert Exp $");

//                                                                          //
// Data Distribution System demo client.  This program uses the DDSClient   //
// class to connect to the dispatcher server.                               //
//                                                                          //
// Arguments: serverhostname = host on which dispatcher parent server       //
//                             is running.  If not specified, the default   //
//                             is to attempt to connect to a parent server  //
//                             running on the same machine as this client.  //
//                                                                          //
//                                                                          //
// This demo illustrates the process of connecting to ddsparentserver,      //
// requesting data (which spawns the ddschildserver), submitting a          //
// subscription request and receiving data from the ddschildserver.         //
// At the end of the job, the connection between the client and childserver //
// is shutdown.                                                             //

int main(int argc, char **argv) {

  MsgStream& ms = MSGSTREAM("DDS",Msg::kInfo);

  ms << "This demo will attempt to set up 1 client connection\n"
     << "with the dispatcher server." << endl;

  const char* serverhostname;
  if (argc > 1) {
    serverhostname = argv[1];
    ms << "The user has requested that the client will attempt to connect\n"
       << "to a parent server running on host " << serverhostname 
       << ".\n" << endl;
  }
  else {
    // Determine the host name of the current machine
    serverhostname = gSystem -> HostName();
    ms << "Since the user has not specified the host of the parent server\n"          
       << "as an argument to this demo, the default is to assume that the\n"
       << "parent server runs on the local host " << serverhostname 
       << ".\n" << endl;
  }

  ms << "To successfully run this program to access data, the user\n"
     << "must first run the DemoDaq program in the Persistency package\n"
     << "to produce the file testdaq.root.\n"
     << "The environment variable DDS_DAQ_DATA_DIR should point to\n"
     << "the directory which contains this file, and the symbolic\n"
     << "link 'currentfile' should exist within this directory to\n"
     << "point to testdaq.root.\n" << endl;

  DDSClient* ddsclient = 0;

  while (1) {
    // This primitive while loop is used to attempt to reestablish connection
    // if initial connection goes down due to a socket error

    // Create a new DDSClient object connected to ther server running on
    // node "serverhostname" and port 9090.
    ddsclient = new DDSClient(serverhostname,9090,DDS::kData,DDS::kEventDisplay,"Control Room");

    // Check validity of connected socket before using it
    if (! ddsclient -> IsValid() ) {
      ms << "Error in creation of socket connected to server." << endl;
      delete ddsclient; ddsclient = 0;
      return 1;  // end of session
    }
    else {
      ms << "Successfully connected to dispatcher server.\n" 
         << ddsclient << endl;
    }

    // Subscribe to data source of interest (kDaq or kDcs, default is kDaq).
    // Currently (I may change this in future versions), you can only 
    // subscribe to one data source at a time.  If you wish to 
    // receive data from multiple sources, you need to set up multiple client
    // connections.  For a given data source however, there is no restriction
    // as to the number of data streams (root trees) you may subscribe to.
    ddsclient->GetSubscription()->SetDataSource(DDS::kDaq);

    // Subscribe to streams of interest
    ddsclient->GetSubscription()->SetStreams("DaqSnarl");

    // Set keepupmode if requesting mode other than default DDS::kFileKeepUp
    // keepupmode DDS::kAll will retrieve all files from the source directory
    // ending in ".root".  If the user has also set the symbolic link 
    // "currentfile" in this directory to a file, the file retrieval will
    // begin with "currentfile", otherwise it will start at the first file
    // in the directory.
    ddsclient->GetSubscription()->SetKeepUpMode(DDS::kRecordKeepUp);
  
    // Optionally add selection cut on one or more streams
    // ddsclient -> GetSubscription() ->SetSelection("DaqSnarl",
    //    "((RawDaqSnarlHeader*)fHeader)->GetSnarl() % 10 == 0 
    //  && ((RawDaqSnarlHeader*)fHeader)->GetTrigSrc()==4");
    // requests only every 10th DaqSnarl generated from trigger source 4.


    // Optionally specify desired synchronization delay (default = 15 sec)
    // Useful when reading data from multiple streams. 0 => no waiting for
    // streams with infrequent or no entries. syncDelay > 0 => wait until difference
    // between last visible timestamp on any tree and current vldtimestamp is
    // greater than syncDelay before giving up on stream(s) with infrequent or
    // no entries. Set very high for perfectly synchronized streams.
    // ddsclient -> GetSubscription() -> SetSyncDelay(15);


    // Submit the subscription to the ddschildserver and check to make sure that
    // it was received okay
    DDS::EMessageType msgrc = ddsclient->Subscribe();
    if (msgrc != DDS::kOk) {
      ms << "An error message " << DDS::AsString(msgrc) 
         << " was received from DDS::Subscribe." << endl;    
      return 1;
    }

    // The client's subscription was submitted successfully.  Note that at 
    // anytime the clients subscription can be changed and resubmitted.
  
    // Once the subscription is sent, the user can begin to receive records.
    // Create a MomNavigator pointer to hold the received records
    msgrc = DDS::kOk;
    Int_t nentry = 0;
    while (msgrc == DDS::kOk) {
      // For testing inactivity shutdown, requires ddsparentserver -t 60
      //if ( nentry == 100 ) {
      //  cout << "sleeping 100 seconds" << endl;
      //  gSystem->Sleep(100*1000);
      //}
      
      
      // 2nd argument to DDSClient::Next can be used to adjust waittime before
      // ddschildserver will return DDS::kTimeOut when no new entry (or file) 
      // is available. The default for this waittime is 120 seconds.  This is
      // adjusted to 10 seconds here to avoid a long wait at the end of the demo.
      // 3rd argument to DDSClient::Next is the number of record sets to
      // advanceby (default = 1).
      MomNavigator* mom = 0;
      UInt_t waittime = 120;
      //Illustration of how to advance to new file using the GoTo*File methods.
      //std::string gotofile;
      //msgrc  = ddsclient->GoToFile("F00001442_0000.mdaq.root",gotofile);
      //cout << "GoToFile " << DDS::AsString(msgrc) << " " << gotofile << endl;
      //msgrc  = ddsclient->GoToNextFile(gotofile);
      //cout <<"GoToNextFile "<< DDS::AsString(msgrc)<< " "<< gotofile << endl;
      //msgrc  = ddsclient->GoToFile("currentfile",gotofile);
      //cout <<"GoTo currentfile "<< DDS::AsString(msgrc)<<" "<<gotofile<<endl;

      msgrc = ddsclient -> Next(mom,waittime); 
      if (msgrc == DDS::kOk) {
        nentry++;
        if ( mom ) {
          // If you are subscribing to multiple streams, then you should
          // loop over the records stored in mom like this:
          TIter fiter = mom -> FragmentIter();
          RawRecord* rawrecord;
          while ( (rawrecord = dynamic_cast<RawRecord*>(fiter.Next())) ) {
            // To determine record source use the temptags, e.g.:
            ms << "Received rawrecord " << nentry << " from stream "
               << rawrecord -> GetTempTags().GetCharString("stream")
               << " tree " << rawrecord -> GetTempTags().GetCharString("tree")
               << " index " << rawrecord -> GetTempTags().GetInt("index")
               << " file " << rawrecord -> GetTempTags().GetCharString("file")
               << endl;
          }
          delete mom; mom = 0; // deleting mom record also deletes RawRecord
        } 
      }
    }

    ms << "\nAfter " << nentry 
       << " total entries retrieved, an error message:\n" 
       << DDS::AsString(msgrc)
       << " was returned from DDS::Next.\n" << endl;
    if (msgrc == DDS::kTimeoutNewFile) {
      ms << "A TimeoutNewFile will occur when the record retrieval\n"
         << "reaches the end of the currentfile, and no new file\n"
         << "becomes available within the adjustable waittime specified\n"
         << "by the client. (So if the client has only intended to\n"
         << "read a set number of files, this is a successful ending.)\n" 
         << endl;
    }
    else if (msgrc == DDS::kTimeoutNewRecord) {
      ms << "A TimeoutNewRecord will occur when the record retrieval\n"
         << "failed to detect a new record from the currentfile, but\n"
         << "the file has not yet been closed by the data generator.\n"
         << "This could be an indication that:\n"
         << "  i)the waittime specified by the client is not long enough, or\n"
         << " ii)the data generator has paused, or\n"
         << "iii)the data generator crashed before closing the file.\n"
         << endl;
    }
   
    //  if ( msgrc == DDS::kSocketError ) {
    //  ms << "The current DDSClient connection will be shutdown\n" 
    //     << " and a new connection will be attempted." << endl;
    //  if (ddsclient) delete ddsclient; ddsclient = 0;
    //}
    //else break;  // out of primitive while loop
    break;
  }
  

  ms << "The demo will now shutdown the connected client."
     << endl; 

  if ( ddsclient ) {
    ddsclient -> Shutdown();
    delete ddsclient; ddsclient = 0;
  }

  return 0;

}

---