DDSChildServer.cxx

Go to the documentation of this file.

//                                                                           //
// DDSChildServer                                                            //
//                                                                           //
// Package: DDS (Data Dispatcher System).                                    //
//                                                                           //
// S. Kasahara 5/2001                                                        //
//                                                                           //
// Purpose: This class is part of the MINOS data dispatcher system.          //
// It is the main class of the child data server, which resides on the same  //
// local network as the online data generator.  The child data server is     //
// executed by a child process forked by the parent server (managed by the   //
// DDSParentServer class).                                                   //
// The child data server handles a single client's request for near-online   //
// data.                                                                     //
//                                                                           //

#include  <errno.h>
#include "TMessage.h"  
#include "TSocket.h"
#include "TSystem.h"
#include "TStopwatch.h"
#include "TPluginManager.h"
#include "TROOT.h"

#include "MinosObjectMap/MomNavigator.h"
#include "Persistency/PerInputStream.h"
#include "Persistency/PerFileManager.h"
#include "Persistency/PerFile.h"
#include "MessageService/MsgService.h"

#include "Dispatcher/DDSChildServer.h"
#include "Dispatcher/DDSSubscription.h"
#include "Dispatcher/DDSFileHandler.h"

#include "Record/RecMinos.h" // needed to transfer data io temptags

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

ClassImp(DDSChildServer)

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

CVSID("$Id: DDSChildServer.cxx,v 1.34 2007/08/10 05:14:02 schubert Exp $");
 
// Definition of methods (alphabetical order)
// ******************************************

DDSChildServer::DDSChildServer(Int_t sockfd, Int_t maxinactive) : fTSocket(0), 
fSubscription(0), fFileHandler(0), fShutdown(false), fPid(0), fMom(0), 
fMessageIn(0), fMaxInactive(maxinactive) {
  // Purpose: Normal constructor for DDSChildServer object.  This constructor 
  //          creates a ROOT TSocket attached to the connected socket 
  //          descriptor sockfd (previously established and authorized
  //          by the parent server).
  //  
  // Argument: sockfd   connected socket descriptor.
  //
  // Return: n/a.
  //
  // Contact: S. Kasahara
  //
  // Notes: Use IsValid() to check if the DDSChildServer was created 
  //        successfully.
  //  
  
  // Get process id of child server for identification purposes
  fPid = gSystem -> GetPid();  

  // Create ROOT TSocket from connected socket descriptor
  fTSocket = new TSocket(sockfd);   

  if (!fTSocket || !fTSocket -> IsValid()) {
    // An error occured during the creation of the TSocket
    MSG("DDS",Msg::kWarning) << "CS_" << fPid  
      << ": Unable to create child server socket with socket descriptor "
      << sockfd << "." << endl;
    if (fTSocket) delete fTSocket; fTSocket = 0;
    fShutdown = true;
  }
  else {
    fServerAddress = fTSocket -> GetLocalInetAddress();
    fClientAddress = fTSocket -> GetInetAddress();
  }

  fInputStreamManager.SetUpdateMode(true);
  fInputStreamManager.SetPrintOpt("brief");

  // Remove default plugin (TXNetFile) for "root:" prefaced file so that
  // TNetFile will be used to connect to rootd server used with dispatcher
  // This was done beginning with root v5.10/00, but apparently should
  // no longer be done beginning with root v5.16/00.
  if ( gROOT -> GetVersionInt() < 51600 )   
    gROOT -> GetPluginManager() -> RemoveHandler("TFile","^root:");
  // Default subscription
  fSubscription = new DDSSubscription();
  fFileHandler = new DDSFileHandler(fSubscription->GetDataSource(),
                                    fSubscription->IsOffLine());  

}

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

  fInputStreamManager.CloseStream();
  if ( IsValid() ) delete fTSocket;  fTSocket = 0;
  if ( fSubscription ) delete fSubscription;  fSubscription = 0;
  if ( fFileHandler ) delete fFileHandler; fFileHandler = 0;
  if ( fMom ) delete fMom; fMom = 0;
  if ( fMessageIn ) delete fMessageIn; fMessageIn = 0;

}

