SharedListener.h

Go to the documentation of this file.

// ----------------------------------------------------------------------------
//
// $Id$
//
// Copyright 2008, 2009, 2010, 2011, 2012  Antonio Franchi and Paolo Stegagno    
//
// This file is part of MIP.
//
// MIP is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// MIP is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with MIP. If not, see <http://www.gnu.org/licenses/>.
//
// Contact info: antonio.franchi@tuebingen.mpg.de stegagno@diag.uniroma1.it
//
// ----------------------------------------------------------------------------




#ifndef __SHARED_LISTENER_INCLUDED_
#define __SHARED_LISTENER_INCLUDED_

#ifdef MIP_HOST_APPLE
#include <applePatch.h>
#endif

#include <list>

#include <Time.h>
#include <Message.h>

#include <pthread.h>

using namespace std;
using namespace MipBaselib;

namespace MipBaselib{

 
 /* @{ */
 
template <class T> void* sharedListenerThread(void* p) { 
 T* sl = (T*) p;

 while(true){
  string msg,ip;
 
  sl->_receive(msg,ip);
  Time currTime;
  Message message(msg);
 
  pthread_mutex_lock(&(sl->_mtx));
 
  /*First loop for the received messages*/
  bool handlerFound = false;
  for(int i=0; i < sl->_handlersNum;i++){
   if( message.checkId( sl->_handlerIds.at(i) ) ){
    if( sl->_toBeDelayed[i]){
     sl->_delayedMsgs[i]->push_back(msg);
     sl->_delayedMsgIps[i]->push_back(ip);
     sl->_delayedMsgTimeStamps[i]->push_back(currTime);
    }else{
     sl->_handlers.at(i)->handleMessage(msg,ip);
    }
    handlerFound = true;
   }
  }
  /*Second loop for the delayed messages*/
  for(int i=0; i < sl->_handlersNum;i++){
 //    cout << "TEST " << i << endl;
   int delayedMsgNum = sl->_delayedMsgs[i]->size();
   bool loop = true;
   Time handlerDelay = sl->_handlerDelays[i];
   while(delayedMsgNum > 0 && loop){
    list<Time>::iterator timeStampIt = sl->_delayedMsgTimeStamps[i]->begin();
    if( currTime - *timeStampIt  > handlerDelay){
     list<string>::iterator msgIt = sl->_delayedMsgs[i]->begin();
     list<string>::iterator ipIt  = sl->_delayedMsgIps[i]->begin();
     sl->_handlers.at(i)->handleMessage(*msgIt,*ipIt);
     sl->_delayedMsgs[i]->pop_front();
     sl->_delayedMsgIps[i]->pop_front();
     sl->_delayedMsgTimeStamps[i]->pop_front();
     delayedMsgNum --;
    }else{
     loop = false;
    }
   }
  }
 
  pthread_mutex_unlock(&(sl->_mtx));
 
  if(!handlerFound){
   cout << "Warning: handler not found for the message." << endl;
   cout << msg  << endl;
  }
 }

 return 0;
}
 
class SharedListener {
  
  vector<MessageHandler*> _handlers;
  vector<string>      _handlerIds;
  vector<bool>       _toBeDelayed;
  vector<Time>       _handlerDelays;
  vector< list < string > * > _delayedMsgs;
  vector< list < string > * > _delayedMsgIps;
  vector< list < Time   > * > _delayedMsgTimeStamps;
  int _handlersNum;
  
  pthread_mutex_t _mtx;
//   ACE_thread_t _listThreadID;
  
  template <class T> friend void* sharedListenerThread(void* p);

  virtual void _receive(string& message, string& address) = 0;
  
 protected:
  
  template <class T> void _runThread(void* p){
//    ACE_Thread::spawn((ACE_THR_FUNC)sharedListenerThread<T>,(T*) p, THR_NEW_LWP | THR_JOINABLE, &_listThreadID, &_listThreadHandle);
   pthread_t st;
   pthread_create(&st, NULL,sharedListenerThread<T>,(T*) p);
  }
  
 public:
  
  SharedListener();
  
  ~SharedListener();
  
  bool registerHandler(MessageHandler *handler,string id,Time addedDelay = Time(0,0));
 
};



/* @} */
 
};// end namespace MipResources{


#endif