Subversion Repositories HomeAutomation

Rev

Rev 1317 | Blame | Compare with Previous | Last modification | View Log | SVN | RSS feed

  1. /*
  2.  * Subscriber.cpp
  3.  *
  4.  *  Created on: Jul 21, 2009
  5.  *      Author: Mattias Runge
  6.  */
  7.  
  8. #include "Subscriber.h"
  9.  
  10. #include "broker/Broker.h"
  11.  
  12. namespace atom {
  13.  
  14. Subscriber::Subscriber(bool receive_from_self)
  15. {
  16.     LOG.setName("Subscriber");
  17.  
  18.     this->receive_from_self_ = receive_from_self;
  19.  
  20.     Broker::GetInstance()->Connect(boost::bind(&Subscriber::Slot_OnMessage, this, _1));
  21. }
  22.  
  23. Subscriber::~Subscriber()
  24. {
  25.     this->Stop();
  26. }
  27.  
  28. void Subscriber::Put(Message::Pointer message)
  29. {
  30.     Broker::GetInstance()->Put(message);
  31. }
  32.  
  33. Message::Pointer CreateMessage()
  34. {
  35.     return Message::Pointer(new Message(static_cast<void *>this));
  36. }
  37.  
  38. void Subscriber::Slot_OnMessage(Message::Åointer message)
  39. {
  40.     if (this->receive_from_self_ || message->GetOrigin() != static_cast<void *>this)
  41.     {
  42.         this->queue_.push(message);
  43.         this->on_message_condition_.notify_all();
  44.     }
  45. }
  46.  
  47. void Subscriber::Run()
  48. {
  49.     while (true)
  50.     {
  51.         boost::mutex::scoped_lock guard(this->guard_mutex_);
  52.  
  53.         try
  54.         {
  55.             while (this->queue_.size() > 0)
  56.             {
  57.                 this->OnMessage(this->queue_.pop());
  58.             }
  59.  
  60.             this->on_message_condition_.wait(guard);
  61.         }
  62.         catch (boost::thread_interrupted e)
  63.         {
  64.             guard.unlock();
  65.             LOG.warn("Thread interrupted");
  66.             break;
  67.         }
  68.     }
  69. }
  70.  
  71. } // namespace atom
  72.