IoDDSStreamItr.cxx

Go to the documentation of this file.

// $Id: IoDDSStreamItr.cxx,v 1.16 2007/08/24 05:37:46 schubert Exp $
//
// Read data from a dispatcher stream
//
// messier@huhepl.harvard.edu
// schubert@hep.umn.edu
#include "IoModules/IoDDSStreamItr.h"

#include "Dispatcher/DDSClient.h"
#include "Dispatcher/DDSSubscription.h"
#include "MessageService/MsgService.h"
#include "MinosObjectMap/MomNavigator.h"
#include "Persistency/Per.h"
#include "Validity/VldContext.h"

#include "TSystem.h"

CVSID("$Id: IoDDSStreamItr.cxx,v 1.16 2007/08/24 05:37:46 schubert Exp $");

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

JobCResult IoDDSStreamItr::GoToFile(const char* filename, 
                                    const char* /* streamlist */) {
//======================================================================
// Go to file filename. filename may be a symbolic link, e.g. "currentfile"
// will advance to the target of the symbolic link "currentfile" in the
// data source directory.  If filename is not absolute fullfilepathname,
// its path is assumed relative to the data source directory defined on
// the server.  If filename is missing in this directory, will position
// pointer to the next file in the alphanumerically sorted list in the
// data source directory.  Returns JobCResult::kAOK or JobCResult::kWarning
// if error return from DDS, or JobCResult::kEndOfInputStream if not
// connected.

  if ( !IsValid() ) return JobCResult::kEndOfInputStream; 

  if ( this -> IsModified() ) { // modified since last subscription
    bool isOk = this -> Subscribe(); 
    if ( !isOk ) {
      MSG("Io",Msg::kWarning)<<" Subscription submit to server failed."<< endl;
      return JobCResult::kWarning;
    }
  }

  std::string returnedfilename;
  DDS::EMessageType msgrc = fDDSClient->GoToFile(filename,returnedfilename);
  if (msgrc != DDS::kOk) { 
    MSG("Io",Msg::kWarning) << 
      "An error message " << DDS::AsString(msgrc) << 
      " was received from DDSClient::GoToFile " << filename << "." << endl;
    return JobCResult::kWarning;
  }

  return JobCResult::kAOK;

}

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

JobCResult IoDDSStreamItr::NextFile(int n, const char* /* streamlist */) {
//======================================================================
// Advance by n files in data source directory. Returns JobCResult::kAOK
// or JobCResult::kWarning if error returned from DDS, or
// JobCResult::kEndOfInputStream if not connected.
//======================================================================

  if ( !IsValid() ) return JobCResult::kEndOfInputStream;

  if ( this -> IsModified() ) { // modified since last subscription
    bool isOk = this -> Subscribe(); 
    if ( !isOk ) {
      MSG("Io",Msg::kWarning)<<" Subscription submit to server failed."<< endl;
      return JobCResult::kWarning;
    }
  }

  for ( int i = 0; i < n; i++ ) {
    std::string returnedfilename;
    DDS::EMessageType msgrc = fDDSClient->GoToNextFile(returnedfilename);
    if (msgrc != DDS::kOk) {
      MSG("Io",Msg::kWarning) << 
        "An error message " << DDS::AsString(msgrc) << 
        " was received from DDSClient::GoToNextFile." << endl;
      return JobCResult::kWarning;
    }
  }

  return JobCResult::kAOK;

}

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

void IoDDSStreamItr::AddFile(const char* /* filename */, int /* at */,
                            const char* /* streamlist */) {
  return;
}

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

void IoDDSStreamItr::RemoveFile(const char* /* filename */, 
                               const char* /* streamlist */) {
  return;
}

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

JobCResult IoDDSStreamItr::PrevFile(int /* n */, const char* /* streamlist */){
  MSG("Io",Msg::kWarning) << " DDS Stream does not support PrevFile" << endl;
  return JobCResult::kWarning;
}

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

const char* IoDDSStreamItr::GetCurrentFile(const char* /* streamlist */) const{
  MSG("Io",Msg::kWarning) << " DDS Stream does not support GetCurrentFile" 
                          << endl;
  return "";
}

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

