Subversion Repositories HomeAutomation

Rev

Rev 1317 | Show entire file | Regard whitespace | Details | Blame | Last modification | View Log | SVN | RSS feed

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