DDSClient.cxx

Go to the documentation of this file.

//                                                                           //
// DDSClient                                                                 //
//                                                                           //
// Package: DDS (Data Dispatcher System).                                    //
//                                                                           //
// S. Kasahara 5/2001                                                        //
//                                                                           //
// Purpose: This class is the interface to be used by the client to receive  //
//          near-online data from the data dispatcher server.                //
//                                                                           //

#include <errno.h>
#include "TClass.h"
#include "TSocket.h"  
#include "MessageService/MsgService.h"
#include "MinosObjectMap/MomNavigator.h"

#include "Dispatcher/DDSClientId.h"
#include "Dispatcher/DDSClient.h"
#include "Dispatcher/DDSSubscription.h"
#include "Dispatcher/DDSPSStatus.h"

#include "Record/RecMinos.h" // for repacking temptags

std::ostream& operator << (std::ostream& ms, DDSClient* cs) { 
                                             return cs->Print(ms); } 

ClassImp(DDSClient)

// Definition of static data members
// *********************************

CVSID("$Id: DDSClient.cxx,v 1.16 2007/08/24 05:23:48 schubert Exp $");
 
// Definition of methods (alphabetical order)
// ******************************************

DDSClient::DDSClient(const char* serverhost, UInt_t serverport, 
                     DDS::EMessageType msgrequest, 
                     DDS::EClientType clienttype, string clientname) : 
                     fStatus(DDS::kSocketError), fTSocket(0), 
                     fSubscription(0), fMessageIn(0), fClientId(0),
                     fPSStatus(0) {
  // Purpose: Default constructor for DDSClient class.  This constructor 
  //          creates a full-duplex communication TSocket attached to 
  //          the serverport at serverhost.  If serverport = 0 (default),
  //          the socket will be connected to the named service "MinosDDS"
  //          instead.
  //  
  // Argument: serverhost  host name of dispatcher parent server
  //                       (e.g. "kasahara.hep.umn.edu").
  //           serverport  port number of dispatcher parent server.  If
  //                       serverport = 0 (default), socket will attempt to
  //                       connect to named service "MinosDDS" instead.
  //           msgrequest  initial service request message sent to parent 
  //                       server.  The services which the client can
  //                       currently request from the parent server are:
  //                       DDS::kData == Request for data (default).
  //                       DDS::kShutdown == Request to shutdown parent server.
  //                       DDS::kStatus = Request for ddsparentserver status.
  //                       (Note that to request shutdown of the parent server,
  //                        the client must be local to the parent or the 
  //                        service will be denied.)
  //
  // Return: n/a.
  //
  // Contact: S. Kasahara
  //
  // Notes: Use IsValid() to check if the DDSClient was created 
  //        successfully.
  //  
  
  // Customize message service to print time stamp for all log levels
  // for DDS stream
  MsgStream* msdds = MsgService::Instance() -> GetStream("DDS");
  for ( int ilvl = Msg::kMinLogLevel ; ilvl < Msg::kNLogLevel; ilvl++ ) {
    msdds -> AddFormat(ilvl,Msg::kTime);
  }
  
  // Connect to dispatcher parent server
  if ( !serverport ) {
    // User has not specified an alternative port for the parent server.  Will
    // attempt to connect to the port assigned to the named service "MinosDDS".
    fTSocket = new TSocket(serverhost,"MinosDDS");
  }
  else {
    fTSocket = new TSocket(serverhost,serverport);   
  }

  if ( !fTSocket || !fTSocket -> IsValid() ) {
    // An error occured during the creation of the TSocket
    if ( serverport ) {
      MSG("DDS",Msg::kWarning) 
        << "Unable to create socket connected to server host\n"
        << serverhost << " and port " << serverport << "." << endl;
    }
    else {
      MSG("DDS",Msg::kWarning)
        << "Unable to create socket connected to server host\n"
        << serverhost << " and named service MinosDDS." << endl;
    }
    if (fTSocket) delete fTSocket; fTSocket = 0;
  }

  else {

    // Store the addresses of the client and server
    fClientAddress = fTSocket -> GetLocalInetAddress();
    fServerAddress = fTSocket -> GetInetAddress();

    // Construct Client identity information. This is sent to ddsparentserver
    fClientId = new DDSClientId(clienttype,clientname);
    
    // Send initiating message to parent server. Client is only allowed
    // one message exchange with parent server.
    fMessageOut.Reset(msgrequest | kMESS_ACK);
    fMessageOut.WriteObject(fClientId); 

    DDS::EMessageType msgrc = SendMessage();
    fStatus = msgrc;
    
    if ( msgrc != DDS::kOk ) {
      MSG("DDS",Msg::kWarning)
        << "The error return message " << DDS::AsString(msgrc) 
        << "\nwas received from the dispatcher parent server in response to\n"
        << "the client's request for " << DDS::AsString(msgrequest) << "." 
        << endl;
      if (fTSocket) delete fTSocket; fTSocket = 0;
    }
    if (msgrequest == DDS::kShutdown || msgrequest == DDS::kStatus ) {
      // Socket is no longer valid if shutdown of parent server was requested.
      if ( fTSocket ) delete fTSocket; fTSocket = 0;
      if ( msgrequest == DDS::kStatus ) {
        fPSStatus = dynamic_cast<DDSPSStatus*>
            (fMessageIn -> ReadObject(fMessageIn->GetClass()));
        if ( !fPSStatus ) {
          MSG("DDS",Msg::kWarning) 
            << "Request for parentserver report status failed to receive "
            << "\nDDSPSStatus object." << endl;
        }
      }
    }
  }

  if ( IsValid() ) {
    // If socket has been created successfully, create default subscription
    fSubscription = new DDSSubscription();
  }
  
}
  