JobCResult IoDDSStreamItr::GoToFile(int /* n */, 
                            const char* /* streamlist */) {
  MSG("Io",Msg::kWarning) << " DDS Stream does not support GoToFile n" << endl;
  return JobCResult::kWarning;
}

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

std::ostream& IoDDSStreamItr::ListFile(std::ostream& os, 
                            const char* /* streamlist */) const
{
  return os;
}

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

IoDDSStreamItr::IoDDSStreamItr(const char* server, unsigned int port,
                               unsigned int nretry, unsigned int delay,
                               DDS::EClientType clienttype,
                               string clientname) :
  fPort(port),fTimeOut(120),fMaxRetry(nretry),fRetryDelay(delay),
  fDDSClient(0),fModified(true),fClientType(clienttype),
  fClientName(clientname) 
{
  this->SetSourceName(server);
  this->SetClientType(clienttype);
  this->SetClientName(clientname);
  this->InitDDSClient();
}

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

IoDDSStreamItr::~IoDDSStreamItr() 
{
  this->ShutdownDDSClient();
}

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

void IoDDSStreamItr::InitDDSClient() 
{
//======================================================================
// Set up the data dispatcher client
//======================================================================

  unsigned int attempt = fMaxRetry+1;
  fDDSClient = 0;
  while (!fDDSClient && attempt) {
    fDDSClient = new DDSClient(this->GetSourceName(),fPort,DDS::kData,
                               fClientType,fClientName);
    --attempt;

    if (fDDSClient==0 || !fDDSClient->IsValid()) {
      // Warning here...
      MSG("Io",Msg::kError)
        << "Unable to create DDSClient to host " << this->GetSourceName()
        << " on port " << fPort << "," << endl
        << "   will attempt " << attempt << " more times." << endl; 
      this->ShutdownDDSClient();
      gSystem->Sleep(1000*fRetryDelay);
    }
  }
  fModified = true;
  
}

//......................................................................
int IoDDSStreamItr::Subscribe() 
{
//======================================================================
// Submit subscription to server.  Returns 1 if successful, else 0.
//======================================================================
  int isOk = 0;
  if ( !IsValid() ) return isOk;

  DDS::EMessageType msgrc = fDDSClient->Subscribe();
  if (msgrc != DDS::kOk) {
    MSG("Io",Msg::kError) << 
      "An error message " << DDS::AsString(msgrc) << 
      " was received from DDS::Subscribe." <<
      "( host " << this->GetSourceName() << 
      ", port " << fPort << ")." << endl;
    this->ShutdownDDSClient();
  }
  else {
    isOk = 1;
    fModified = false;
  }

  return isOk;

}

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

void IoDDSStreamItr::ShutdownDDSClient()
{
  if (fDDSClient) { delete fDDSClient; fDDSClient = 0; }
}

//......................................................................
  
int IoDDSStreamItr::LoadRecords(MomNavigator* mom) 
{
//======================================================================
// Load the record at the current position in the stream
//======================================================================

  if ( this -> IsModified() ) { // modified since last subscription
    bool isOk = this -> Subscribe(); 
    if ( !isOk ) return 0;
  }

  // DDSClient only has a Next
  DDS::EMessageType rtn = fDDSClient->Next(mom, fTimeOut, 1);
  if (rtn != DDS::kOk) {
    MSG("Io",Msg::kWarning) << "DDSClient::Next returned error message: "
                            << DDS::AsString(rtn) << "." << endl;
    return 0;
  }
  
  return 1;
}

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

int IoDDSStreamItr::Increment(int n, MomNavigator* m)
{
//======================================================================
// Advance the position in the stream n spaces.
//=====================================================================
  if (n==0) return 0;

  if ( this -> IsModified() ) { // modified since last subscription
    bool isOk = this -> Subscribe(); 
    if ( !isOk ) return 0;
  }
  
  // Advance by n-1 since Get() is implemented by Next()...
  if (n==1) return 1;

  // For advances beyond just one...
  MomNavigator* mtmp = 0;
  bool dodel = false;
  if (m==0) { mtmp = new MomNavigator; dodel = true; }
  else      { mtmp = m; }

  DDS::EMessageType rtn = fDDSClient->Next(mtmp, fTimeOut, n-1);

  if (dodel) { delete mtmp; mtmp = 0; }

  if (rtn != DDS::kOk) {
    MSG("Io",Msg::kWarning) << "DDSClient::Next returned error message: "
                            << DDS::AsString(rtn) << "." << endl;
    return 0;
  }

  return n;
}

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

