201704.mbox/raw/%3c686d49236f5546b48d9640530447a834@git.apache.org%3e

 #ifndef BROKER_HPP

- #define BROKER_HPP

-

- /*

-  * Licensed to the Apache Software Foundation (ASF) under one

-  * or more contributor license agreements.  See the NOTICE file

-  * distributed with this work for additional information

-  * regarding copyright ownership.  The ASF licenses this file

-  * to you under the Apache License, Version 2.0 (the

-  * "License"); you may not use this file except in compliance

-  * with the License.  You may obtain a copy of the License at

-  *

-  *   http://www.apache.org/licenses/LICENSE-2.0

-  *

-  * Unless required by applicable law or agreed to in writing,

-  * software distributed under the License is distributed on an

-  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY

-  * KIND, either express or implied.  See the License for the

-  * specific language governing permissions and limitations

-  * under the License.

-  */

-

-

- #include <proton/connection.hpp>

- #include <proton/delivery.hpp>

- #include <proton/messaging_handler.hpp>

- #include <proton/message.hpp>

- #include <proton/sasl.hpp>

- #include <proton/sender.hpp>

- #include <proton/tracker.hpp>

- #include <proton/transport.hpp>

- #include <proton/sender_options.hpp>

- #include <proton/receiver_options.hpp>

- #include <proton/source_options.hpp>

- #include <proton/target_options.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_; }

-

-     void subscribe(proton::sender s) {

-         consumers_.push_back(s);

-     }

-

-     // Return true if queue can be deleted.

-     bool unsubscribe(proton::sender s) {

-         consumers_.remove(s);

-         return (consumers_.size() == 0 && (dynamic_ || messages_.size() == 0));

-     }

-

-     void publish(const proton::message &m) {

-         messages_.push_back(m);

-         dispatch(0);

-     }

-

-     void dispatch(proton::sender *s) {

-         while (deliver_to(s)) {}

-     }

-

-     bool deliver_to(proton::sender *s) {

-         // Deliver to single sender if supplied, else all consumers

-         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()) {

-             if (s->credit()) {

-                 const proton::message& m = messages_.front();

-

-                 s->send(m);

-                 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() {}

-

-     // Get or create a queue.

-     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;

-     }

-

-     // Create a dynamic queue with a unique name.

-     virtual queue &dynamic() {

-         std::ostringstream os;

-         os << "q" << next_id_++;

-         queue *q = queues_[os.str()] = new queue(os.str(), true);

-

-         return *q;

-     }

-

-     // Delete the named queue

-     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_; // Use to generate unique queue IDs.

- };

-

- #include <proton/config.hpp>

-

- class broker_handler : public proton::messaging_handler {

-   public:

-     broker_handler(queues& qs) : queues_(qs) {}

-

-     void on_transport_open(proton::transport &t) OVERRIDE {

-         std::cout << "Connection from user: " << t.sasl().user() << " (mechanism: " << t.sasl().mech() << ")" << std::endl;

-     }

-

-     void on_sender_open(proton::sender &sender) OVERRIDE {

-         proton::source src(sender.source());

-         queue &q = src.dynamic() ?

-             queues_.dynamic() : queues_.get(src.address());

-         sender.open(proton::sender_options().source(proton::source_options().address(q.name())));

-         q.subscribe(sender);

-         std::cout << "broker outgoing link from " << q.name() << std::endl;

-     }

-

-     void on_receiver_open(proton::receiver &receiver) OVERRIDE {

-         std::string address = receiver.target().address();

-         if (!address.empty()) {

-             receiver.open(proton::receiver_options().target(proton::target_options().address(address)));

-             std::cout << "broker incoming link to " << address << std::endl;

-         }

-     }

-

-     void unsubscribe(proton::sender lnk) {

-         std::string address = lnk.source().address();

-

-         if (queues_.get(address).unsubscribe(lnk)) {

-             queues_.erase(address);

-         }

-     }

-

-     void on_sender_close(proton::sender &sender) OVERRIDE {

-         unsubscribe(sender);

-     }

-

-     void on_connection_close(proton::connection &c) OVERRIDE {

-         remove_stale_consumers(c);

-     }

-

-     void on_transport_close(proton::transport &t) OVERRIDE {

-         remove_stale_consumers(t.connection());

-     }

-

-     void on_transport_error(proton::transport &t) OVERRIDE {

-         std::cout << "broker client disconnect: " << t.error().what() << std::endl;

-     }

-

-     void on_error(const proton::error_condition &c) OVERRIDE {

-         std::cerr << "broker error: " << c.what() << std::endl;

-     }

-

-     void remove_stale_consumers(proton::connection connection) {

-         proton::sender_range sr = connection.senders();

-         for (proton::sender_iterator i = sr.begin(); i != sr.end(); ++i) {

-             if (i->active())

-                 unsubscribe(*i);

-         }

-     }

-

-     void on_sendable(proton::sender &s) OVERRIDE {

-         std::string address = s.source().address();

-

-         queues_.get(address).dispatch(&s);

-     }

-

-     void on_message(proton::delivery &d, proton::message &m) OVERRIDE {

-         std::string address = d.receiver().target().address();

-         queues_.get(address).publish(m);

-     }

-

-   protected:

-     queues& queues_;

- };

-

- #endif // BROKER_HPP

-proton::link::credit
int credit() const
Credit available on the link.

-proton::message
An AMQP message.
Definition: message.hpp:49

-proton::sender
A channel for sending messages.
Definition: sender.hpp:40

-proton::sender_options
Options for creating a sender.
Definition: sender_options.hpp:64

