#include "/cvmfs/nova-development.opensciencegrid.org/novasoft/releases/N21-01-24/rms/cxx/include/ClientListenerLoop.h"

Inheritance diagram for gov::fnal::cd::rms::ClientListenerLoop:

Public Member Functions
	ClientListenerLoop (util::LinkedBlockingQueue< boost::shared_ptr< std::string > > *inputQueue)

	~ClientListenerLoop ()

void	run ()

void	stop ()

void	setMessageListener (RmsMessageListener *listener)

void	setMessageFilter (MessageFilter *filter)

virtual bool	isRunning () const

virtual bool	isDone () const

virtual void	watchfulSleep (int microseconds, int numberOfIntervals=10)

Protected Attributes
bool	_running

bool	_stopRequested

bool	_hasCompleted

Private Attributes
util::LinkedBlockingQueue< boost::shared_ptr< std::string > > *	_inputQueue

RmsMessageListener *	_clientListener

MessageFilter *	_messageFilter

Detailed Description

The ClientListenerLoop decouples receiveing messages from the provider and sending them to any message listeners.

Author: Kurt Biery; Steve Foulkes

Version

Revision: 1.2.18.1

Date: 2019/09/27 00:07:31

Definition at line 30 of file ClientListenerLoop.h.

Constructor & Destructor Documentation

gov::fnal::cd::rms::ClientListenerLoop::ClientListenerLoop ( util::LinkedBlockingQueue< boost::shared_ptr< std::string > > * inputQueue )

Create a new ClientListenerLoop with not client listner and no message filter. The loop will just spin until a client listener has been set.

Parameters

inputQueue The input queue to pull messages off of.

Definition at line 20 of file ClientListenerLoop.cpp.

References _clientListener, _inputQueue, and _messageFilter.

                                                                                                         {
     _inputQueue = inputQueue;
     _clientListener = NULL;
     _messageFilter = NULL;
 }

gov::fnal::cd::rms::ClientListenerLoop::~ClientListenerLoop ( )

Destructor for the ClientListenerLoop. This sets the stop flag to true so that the thread will exit. The proper way to shutdown the loop is to call the stop() method and then join() on the thread handle. Otherwise, its possible the thread could exit while holding the input queue lock which may cause a crash.

Definition at line 34 of file ClientListenerLoop.cpp.

References novadaq::Runnable::_running, and novadaq::Runnable::_stopRequested.

                                         {
     _stopRequested = true;
 
     while(_running) {
         usleep(100000);
     }
 }

Member Function Documentation

virtual bool novadaq::Runnable::isDone ( ) const

inlinevirtualinherited

Tests if the Runnable object has completed its running.

Returns: true if the object is done, false otherwise.

Definition at line 80 of file Runnable.h.

References novadaq::Runnable::_hasCompleted.

Referenced by novadaq::BackgroundProcess::timedWaitForDone().

                                 {
         return _hasCompleted;
     }

virtual bool novadaq::Runnable::isRunning ( ) const

inlinevirtualinherited

Tests if the Runnable object is running.

Returns: true if the object is running, false otherwise.

Definition at line 71 of file Runnable.h.

References novadaq::Runnable::_running.

                                    {
         return _running;
     }

void gov::fnal::cd::rms::ClientListenerLoop::run ( )

virtual

This method runs in its own thread. If no client listener has been set, it will spin until one is set. After that, it will pull messages off of the queue, deserialize them, and then pass them through a message filter if one has been set. Lastly, the message is passed to the listener.

Implements novadaq::Runnable.

Definition at line 58 of file ClientListenerLoop.cpp.

References _clientListener, _inputQueue, _messageFilter, novadaq::Runnable::_running, novadaq::Runnable::_stopRequested, om::cout, allTimeWatchdog::endl, gov::fnal::cd::rms::RmsMessageListener::messageReceived(), gov::fnal::cd::rms::util::LinkedBlockingQueue< QueueType >::poll(), and gov::fnal::cd::rms::MessageFilter::verify().

                              {
     boost::shared_ptr<std::string> messageText;
 
     _stopRequested = false;
     _running = true;
 
     while (!_stopRequested) {
         if (_clientListener) {
             messageText = _inputQueue->poll(1);
         }
         else {
             usleep(200000);
             continue;
         }
             
         if (!messageText) {
             continue;
         }
         // 19-Jun-2007, KAB - we need to create a new RmsMessage each time
         // since XSD deserialization doesn't overwrite existing element 
         // objects.
         boost::shared_ptr<base::RmsMessage> 
             receivedMessage (new base::RmsMessage);
         
         try {
             receivedMessage->deserialize(*messageText);
         }
         catch (const ::xsd::cxx::tree::exception<char> &excpt) {
             std::cout << "Exception thrown during deserialization ";
             std::cout << "of RMS message!" << std::endl;
             std::cout << excpt << std::endl;
         }
         
         if (_messageFilter) {
             if (_messageFilter->verify(receivedMessage)) {
                 _clientListener->messageReceived(receivedMessage);
             }
         }
         else {
             _clientListener->messageReceived(receivedMessage);
         }
     }
 
     _running = false;
 }