void DDSChildServer::Get() {
  // Purpose: For now a private method to respond to a request to load
  //          a the current record set from the stream manager.
  //
  // Argument: none.
  //
  // Return: none.
  //
  // Contact: S. Kasahara
  //

  // If very first call to Next(), need to create Mom container
  if (!fMom) fMom = new MomNavigator();

  // Load current record set in Mom
  fInputStreamManager.Get(fMom);
  // Because fTempTags stored in records won't survive the i/o trip
  // need to unpack them here and repack them on the other side
  TIter riter = fMom->FragmentIter();
  RecMinos* record = 0;
  while ( (record = dynamic_cast<RecMinos*>(riter.Next())) ) {
    Registry* temptags = new Registry(record->GetTempTags()); // get copy
    fMom->AdoptFragment(temptags); // adopted by mom
  }
  // Load retrieved mom object into fMessageOut for shipping
  fMessageOut.Reset(DDS::kOk);
  fMessageOut.WriteObject(fMom);
  return;

}

void DDSChildServer::GoToFile() {
  // Purpose: This Service responds to a request from the client to
  //          advance to a user-specified filename.
  //
  // Argument: none.
  //
  // Return: none.
  //
  // Contact: S. Kasahara
  //

  if ( !fFileHandler || !fFileHandler -> IsValid()) {
    fMessageOut.Reset(DDS::kFileError);
    MSG("DDS",Msg::kWarning) << "CS_" << fPid 
         << " GoToFile called but DDSFileHandler is InValid." << endl; 
    return;
  }

  // Retrieve filename from fMessageIn 
  char filename[512];
  (*fMessageIn) >> filename;
  std::string newfilename;
  if ( !strncmp(filename,"next",4) ) {
    MSG("DDS",Msg::kDebug) << "CS_" << fPid << " GoToNextFile requested." 
                           << endl;
    newfilename = fFileHandler->GoToNextFile();
  }
  else if ( !strncmp(filename,"symlink",7) ) {
    MSG("DDS",Msg::kDebug) << "CS_" << fPid << " GoToSymLinkFile requested." 
                           << endl;
    newfilename = fFileHandler->GoToSymLinkFile();
  }
  else {
    MSG("DDS",Msg::kDebug) << "CS_" << fPid << " GoToFile " << filename 
                           << " requested." << endl;
    newfilename = fFileHandler->GoToFile(filename);
  }

  if ( !newfilename.empty() ) {
    MSG("DDS",Msg::kVerbose) << "CS_" << fPid 
               << " PerInputStreamManager status before closing old file:\n" 
               << fInputStreamManager;
    fInputStreamManager.CloseFile();
    MSG("DDS",Msg::kDebug) << "CS_" << fPid << " Advancing to " << newfilename 
                           << "." << endl;
    if(!fInputStreamManager.SetFile("*",newfilename,Per::kRead)) {
      fMessageOut.Reset(DDS::kFileError);
    }
    else {
      fMessageOut.Reset(DDS::kOk);
      fMessageOut << newfilename.c_str();
      MSG("DDS",Msg::kVerbose) << "CS_" << fPid 
          << " PerInputStreamManager status after opening new file:\n" 
                               << fInputStreamManager;
    }
  }
  else {
    MSG("DDS",Msg::kDebug) << "CS_" << fPid 
           << " Failed to retrieve new filename." << endl;     
    fMessageOut.Reset(DDS::kFileError);
  }

  return;

}

bool DDSChildServer::IsClientConnected() const {
  // Purpose: Check client/server socket connection to make sure that client 
  //          has not departed abruptly.
  //
  // Argument: none.
  //
  // Return: true/false.
  //
  // Contact: S. Kasahara
  //

  TSystem::ResetErrno();
  if (!fTSocket || !fTSocket->IsValid()) return false;

  Int_t n;
  UInt_t len;
  Int_t savevalue;
  fTSocket->GetOption(kNoBlock,savevalue);
  fTSocket->SetOption(kNoBlock,1); // set no block
  Int_t socketdescriptor = fTSocket->GetDescriptor();
  n = gSystem->RecvRaw(socketdescriptor, &len, sizeof(UInt_t),kPeek);
  fTSocket->SetOption(kNoBlock,savevalue); // reset noblock option

  if ( n <= 0 && n != -4) return false; // socket error
  return true; // connection is valid

}