int IoDDSStreamItr::Decrement(int n, MomNavigator* /* mom */)
{
  MSG("Io",Msg::kWarning) << 
    "DDS Stream does not support Decrement(" << n << ").\n";
  return 0;
}

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

JobCResult IoDDSStreamItr::GoTo(const VldContext& vld, MomNavigator* /* mom */)
{
  MSG("Io",Msg::kWarning) << 
    "DDS Stream does not support GoTo(" << vld << ").\n";
  return JobCResult::kWarning;
}

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

int IoDDSStreamItr::GoToEOF() 
{
  MSG("Io",Msg::kWarning) << 
    "DDS Stream does not support GoToEOF().\n";
  return 0;
}

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

int IoDDSStreamItr::SetMaxSyncDelay(unsigned int maxSyncDelay) 
{ 
  // Subscription is modified with the new max sync delay but is not sent
  // until the IoDDSStreamItr::Subscribe method is invoked.
  int isOpen = 0;
  if ( this -> IsValid() ) {
    fDDSClient -> GetSubscription() -> SetMaxSyncDelay(maxSyncDelay);
    isOpen = 1;
  }

  fModified = true;
  return isOpen;

}


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

int IoDDSStreamItr::SetOffLine(bool offLine) 
{ 
  // Subscription is modified with the new offline status but is not sent
  // until the IoDDSStreamItr::Subscribe method is invoked.
  int isOpen = 0;
  if ( this -> IsValid() ) {
    fDDSClient -> GetSubscription() -> SetOffLine(offLine);
    isOpen = 1;
  }

  fModified = true;
  return isOpen;

}

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

int IoDDSStreamItr::SetDataSource(unsigned int dataSource) 
{ 
  // Subscription is modified with the new data source but is not sent
  // until the IoDDSStreamItr::Subscribe method is invoked.
  int isOpen = 0;
  if ( this -> IsValid() ) {
    DDS::EDataSource sourceEnum = static_cast<DDS::EDataSource>(dataSource);
    fDDSClient -> GetSubscription() -> SetDataSource(sourceEnum);
    isOpen = 1;
  }

  fModified = true;
  return isOpen;

}

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

int IoDDSStreamItr::SetKeepUpMode(unsigned int keepUpMode) 
{ 
  // Subscription is modified with the new keep up mode but is not sent
  // until the IoDDSStreamItr::Subscribe method is invoked.
  int isOpen = 0;
  if ( this -> IsValid() ) {
    DDS::EKeepUpMode keepUpEnum = static_cast<DDS::EKeepUpMode>(keepUpMode);
    fDDSClient -> GetSubscription() -> SetKeepUpMode(keepUpEnum);
    isOpen = 1;
  }

  fModified = true;
  return isOpen;

}

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

void IoDDSStreamItr::SetTimeOut(unsigned int timeout) { fTimeOut = timeout; }

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

void IoDDSStreamItr::SetClientType(DDS::EClientType clienttype) 
                                                { fClientType = clienttype; }

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

void IoDDSStreamItr::SetClientName(string clientname) 
                                                { fClientName = clientname; }

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

int IoDDSStreamItr::Streams(const char* streamList) 
{ 
  // Subscription is modified with the new streamList but is not sent
  // until the IoDDSStreamItr::Subscribe method is invoked.
  Int_t numStream = 0;
  if ( this->IsValid() ) {
    numStream = fDDSClient -> GetSubscription() -> SetStreams(streamList);
  }

  fModified = true;
  return numStream;

}

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

int IoDDSStreamItr::Select(const char* stream, const char* selection,
                           bool /* isRequired */) 
{ 
  // Subscription is modified with the new selection cut but is not sent
  // until the IoDDSStreamItr::Subscribe method is invoked.
  int isOpen = 0;
  if ( this->IsValid() ) {
    fDDSClient -> GetSubscription() -> SetSelection(stream,selection);
    isOpen = 1;
  }
  fModified = true;

  return isOpen;

}

---