void gov::fnal::cd::rms::ClientListenerLoop::setMessageFilter ( MessageFilter * filter )

Set the message filter that will get called everytime a message is pulled off the queue to verify that it is relevant to the listener.

Parameters

filter The message filter

Definition at line 122 of file ClientListenerLoop.cpp.

References _messageFilter.

Referenced by gov::fnal::cd::rms::RmsProducer::initialize(), and gov::fnal::cd::rms::RmsConsumer::RmsConsumer().

                                                                {
     _messageFilter = filter;
 }

void gov::fnal::cd::rms::ClientListenerLoop::setMessageListener ( RmsMessageListener * listener )

Set the message listener that gets called when a message is received.

Parameters

listener The listener that will get called when a message is received.

Definition at line 111 of file ClientListenerLoop.cpp.

References _clientListener.

Referenced by gov::fnal::cd::rms::RmsConsumer::setMessageListener(), and gov::fnal::cd::rms::RmsProducer::setReplyListener().

                                                                         {
     _clientListener = listener;
 }

void gov::fnal::cd::rms::ClientListenerLoop::stop ( )

virtual

Set the stop flag so that the thread exits. After doing this, join() on the thread handle to make sure that the thread has exited.

Reimplemented from novadaq::Runnable.

Definition at line 47 of file ClientListenerLoop.cpp.

References novadaq::Runnable::_stopRequested.

Referenced by gov::fnal::cd::rms::RmsProducer::close(), and gov::fnal::cd::rms::RmsConsumer::close().

                               {
     _stopRequested = true;
 }

virtual void novadaq::Runnable::watchfulSleep	(	int	microseconds,
		int	numberOfIntervals = `10`
	)

inlinevirtualinherited

Sleeps for the specified amount of time, but periodically checks if a stop has been requested, and returns if so.

Definition at line 88 of file Runnable.h.

References novadaq::Runnable::_stopRequested, and compare_h5_caf::idx.

                                                            {
         if (numberOfIntervals < 1) {numberOfIntervals = 1;}
         int intervalTime = microseconds / numberOfIntervals;
         if (intervalTime < 1) {intervalTime = 1;}
         for (int idx = 0; idx < numberOfIntervals; ++idx) {
             if (_stopRequested) {break;}
             usleep(intervalTime);
         }
     }

Member Data Documentation

RmsMessageListener* gov::fnal::cd::rms::ClientListenerLoop::_clientListener

private

The listener that we will be sending messages to.

Definition at line 50 of file ClientListenerLoop.h.

Referenced by ClientListenerLoop(), run(), and setMessageListener().

bool novadaq::Runnable::_hasCompleted

protectedinherited

Tracks whether the Runnable object has completed its running.

Definition at line 115 of file Runnable.h.

Referenced by novadaq::Runnable::isDone(), novadaq::BackgroundProcess::run(), novadaq::Runnable::Runnable(), novadaq::BackgroundProcess::stop(), novadaq::Runnable::stop(), and novadaq::BackgroundProcess::waitForDone().

util::LinkedBlockingQueue<boost::shared_ptr<std::string> >* gov::fnal::cd::rms::ClientListenerLoop::_inputQueue

private

The queue that we will be pulling messages off of.

Definition at line 45 of file ClientListenerLoop.h.

Referenced by ClientListenerLoop(), and run().

MessageFilter* gov::fnal::cd::rms::ClientListenerLoop::_messageFilter

private

The filter that we will be running messages through to determine if they are relevant to the listener.

Definition at line 56 of file ClientListenerLoop.h.

Referenced by ClientListenerLoop(), run(), and setMessageFilter().

bool novadaq::Runnable::_running

protectedinherited

Tracks whether the Runnable object is running or not (this can be before it has started or after it has finished)

Definition at line 105 of file Runnable.h.

Referenced by novadaq::Runnable::isRunning(), novadaq::BackgroundProcess::run(), run(), NdmcClient::run(), novadaq::Runnable::Runnable(), and ~ClientListenerLoop().

bool novadaq::Runnable::_stopRequested

protectedinherited

Indicates that a request has been made to stop running.

Definition at line 110 of file Runnable.h.

Referenced by run(), NdmcClient::run(), novadaq::Runnable::Runnable(), novadaq::BackgroundProcess::stop(), stop(), novadaq::Runnable::stop(), novadaq::Runnable::watchfulSleep(), and ~ClientListenerLoop().

The documentation for this class was generated from the following files:

/cvmfs/nova-development.opensciencegrid.org/novasoft/releases/N21-01-24/rms/cxx/include/ClientListenerLoop.h
/cvmfs/nova-development.opensciencegrid.org/novasoft/releases/N21-01-24/rms/cxx/src/ClientListenerLoop.cpp

Public Member Functions

Protected Attributes

Private Attributes

Detailed Description

Constructor & Destructor Documentation

Member Function Documentation

Member Data Documentation