#include "options.hpp"
#include <proton/connection.hpp>
#include <proton/container.hpp>
#include <proton/default_container.hpp>
#include <proton/delivery.hpp>
#include <proton/error_condition.hpp>
#include <proton/listener.hpp>
#include <proton/messaging_handler.hpp>
#include <proton/message.hpp>
#include <proton/receiver_options.hpp>
#include <proton/sender.hpp>
#include <proton/sender_options.hpp>
#include <proton/source_options.hpp>
#include <proton/target_options.hpp>
#include <proton/tracker.hpp>
#include <proton/transport.hpp>
#include <proton/url.hpp>
#include <deque>
#include <iostream>
#include <list>
#include <map>
#include <string>
#include "fake_cpp11.hpp"
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 ~queues() {}
virtual queue &get(const std::string &address) {
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_;
int next_id_;
};
public:
broker_handler(queues& qs) : queues_(qs) {}
queue *q;
if (src.dynamic()) {
} else if (!src.address().empty()) {
q = &queues_.get(src.address());
} else {
return;
}
q->subscribe(sender);
std::cout << "broker outgoing link from " << q->name() << std::endl;
}
if (!address.empty()) {
std::cout << "broker incoming link to " << address << std::endl;
} else {
}
}
if (queues_.get(address).unsubscribe(lnk)) {
queues_.erase(address);
}
}
unsubscribe(sender);
}
remove_stale_consumers(c);
}
}
std::cout <<
"broker client disconnect: " << t.
error().
what() << std::endl;
}
std::cerr <<
"broker error: " << c.
what() << std::endl;
}
proton::sender_range r = connection.
senders();
for (proton::sender_iterator i = r.begin(); i != r.end(); ++i) {
if (i->active())
unsubscribe(*i);
}
}
queues_.get(address).dispatch(&s);
}
queues_.get(address).publish(m);
}
protected:
queues& queues_;
};
class broker {
public:
broker(const std::string& url) : handler_(url, queues_) {}
private:
class my_handler : public broker_handler {
public:
my_handler(const std::string& u, queues& qs) : broker_handler(qs), url_(u) {}
std::cout << "broker listening on " << url_ << std::endl;
}
private:
const std::string& url_;
};
private:
queues queues_;
my_handler handler_;
};
int main(int argc, char **argv) {
std::string url("0.0.0.0");
example::options opts(argc, argv);
opts.add_value(url, 'a', "address", "listen on URL", "URL");
try {
opts.parse();
broker b(url);
proton::default_container(b.handler()).run();
return 0;
} catch (const example::bad_option& e) {
std::cout << opts << std::endl << e.what() << std::endl;
} catch (const std::exception& e) {
std::cerr << e.what() << std::endl;
}
return 1;
}