void DDSChildServer::Next() {
  // Purpose: This Service responds to a request from the client to
  //          send the next available entry satisfying the client's
  //          subscription.  This method is activated when the DDS::kNext
  //          message is received in DDSChildServer::Run().
  //
  // Argument: none.
  //
  // Return: none.
  //
  // Contact: S. Kasahara
  //

  // Client must have submitted subscription first
  if ( !fSubscription ) {
    fMessageOut.Reset(DDS::kError);
    MSG("DDS",Msg::kWarning) << "CS_" << fPid 
         << " Next called but no subscription has been submitted." << endl; 
    return;
  }

  if ( !fFileHandler || !fFileHandler -> IsValid()) {
    fMessageOut.Reset(DDS::kFileError);
    MSG("DDS",Msg::kWarning) << "CS_" << fPid 
         << " Next called but DDSFileHandler is InValid." << endl; 
    return;
  }

  // If very first call to Next(), need to create Mom container
  if (!fMom) fMom = new MomNavigator();

  // Retrieve waittime (secs) and advanceby from fMessageIn and start timer
  UInt_t waittime;
  (*fMessageIn) >> waittime;
  UInt_t advanceby;
  (*fMessageIn) >> advanceby;
  TStopwatch timer; timer.Start();
  bool isNewSet = false; // at least one new record set available

  Int_t keepUpWindow = fSubscription -> GetKeepUpWindow();
  while ( timer.RealTime() < waittime ) {
    timer.Continue();
    fMom->Clear();  // clear mom object
    
    DDS::EKeepUpMode keepupmode = fSubscription -> GetKeepUpMode();
    if ( (keepupmode == DDS::kFileKeepUp || keepupmode == DDS::kRecordKeepUp)
         && fFileHandler->NewFileAvailable() ) {
      // Set new file for managed streams
      std::string newfilename = fFileHandler->GoToSymLinkFile();
      MSG("DDS",Msg::kDebug) << "CS_" << fPid
        << " Next detected NewFileAvailable and keepupmode is " 
        << DDS::AsString(keepupmode) << ", called GoToSymLinkFile." << endl;
      if ( !newfilename.empty() ) {
        MSG("DDS",Msg::kVerbose) << "CS_" << fPid 
              << " PerInputStreamManager status before closing old file:\n" 
                                 << fInputStreamManager;
        fInputStreamManager.CloseFile();
        MSG("DDS",Msg::kDebug) << "CS_" << fPid << " Advancing to " 
                              << newfilename << "." << endl;
        if(!fInputStreamManager.SetFile("*",newfilename,Per::kRead)) {
          fMessageOut.Reset(DDS::kFileError);
          return;
        }
        else {
          MSG("DDS",Msg::kVerbose) << "CS_" << fPid 
              << " PerInputStreamManager status after opening new file:\n" 
                                   << fInputStreamManager;
        }
      }
      else {
        MSG("DDS",Msg::kDebug) << "CS_" << fPid 
                        <<  " Failed to retrieve new filename." << endl;     
      }
    }
    if ( !fInputStreamManager.IsValidSelectionString() ) {
      MSG("DDS",Msg::kDebug) << "CS_" << fPid
                     << " Has invalid subscription selection string." << endl;
      fMessageOut.Reset(DDS::kInvalidSelection);
      return;
    }
    
    if ( fInputStreamManager.Next(0,advanceby) > 0 ) {
      isNewSet = true;
      if ( keepupmode != DDS::kRecordKeepUp ) {
        // Load current record set in Mom
        this -> Get();
        return;
      }
      else {
        Int_t lastTime = (Int_t)
          (fInputStreamManager.GetLastEntryVld().GetTimeStamp().GetSec());
        Int_t currentTime = (Int_t)
          (fInputStreamManager.GetCurrentVld().GetTimeStamp().GetSec());
        if ( lastTime - currentTime <= keepUpWindow ) {
          this -> Get();
          return;
        }
      }
    }
    else {
      // Determine if file has reached end or has been aborted
      if ( fFileHandler->NewFileAvailable() && 
          !fInputStreamManager.IsFileEnd() ) {
        std::string currentfilename = "";
        const PerStreamManager::StreamMap& sm_mgr
                                      = fInputStreamManager.GetStreamMap();
        PerStreamManager::StreamMapConstItr itr_mgr = sm_mgr.begin();
        const PerInputStream* instream = 0;
        if ( itr_mgr != sm_mgr.end() ) instream 
                        = dynamic_cast<PerInputStream*>(itr_mgr->second);
        if (instream) currentfilename = instream->GetFullFilePathName();
        PerFileManager& filemgr = PerFileManager::Instance();
        if ( !((filemgr.GetOpenedFile(currentfilename))->HasFileEndKey()) ) {
          std::string symlinktargetname = fFileHandler->GetSymLinkTargetName();
          if ( symlinktargetname != currentfilename ) {
            MSG("DDS",Msg::kDebug) << "CS_" << fPid << " Current file " 
            << currentfilename << "\n not closed but symlink currentfile " 
            << symlinktargetname << "\n has moved on." 
            << " Assume abort, finish current file, and move to next file." 
            << endl;
            fInputStreamManager.SetFileEnd();
          }
        }
      }
      else if ( fInputStreamManager.IsFileEnd() ){  
        // All streams have reached end and writer has closed file
        // Attempt to update file for all managed streams
        std::string fullfilepathname="";
        if ( keepupmode == DDS::kAll) {
          fullfilepathname = fFileHandler->GoToNextFile();
          if ( !fullfilepathname.empty() ) {
            MSG("DDS",Msg::kDebug) << "CS_" << fPid
            << " Next call to GoToNextFile retrieved file\n\t" 
            << fullfilepathname <<  "." << endl;
          }
        }
        else if ( fFileHandler->NewFileAvailable() ) {
          fullfilepathname = fFileHandler->GoToSymLinkFile();
          MSG("DDS",Msg::kDebug) << "CS_" << fPid
          << " Next call to GoToSymLinkFile retrieved filename\n\t" 
          << fullfilepathname << "." << endl;
        }
        if (!fullfilepathname.empty()) {
          MSG("DDS",Msg::kVerbose) << "CS_" << fPid 
           << " PerInputStreamManager status before closing old file:\n" 
           << fInputStreamManager;
          fInputStreamManager.CloseFile();
          if(!fInputStreamManager.SetFile("*",fullfilepathname,Per::kRead)) {
            fMessageOut.Reset(DDS::kFileError);
            return;
          }
          MSG("DDS",Msg::kVerbose) << "CS_" << fPid 
                                   << " PerInputStreamManager status after opening new file:\n" 
                                   << fInputStreamManager;
        }
        else {
          if ( keepupmode == DDS::kRecordKeepUp && isNewSet ) {
            // File end was reached. Must back up a set and then Get
            fInputStreamManager.Previous(0,advanceby);
            this -> Get();
            return;
          }
          // Sleep for 1 sec while we wait for new file
          gSystem->Sleep(1000);
          // After waking up, test socket connection to make sure client is
          // still there.
          if ( !IsClientConnected() ) {
            fMessageOut.Reset(DDS::kSocketError);
            return;
          } 
        }
      }
      else {
        if ( keepupmode == DDS::kRecordKeepUp && isNewSet ) {
          // don't wait, retrieve most recent new set available
          this -> Get();
          return;
        }
        // Sleep for 1 sec while we wait for updated tree
        gSystem->Sleep(1000);
        // After waking up, test socket connection to make sure client is
        // still there.
        if ( !IsClientConnected() ) {
          fMessageOut.Reset(DDS::kSocketError);
          return;
        } 
      }
    }
  }
  timer.Stop();
  // Next reached "TimeOut", find out if its because it was waiting for
  // a new record from the current file or because it was waiting for a new
  // file.
  if ( !fInputStreamManager.IsFileEnd() ) {
    fMessageOut.Reset(DDS::kTimeoutNewRecord);
  }
  else {
    fMessageOut.Reset(DDS::kTimeoutNewFile);
  }
  
  return;

}

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

  if ( IsValid() ) {
    ms << "CS_" << fPid << ": connected to client at " 
    // Print the client's host name, address, and port number
       << fClientAddress.GetHostName() <<"/"<< fClientAddress.GetHostAddress()
       << "(port " << fClientAddress.GetPort() << ")." << endl;
  }
  else {
    ms << "CS_" << fPid << ": childserver socket is not connected." << endl;
  }

  return ms;

}