-broker_handler
Common handler logic for brokers.
Definition: broker.hpp:160

-broker_handler::on_receiver_open
void on_receiver_open(proton::receiver &receiver) OVERRIDE
The remote peer opened the link.
Definition: broker.hpp:177

-broker_handler::on_sendable
void on_sendable(proton::sender &s) OVERRIDE
A message can be sent.
Definition: broker.hpp:221

-broker_handler::on_error
void on_error(const proton::error_condition &c) OVERRIDE
Fallback error handling.
Definition: broker.hpp:209

-proton::connection
A connection to a remote AMQP peer.
Definition: connection.hpp:48

-proton::source
A point of origin for messages.
Definition: source.hpp:41

-broker_handler::on_sender_close
void on_sender_close(proton::sender &sender) OVERRIDE
The remote peer closed the link.
Definition: broker.hpp:193

-broker_handler::on_connection_close
void on_connection_close(proton::connection &c) OVERRIDE
The remote peer closed the connection.
Definition: broker.hpp:197

-queue
A simple implementation of a queue.
Definition: broker.hpp:50

-proton::delivery
A received message.
Definition: delivery.hpp:40

-proton::sender::source
class source source() const
Get the source node.

-broker_handler::on_sender_open
void on_sender_open(proton::sender &sender) OVERRIDE
The remote peer opened the link.
Definition: broker.hpp:168

-queues
A collection of queues and queue factory, used by a broker.
Definition: broker.hpp:118

-broker_handler::on_message
void on_message(proton::delivery &d, proton::message &m) OVERRIDE
A message is received.
Definition: broker.hpp:227

-proton::source_options
Options for creating a source node for a sender or receiver.
Definition: source_options.hpp:45

-broker_handler::on_transport_error
void on_transport_error(proton::transport &t) OVERRIDE
The underlying network transport has closed with an error condition.
Definition: broker.hpp:205

-proton::connection::senders
sender_range senders() const
Return all senders on this connection.

-proton::receiver_options
Options for creating a receiver.
Definition: receiver_options.hpp:62

-broker_handler::on_transport_close
void on_transport_close(proton::transport &t) OVERRIDE
The underlying network transport has closed.
Definition: broker.hpp:201

-proton::receiver
A channel for receiving messages.
Definition: receiver.hpp:39

-proton::source::address
std::string address() const
The address of the source.

-broker_handler::on_transport_open
void on_transport_open(proton::transport &t) OVERRIDE
The underlying network transport is open.
Definition: broker.hpp:164

-proton::messaging_handler
A handler for Proton messaging events.
Definition: messaging_handler.hpp:75

-proton::transport
A network channel supporting an AMQP connection.
Definition: transport.hpp:42

-proton::target_options
Options for creating a target node for a sender or receiver.
Definition: target_options.hpp:47

-proton::sender::send
tracker send(const message &m)
Send a message on the sender.

-proton::error_condition
Describes an endpoint error state.
Definition: error_condition.hpp:37

-
-Classes
class	queue
	A simple implementation of a queue. More...

class	queues
	A collection of queues and queue factory, used by a broker. More...

class	broker_handler
	Common handler logic for brokers. More...
broker_handler(queues &qs) (defined in broker_handler)	broker_handler	inline
messaging_handler() (defined in messaging_handler)	messaging_handler
on_connection_close(proton::connection &c) OVERRIDE	broker_handler	inlinevirtual
on_connection_error(connection &c)	messaging_handler	virtual
on_connection_open(connection &c)	messaging_handler	virtual
on_container_start(container &c)	messaging_handler	virtual
on_delivery_settle(delivery &d)	messaging_handler	virtual
on_error(const proton::error_condition &c) OVERRIDE	broker_handler	inlinevirtual
on_message(proton::delivery &d, proton::message &m) OVERRIDE	broker_handler	inlinevirtual
on_receiver_close(receiver &l)	messaging_handler	virtual
on_receiver_drain_finish(receiver &r)	messaging_handler	virtual
on_receiver_error(receiver &l)	messaging_handler	virtual
on_receiver_open(proton::receiver &receiver) OVERRIDE	broker_handler	inlinevirtual
on_sendable(proton::sender &s) OVERRIDE	broker_handler	inlinevirtual
on_sender_close(proton::sender &sender) OVERRIDE	broker_handler	inlinevirtual
on_sender_drain_start(sender &s)	messaging_handler	virtual
on_sender_error(sender &l)	messaging_handler	virtual
on_sender_open(proton::sender &sender) OVERRIDE	broker_handler	inlinevirtual
on_session_close(session &s)	messaging_handler	virtual
on_session_error(session &s)	messaging_handler	virtual
on_session_open(session &s)	messaging_handler	virtual
on_tracker_accept(tracker &d)	messaging_handler	virtual
on_tracker_reject(tracker &d)	messaging_handler	virtual
on_tracker_release(tracker &d)	messaging_handler	virtual
on_tracker_settle(tracker &d)	messaging_handler	virtual
on_transport_close(proton::transport &t) OVERRIDE	broker_handler	inlinevirtual
on_transport_error(proton::transport &t) OVERRIDE	broker_handler	inlinevirtual
on_transport_open(proton::transport &t) OVERRIDE	broker_handler	inlinevirtual
remove_stale_consumers(proton::connection connection) (defined in broker_handler)	broker_handler	inline
unsubscribe(proton::sender lnk) (defined in broker_handler)	broker_handler	inline
~messaging_handler() (defined in messaging_handler)	messaging_handler	virtual