DDSClient::~DDSClient() {
  // Purpose: Destructor.
  //
  // Argument: n/a.
  //
  // Return: n/a.
  //
  // Contact: S. Kasahara
  //

  if ( IsValid() ) Shutdown();
  if ( fSubscription ) delete fSubscription; fSubscription = 0;
  if ( fMessageIn ) delete fMessageIn; fMessageIn = 0;
  if ( fClientId ) delete fClientId; fClientId = 0;
  if ( fPSStatus ) delete fPSStatus; fPSStatus = 0;
  
}

DDS::EMessageType DDSClient::GoToFile(std::string filename,
                                      std::string& returnedfilename) {
  // 
  // Purpose: Advance to serving data from file filename.
  //
  // Arguments:filename  filename of interest, e.g. F00001782_0000.mdaq.root.
  //                     Specifying a relative path means that the path
  //                     will be assumed relative to the data source
  //                     directory defined on the server.
  //                     filename may be a symbolic link file, e.g.
  //                     'currentfile' will advance to the file pointed
  //                     to by 'currentfile' in the data source directory.
  //
  //           returnedfilename  filename actually advanced to on the server 
  //                             side.
  //
  // Return: DDS::kOk if returnedfilename is not empty, else errorcode.
  //
  // Contact: S. Kasahara
  //
  // Notes:  If file is not found, the server will be set to serve
  //         data from the first file alphanumerically beyond the requested
  //         filename.
  //         Special arguments: "next" (advance to next file)
  //

  DDS::EMessageType msgrc;  // return status of service request
  char newfilename[512];

  if ( this -> IsValid() ) {
    // reset message buffer and tell it type of msg
    fMessageOut.Reset(DDS::kGoToFile | kMESS_ACK);
    fMessageOut << filename.c_str();  // load filename into message buffer

    // Send fMessageOut to childserver
    msgrc = SendMessage();
    if ( msgrc == DDS::kOk ) {
      // fMessageIn should contain newfilename string
      (*fMessageIn) >> newfilename;
    }
  }
  else {
    MSG("DDS",Msg::kWarning) 
    << "GoToFile requested with invalid socket connection." << endl;
    msgrc = DDS::kSocketError; 
  }
  returnedfilename = newfilename;
  return msgrc;

}

DDS::EMessageType DDSClient::GoToNextFile(std::string& returnedfilename) {
  // 
  // Purpose: Advance to serving data from next file in directory.
  //          This request only makes sense when user has subscribed in
  //          DDS::kAll mode.
  //
  // Arguments: returnedfilename.  Name of file actually advanced to.
  //
  // Return: DDS::kOk if returnedfilename is not empty, else errorcode.
  //
  // Contact: S. Kasahara
  //

  return this -> GoToFile("next",returnedfilename);

}