Int_t DDSChildServer::Run() {
  // Purpose: This is the main method of the DDSChildServer.  It listens
  //          for and responds to client service requests.  Each request
  //          received (of type DDS::EMessageType) is responded to in
  //          return with a single message of type DDS::EMessageType sent
  //          to the client upon completion of servicing the request.
  //          In this way the client can remain in sync with the child
  //          server's processing of its requests.
  //
  //          The services which the client can currently request from the
  //          child server are:
  //          DDS::kGoToFile == Request to advance to new file.
  //          DDS::kNext == Send next entry satisfying subscription.
  //          DDS::kShutdown == Request to shutdown child server.
  //          DDS::kSubscribe == Request to submit new subscription.
  //          
  //          The return status with which the child server can respond to
  //          the client are those returned by the service methods: 
  //          GoToFile(),Next(),Shutdown(), and Subscribe() as well as:
  //          DDS::kMessageUnknown == Unrecognized message received.
  //          DDS::kSocketError == An error return was received on the
  //                               socket receipt of the client's message.   
  //
  // Argument: none.
  //
  // Return: Return code = 0 => normal finish
  //                     = 1 => socket error caused premature end
  //
  // Contact: S. Kasahara
  //
  // Notes: The DDSChildServer will remain in the Run method until it receives
  //        the DDS::kShutdown message from its client.  
  // 

  Int_t rc = 0;

  TStopwatch timer;
  timer.Start();

  while ( !fShutdown ) {
    // Wait for new message from client
    fTSocket -> SetOption(kNoBlock,1);
    Int_t rettype = fTSocket->Recv(fMessageIn);
    fTSocket -> SetOption(kNoBlock,0);
    if ( rettype != -4 ) {
      timer.Reset();
      if ( rettype > 0 ) {
        switch ( fMessageIn -> What() ) {

        case DDS::kGoToFile:
          // Request to send next entry satisfying subscription
          GoToFile();
          break;

        case DDS::kNext:
           // Request to send next entry satisfying subscription
          Next();
          break;

        case DDS::kShutdown:
          // Request to shutdown child server
          Shutdown();
          break;

        case DDS::kSubscribe:
          // Request to submit new subscription
          Subscribe();
          break;

        default:
          MSG("DDS",Msg::kWarning)<< "CS_"<< fPid << ": Unknown client message: "
                                << fMessageIn -> What() << " received.\n"
                                << "Unable to process client request." << endl;
          fMessageOut.Reset(DDS::kMessageUnknown);
          break;
        } // end of switch block

        if ( fMessageIn ) delete fMessageIn; fMessageIn = 0; 

      } // end of socket received message block
      else {
        // Error return on TSocket::Recv call
        fShutdown = true;
        delete fTSocket; fTSocket = 0;
        MSG("DDS",Msg::kWarning) << "CS_" << fPid
              << ": Socket error detected on TSocket::Recv." <<
              "\n ChildServer will be shutdown." << endl;
        rc = 1;
      }
    
      if ( fTSocket && IsClientConnected() ) {
        // send return status & objects to client
        if ( fTSocket -> Send(fMessageOut) < 0 ) {
          // An error occurred while sending message
          if (errno == EPIPE) {
            // Broken pipe between child server and client is fatal
            fShutdown = true;
            delete fTSocket; fTSocket = 0;
            MSG("DDS",Msg::kWarning) << "CS_" << fPid
               << ": Socket error detected on TSocket::Send." 
               << "\n ChildServer will be shutdown." << endl;
            rc = 1;
          }
        }
      }
      else if (!fShutdown) {
        // Error detected in socket connection
        fShutdown = true;
        delete fTSocket; fTSocket = 0;
        MSG("DDS",Msg::kWarning) << "CS_" << fPid
        << ": Socket error detected indicating client has disconnected.\n"
                               << " ChildServer will be shutdown."<< endl;
        rc = 1;
      }
    }
    else {
      if ( timer.RealTime() > fMaxInactive ) {
        MSG("DDS",Msg::kWarning) << "CS_" << fPid
                                 << ": Inactivity time limit of "
                                 << fMaxInactive << "(sec) reached.\n"
                                 << "ChildServer will be shutdown." << endl;
        Shutdown();
        fMessageOut.Reset(DDS::kInactive);
        rc = 2;
        if ( fTSocket && IsClientConnected() ) fTSocket->Send(fMessageOut);
      }
      else { 
        timer.Continue();
        gSystem->Sleep(10); // sleep 10 msec so as to avoid excessive cpu
      }
    }
  } // End of loop over client requests 

  return rc;

}  

