Common logic for a simple "mini broker" that creates creates queues automatically when a client tries to send or subscribe. This file contains the queue
class that queues messages and the broker_handler
class that manages queues and links and transfers messages to/from clients.
#ifndef BROKER_HPP
#define BROKER_HPP
#include "proton/event.hpp"
#include "proton/message.hpp"
#include "proton/handler.hpp"
#include "proton/sender.hpp"
#include "proton/transport.hpp"
#include "proton/url.hpp"
#include <iostream>
#include <deque>
#include <map>
#include <list>
#include <sstream>
class queue {
public:
queue(const std::string &name, bool dynamic = false) : name_(name), dynamic_(dynamic) {}
std::string name() const { return name_; }
consumers_.push_back(s);
}
consumers_.remove(s);
return (consumers_.size() == 0 && (dynamic_ || messages_.size() == 0));
}
messages_.push_back(m);
dispatch(0);
}
while (deliver_to(s)) {}
}
int count = s ? 1 : consumers_.size();
if (!count) return false;
bool result = false;
sender_list::iterator it = consumers_.begin();
if (!s && count) {
s = &*it;
}
while (messages_.size()) {
messages_.pop_front();
result = true;
}
if (--count) {
it++;
} else {
return result;
}
}
return false;
}
private:
typedef std::deque<proton::message> message_queue;
typedef std::list<proton::sender> sender_list;
std::string name_;
bool dynamic_;
message_queue messages_;
sender_list consumers_;
};
class queues {
public:
queues() : next_id_(0) {}
virtual queue &get(const std::string &address = std::string()) {
if (address.empty()) {
throw std::runtime_error("empty queue name");
}
queue*& q = queues_[address];
if (!q) q = new queue(address);
return *q;
}
virtual queue &dynamic() {
std::ostringstream os;
os << "q" << next_id_++;
queue *q = queues_[os.str()] = new queue(os.str(), true);
return *q;
}
virtual void erase(std::string &name) {
delete queues_[name];
queues_.erase(name);
}
protected:
typedef std::map<std::string, queue *> queue_map;
queue_map queues_;
uint64_t next_id_;
};
public:
broker_handler(queues& qs) : queues_(qs) {}
queue &q = remote_source.dynamic() ?
queues_.dynamic() : queues_.get(remote_source.address());
std::cout << "broker outgoing link from " << q.name() << std::endl;
} else {
if (!address.empty()) {
std::cout << "broker incoming link to " << address << std::endl;
}
}
}
if (queues_.get(address).unsubscribe(lnk)) {
queues_.erase(address);
}
}
}
}
}
}
}
std::cerr <<
"broker error: " << e.
name() <<
":" << c.
str() << std::endl;
}
for (proton::link_iterator l = r.begin(); l != r.end(); ++l) {
if (!!l->sender()) {
unsubscribe(l->sender());
}
}
}
queues_.get(address).dispatch(&s);
}
}
protected:
queues& queues_;
};
#endif // BROKER_HPP