DDS::EMessageType DDSClient::Next(MomNavigator*& mom,UInt_t waittime,
                                  UInt_t advanceby) {
  // 
  // Purpose: Request and receive next available entry satisfying client's
  //          subscription from child server.
  //
  // Arguments:mom       pointer to MomNavigator object to be filled with
  //                     retrieved MomNavigator from dispatcher server.
  //                     See Notes. 
  //           waittime  time(sec) that client wishes to wait for next 
  //                     available entry satisfying subscription before 
  //                     timing out (default = 120).
  //           advanceby number of record sets to advance by through 
  //                     stream(s) (default = 1).
  //
  // Return: DDS::EMessageType  return status of service request.
  //
  // Contact: S. Kasahara
  //
  // Notes:  For historic reasons, the user can specify the input 
  //         MomNavigator* mom pointer in one of two ways:
  //         1)mom == 0.  In this case, a new MomNavigator is created and mom
  //                      is set to point to it.  
  //         2)mom == a pointer to an existing MomNavigator object.  In this
  //                  case, objects retrieved by the dispatcher are added to
  //                  the existing mom object.
  //         In both cases, the user owns and must delete the MomNavigator
  //         object pointed to by mom.
  //

  DDS::EMessageType msgrc;  // return status of service request
  // Local copy of Mom used to receive collection of objects from childserver
  MomNavigator* localMom = 0;  

  if ( IsValid() ) {
    // reset message buffer and tell it type of msg
    fMessageOut.Reset(DDS::kNext | kMESS_ACK); 
    fMessageOut << waittime;  // load requested waittime into message buffer
    fMessageOut << advanceby; // load requested advanceby interval
    // Send fMessageOut to childserver
    msgrc = SendMessage();
    if ( msgrc == DDS::kOk ) {
      // fMessageIn should contain object of type MomNavigator
      localMom = dynamic_cast<MomNavigator*>
                 (fMessageIn -> ReadObject(fMessageIn->GetClass()));
      if (!localMom) {
        MSG("DDS",Msg::kWarning) 
        << "Next failed to receive expected MomNavigator object." << endl;
      }
      else {
        // if user hasn't supplied a mom pointer, initialize one here
        if (!mom) mom = new MomNavigator(); // this is dumb, but historic
        // Loop over Mom fragments to repack fTempTags object in record
        // that otherwise wouldn't survive i/o.  This is ugly.
        // Add received records to users Mom
        TObjArray* objArray =const_cast<TObjArray*>
                               (localMom -> GetFragmentArray());
        Int_t nrecord = objArray -> GetEntriesFast();
        for ( Int_t irec=0; irec < nrecord/2; irec++ ) {
          //record is removed from localMom to pass to usr's mom for ownership 
          RecMinos* record = dynamic_cast<RecMinos*>(objArray->RemoveAt(irec));
          Registry* temptags=dynamic_cast<Registry*>
                               (objArray->At(irec+nrecord/2));
          record->GetTempTags().Set("stream",temptags->GetCharString("stream"));
          record->GetTempTags().Set("tree",temptags->GetCharString("tree"));
          record->GetTempTags().Set("file",temptags->GetCharString("file"));
          record->GetTempTags().Set("index",temptags->GetInt("index"));
          record->GetTempTags().SetName(temptags->GetName());
          mom -> AdoptFragment(record); // new owner of record
        }
        delete localMom; localMom=0; // clears array and deletes temptags
      }
    } 
  }
  else {
    MSG("DDS",Msg::kWarning) 
    << "Next requested with invalid socket connection." << endl;
    msgrc = DDS::kSocketError; 
  }

  return msgrc;

}

std::ostream& DDSClient::Print(std::ostream& ms) const {
  // Purpose: Print DDSClient status on std::ostream.
  //
  // Argument: ms  std::ostream to print on.
  //
  // Return: std::ostream reference.
  //
  // Contact: S. Kasahara
  //

  if ( IsValid() ) {
    ms << "Client socket connected from local address:" << endl;
    // Print the local host name, address, and port number
    ms << fClientAddress.GetHostName() << "/"<< fClientAddress.GetHostAddress()
       << "(port " << fClientAddress.GetPort() << ")." << endl;
    ms << "to a child server at remote address:" << endl;
    // Print the child server's host name, address, and port number
    ms << fServerAddress.GetHostName() << "/"<< fServerAddress.GetHostAddress()
       << "(port " << fServerAddress.GetPort() << ")." << endl;
  }
  else {
    ms << "Client to child server socket is not connected." << endl;
  }

  return ms;

}