void DDSChildServer::Shutdown() {
  // Purpose: This Service shuts down the child server.  This method is 
  //          activated when the DDS::kShutdown message is received in
  //          DDSChildServer::Run().
  //
  // Argument: none.
  //
  // Return: none.
  //
  // Contact: S. Kasahara
  //

  fShutdown = true;   // stops processing loop in Run
  MSG("DDS",Msg::kVerbose) << "CS_" << fPid 
     << " PerInputStreamManager status before shutdown:\n" 
     << fInputStreamManager; 
  fInputStreamManager.CloseStream();  // close all managed streams
  fMessageOut.Reset(DDS::kOk);

  return;

}

void DDSChildServer::Subscribe() {
  // Purpose: This Service receives and processes a new subscription from
  //          the client.  This method is activated when the DDS::kSubscribe
  //          message is received in DDSChildServer::Run().
  //
  // Argument: none.
  //
  // Return: none.
  //
  // Contact: S. Kasahara
  //

  // If previous subscription exists, must delete it
  if ( fSubscription) delete fSubscription; fSubscription = 0;

  // Attempt to receive DDSSubscription object from fMessageIn
  fSubscription = dynamic_cast<DDSSubscription*>
                 (fMessageIn -> ReadObject(fMessageIn->GetClass()));
  if ( !fSubscription ) {
    MSG("DDS",Msg::kWarning) << "CS_" << fPid
    << "\nSubscribe failed to receive expected DDSSubscription from client." 
    << endl;
    fMessageOut.Reset(DDS::kError);
  }
  else {
    // Subscription received successfully
    MSG("DDS",Msg::kDebug) << "CS_" << fPid
                           << " Subscribe received subscription:\n"
                           << fSubscription;

    fMessageOut.Reset(DDS::kOk);
    if (fFileHandler && ( (fSubscription -> GetDataSource())  
                        !=(fFileHandler  -> GetDataSource())
                       || (fSubscription -> IsOffLine())
                        !=(fFileHandler -> IsOffLine()) ) ) {
      delete fFileHandler; fFileHandler=0;
    }
    if (!fFileHandler) 
      fFileHandler = new DDSFileHandler(fSubscription -> GetDataSource(),
                                        fSubscription -> IsOffLine());
    // Determine current file from filehandler
    std::string currentFileName = fFileHandler->GetCurrentFileName();

    // Open new streams in subscription list
    PerInputStream* instream;
    const DDSSubscription::StreamMap& sm_sub = fSubscription->GetStreamMap();
    for (DDSSubscription::StreamMapConstItr itr_sub = sm_sub.begin();
                                            itr_sub!= sm_sub.end();++itr_sub) {
      string streamName = (itr_sub->first).Data();
      if ( !(instream = dynamic_cast<PerInputStream*>
                        (fInputStreamManager.GetOpenedStream(streamName))) ) {
        // For DDS application, streamname & treename are the same;
        instream = fInputStreamManager.OpenStream(streamName,streamName);
        if (!currentFileName.empty())instream -> SetFile(currentFileName);
      }
      if ( instream ) instream -> SetSelection((itr_sub->second).Data());
    }

    fInputStreamManager.SetMaxSyncDelay(fSubscription -> GetMaxSyncDelay());

    // Finally,  close all streams not on subscription list
    const PerStreamManager::StreamMap& sm_mgr
                                      = fInputStreamManager.GetStreamMap();
    for ( PerStreamManager::StreamMapConstItr itr_mgr = sm_mgr.begin();
                                           itr_mgr!= sm_mgr.end(); ++itr_mgr) {
      std::string streamName = itr_mgr->first;
      DDSSubscription::StreamMapConstItr itr_sub 
                                        = sm_sub.find(streamName.c_str());
      if (itr_sub == sm_sub.end())fInputStreamManager.CloseStream(streamName);
    }
  }
  
  return;

}

---