DDS::EMessageType DDSClient::SendMessage() {
  //
  // Purpose: Send message previously loaded in fMessageOut to child server.
  //
  // Argument: none.
  //
  // Return: msgrc of type DDS::EMessageType received from server. Complete
  //         message received from server is stored in fMessageIn.
  //
  // Contact: S. Kasahara
  //

  DDS::EMessageType msgrc = DDS::kSocketError;  // default return code
  // Delete previously retrieved message if any
  if (fMessageIn) delete fMessageIn; fMessageIn = 0;

  // Send requested message to server
  Int_t rc = fTSocket -> Send(fMessageOut);
  if (rc < 0) {
    MSG("DDS",Msg::kWarning) << "\nReceived socket error code "
                 << rc << " in send of message to server." << endl; 
    //  if ( errno == EPIPE ) {
      // Broken pipe between child server and client is fatal
      MSG("DDS",Msg::kWarning) 
      << "\nSocket will be shutdown." << endl;
      if ( fTSocket ) delete fTSocket; fTSocket = 0;
    // }
  }
  else {
    // Wait for return message from user
    Int_t rc = fTSocket -> Recv(fMessageIn);
    if (rc < 0) {
      MSG("DDS",Msg::kWarning) << "\nReceived socket error code "
                 << rc << " in send of message to server." << endl; 
      //if ( errno == EPIPE ) {
        // Broken pipe between child server and client is fatal
        MSG("DDS",Msg::kWarning) 
        << "\nSocket will be shutdown." << endl;
        if ( fTSocket ) delete fTSocket; fTSocket = 0;
      // }
    }
    else {
      if ( ! fMessageIn ) {
        MSG("DDS",Msg::kWarning) 
          << "Received empty fMessageIn from fTSocket->Recv()"
          << endl;
      }
      else 
        msgrc = static_cast< DDS::EMessageType > (fMessageIn -> What());
    }
  }

  return msgrc;

}


DDS::EMessageType DDSClient::Shutdown() {
  //
  // Purpose: Send child server the shutdown message.
  //
  // Argument: none.
  //
  // Return: DDS::EMessageType  return status of service request
  //
  // Contact: S. Kasahara
  //

  DDS::EMessageType msgrc;  // return status of service request

  if ( IsValid() ) {
    // reset message buffer & tell it msg type
    fMessageOut.Reset(DDS::kShutdown | kMESS_ACK); 
    // Send fMessageOut to childserver
    msgrc = SendMessage();
  }
  else {
    MSG("DDS",Msg::kWarning) 
    << "Shutdown requested with invalid socket connection." << endl;
    msgrc = DDS::kSocketError; 
  }

  if (fTSocket) delete fTSocket; fTSocket = 0;

  return msgrc;

}

DDS::EMessageType DDSClient::Subscribe() {
  //
  // Purpose: Submit subscription object to the dispatcher child server.
  //
  // Argument: none.
  //
  // Return: DDS::EMessageType  return status of service request
  //
  // Contact: S. Kasahara
  //
  // Notes: The subscription object of type DDSSubscription can be retrieved 
  //        and modified by the client via the DDSClient::GetSubscription() 
  //        method.  There is no limit to the number of times that the
  //        subscription object can be modified and resubmitted to the server.
  //

  DDS::EMessageType msgrc;  // return status of service request

  if ( IsValid() ) {
    // reset buffer & tell it msg type
    fMessageOut.Reset(DDS::kSubscribe | kMESS_ACK); 
    fMessageOut.WriteObject(fSubscription); 
    msgrc = SendMessage(); // Submit subscription to server
  }
  else {
    MSG("DDS",Msg::kWarning) 
    << "Subscribe requested with invalid socket connection." << endl;
    msgrc = DDS::kSocketError; 
  }

  return msgrc;

}

---