http://git-wip-us.apache.org/repos/asf/qpid-site/blob/701caf21/content/releases/qpid-proton-master/proton/cpp/api/broker_8cpp-example.html ---------------------------------------------------------------------- diff --git a/content/releases/qpid-proton-master/proton/cpp/api/broker_8cpp-example.html b/content/releases/qpid-proton-master/proton/cpp/api/broker_8cpp-example.html index 9d11304..85378d7 100755 --- a/content/releases/qpid-proton-master/proton/cpp/api/broker_8cpp-example.html +++ b/content/releases/qpid-proton-master/proton/cpp/api/broker_8cpp-example.html @@ -93,8 +93,8 @@ $(document).ready(function(){initNavTree('broker_8cpp-example.html','');}); <div class="title">broker.cpp</div> </div> </div><!--header--> <div class="contents"> -<p>A broker using the <code><a class="el" href="classproton_1_1container.html" title="A top-level container of connections, sessions, and links. ">proton::container</a></code>. You can use this to run other examples that reqiure an intermediary, or you can use any AMQP 1.0 broker. This broker creates queues automatically when a client tries to send or subscribe.</p> -<div class="fragment"><div class="line"><span class="comment">/*</span></div><div class="line"><span class="comment"> * Licensed to the Apache Software Foundation (ASF) under one</span></div><div class="line"><span class="comment"> * or more contributor license agreements. See the NOTICE file</span></div><div class="line"><span class="comment"> * distributed with this work for additional information</span></div><div class="line"><span class="comment"> * regarding copyright ownership. The ASF licenses this file</span></div><div class="line"><span class="comment"> * to you under the Apache License, Version 2.0 (the</span></div><div class="line"><span class="comment"> * "License"); you may not use this file except in compliance</span></div><div class="line"><span class="comment"> * with the License. You may obtain a copy of the License at</span></div><div class="line"><span class="comment"> *</span></div><div class="line"><span class="comment"> * http://www.apache.org/li censes/LICENSE-2.0</span></div><div class="line"><span class="comment"> *</span></div><div class="line"><span class="comment"> * Unless required by applicable law or agreed to in writing,</span></div><div class="line"><span class="comment"> * software distributed under the License is distributed on an</span></div><div class="line"><span class="comment"> * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY</span></div><div class="line"><span class="comment"> * KIND, either express or implied. See the License for the</span></div><div class="line"><span class="comment"> * specific language governing permissions and limitations</span></div><div class="line"><span class="comment"> * under the License.</span></div><div class="line"><span class="comment"> */</span></div><div class="line"></div><div class="line"><span class="preprocessor">#include "options.hpp"</span></div><div class="line"></div><div class="line"><span class="preprocessor">#include <<a class="co de" href="connection_8hpp.html">proton/connection.hpp</a>></span></div><div class="line"><span class="preprocessor">#include <<a class="code" href="connection__options_8hpp.html">proton/connection_options.hpp</a>></span></div><div class="line"><span class="preprocessor">#include <<a class="code" href="container_8hpp.html">proton/container.hpp</a>></span></div><div class="line"><span class="preprocessor">#include <<a class="code" href="delivery_8hpp.html">proton/delivery.hpp</a>></span></div><div class="line"><span class="preprocessor">#include <<a class="code" href="error__condition_8hpp.html">proton/error_condition.hpp</a>></span></div><div class="line"><span class="preprocessor">#include <<a class="code" href="function_8hpp.html">proton/function.hpp</a>></span></div><div class="line"><span class="preprocessor">#include <<a class="code" href="listen__handler_8hpp.html">proton/listen_handler.hpp</a>></span></div><div class="line"><span class="p reprocessor">#include <<a class="code" href="listener_8hpp.html">proton/listener.hpp</a>></span></div><div class="line"><span class="preprocessor">#include <<a class="code" href="message_8hpp.html">proton/message.hpp</a>></span></div><div class="line"><span class="preprocessor">#include <<a class="code" href="messaging__handler_8hpp.html">proton/messaging_handler.hpp</a>></span></div><div class="line"><span class="preprocessor">#include <<a class="code" href="receiver__options_8hpp.html">proton/receiver_options.hpp</a>></span></div><div class="line"><span class="preprocessor">#include <<a class="code" href="sender__options_8hpp.html">proton/sender_options.hpp</a>></span></div><div class="line"><span class="preprocessor">#include <<a class="code" href="source__options_8hpp.html">proton/source_options.hpp</a>></span></div><div class="line"><span class="preprocessor">#include <<a class="code" href="target_8hpp.html">proton/target.hpp</a>></span ></div><div class="line"><span class="preprocessor">#include <<a >class="code" >href="target__options_8hpp.html">proton/target_options.hpp</a>></span></div><div > class="line"><span class="preprocessor">#include <<a class="code" >href="tracker_8hpp.html">proton/tracker.hpp</a>></span></div><div >class="line"><span class="preprocessor">#include <<a class="code" >href="transport_8hpp.html">proton/transport.hpp</a>></span></div><div >class="line"><span class="preprocessor">#include <<a class="code" >href="work__queue_8hpp.html">proton/work_queue.hpp</a>></span></div><div >class="line"></div><div class="line"><span class="preprocessor">#include ><deque></span></div><div class="line"><span >class="preprocessor">#include <iostream></span></div><div >class="line"><span class="preprocessor">#include <map></span></div><div >class="line"><span class="preprocessor">#include ><string></span></div><div class="line"></div><div class="line"><span >class="preprocesso r">#if PN_CPP_SUPPORTS_THREADS</span></div><div class="line"><span class="preprocessor">#include <thread></span></div><div class="line"><span class="preprocessor">#endif</span></div><div class="line"></div><div class="line"><span class="preprocessor">#include "fake_cpp11.hpp"</span></div><div class="line"></div><div class="line"><span class="comment">// This is a simplified model for a message broker, that only allows for messages to go to a</span></div><div class="line"><span class="comment">// single receiver.</span></div><div class="line"><span class="comment">//</span></div><div class="line"><span class="comment">// This broker is multithread safe and if compiled with C++11 with a multithreaded Proton</span></div><div class="line"><span class="comment">// binding library will use as many threads as there are thread resources available (usually</span></div><div class="line"><span class="comment">// cores)</span></div><div class="line"><span class="comment">//</spa n></div><div class="line"><span class="comment">// Queues are only created and never destroyed</span></div><div class="line"><span class="comment">//</span></div><div class="line"><span class="comment">// Broker Entities (that need to be individually serialised)</span></div><div class="line"><span class="comment">// QueueManager - Creates new queues, finds queues</span></div><div class="line"><span class="comment">// Queue - Queues msgs, records subscribers, sends msgs to subscribers</span></div><div class="line"><span class="comment">// Connection - Receives Messages from network, sends messages to network.</span></div><div class="line"></div><div class="line"><span class="comment">// Work</span></div><div class="line"><span class="comment">// FindQueue(queueName, connection) - From a Connection to the QueueManager</span></div><div class="line"><span class="comment">// This will create the queue if it doesn't already exist and send a BoundQueue</span></div><div cla ss="line"><span class="comment">// message back to the connection.</span></div><div class="line"><span class="comment">// BoundQueue(queue) - From the QueueManager to a Connection</span></div><div class="line"><span class="comment">//</span></div><div class="line"><span class="comment">// QueueMsg(msg) - From a Connection (receiver) to a Queue</span></div><div class="line"><span class="comment">// Subscribe(sender) - From a Connection (sender) to a Queue</span></div><div class="line"><span class="comment">// Flow(sender, credit) - From a Connection (sender) to a Queue</span></div><div class="line"><span class="comment">// Unsubscribe(sender) - From a Connection (sender) to a Queue</span></div><div class="line"><span class="comment">//</span></div><div class="line"><span class="comment">// SendMsg(msg) - From a Queue to a Connection (sender)</span></div><div class="line"><span class="comment">// Unsubscribed() - From a Queue to a Connection (sender)</span></div><div class="line"></div><div class="line"></div><div class="line"><span class="comment">// Simple debug output</span></div><div class="line"><span class="keywordtype">bool</span> verbose;</div><div class="line"><span class="preprocessor">#define DOUT(x) do {if (verbose) {x};} while (false)</span></div><div class="line"></div><div class="line"><span class="keyword">class </span>Queue;</div><div class="line"><span class="keyword">class </span>Sender;</div><div class="line"></div><div class="line"><span class="keyword">typedef</span> std::map<proton::sender, Sender*> senders;</div><div class="line"></div><div class="line"><span class="keyword">class </span>Sender : <span class="keyword">public</span> <a name="_a0"></a><a class="code" href="classproton_1_1messaging__handler.html">proton::messaging_handler</a> {</div><div class="line"> <span class="keyword">friend</span> <span class="keyword">class </span>connection_handler;</div><div class="line"></div><div class="line"> <a name= "_a1"></a><a class="code" href="classproton_1_1sender.html">proton::sender</a> sender_;</div><div class="line"> senders& senders_;</div><div class="line"> <a name="_a2"></a><a class="code" href="classproton_1_1work__queue.html">proton::work_queue</a>& work_queue_;</div><div class="line"> std::string queue_name_;</div><div class="line"> Queue* queue_;</div><div class="line"> <span class="keywordtype">int</span> pending_credit_;</div><div class="line"></div><div class="line"> <span class="comment">// Messaging handlers</span></div><div class="line"> <span class="keywordtype">void</span> <a name="a3"></a><a class="code" href="classproton_1_1messaging__handler.html#aa24f522a68cdf382762702cece7790e7">on_sendable</a>(<a class="code" href="classproton_1_1sender.html">proton::sender</a> &sender) OVERRIDE;</div><div class="line"> <span class="keywordtype">void</span> <a name="a4"></a><a class="code" href="classproton_1_1messaging__handler.html#a7942b98699b d21e827041626ee1b5c84">on_sender_close</a>(<a class="code" href="classproton_1_1sender.html">proton::sender</a> &sender) OVERRIDE;</div><div class="line"></div><div class="line"><span class="keyword">public</span>:</div><div class="line"> Sender(<a class="code" href="classproton_1_1sender.html">proton::sender</a> s, senders& ss) :</div><div class="line"> sender_(s), senders_(ss), work_queue_(s.work_queue()), queue_(0), pending_credit_(0)</div><div class="line"> {}</div><div class="line"></div><div class="line"> <span class="keywordtype">bool</span> add(<a name="_a5"></a><a class="code" href="classproton_1_1work.html">proton::work</a> f) {</div><div class="line"> <span class="keywordflow">return</span> work_queue_.<a name="a6"></a><a class="code" href="classproton_1_1work__queue.html#a59dae2153455bc095477a3b66a0b681e">add</a>(f);</div><div class="line"> }</div><div class="line"></div><div class="line"></div><div class="line"> <span class="keywordt ype">void</span> boundQueue(Queue* q, std::string qn);</div><div class="line"> <span class="keywordtype">void</span> sendMsg(<a name="_a7"></a><a class="code" href="classproton_1_1message.html">proton::message</a> m) {</div><div class="line"> DOUT(std::cerr << <span class="stringliteral">"Sender: "</span> << <span class="keyword">this</span> << <span class="stringliteral">" sending\n"</span>;);</div><div class="line"> sender_.<a name="a8"></a><a class="code" href="classproton_1_1sender.html#a214eb30b24e6831d016a47b9dddda830">send</a>(m);</div><div class="line"> }</div><div class="line"> <span class="keywordtype">void</span> unsubscribed() {</div><div class="line"> DOUT(std::cerr << <span class="stringliteral">"Sender: "</span> << <span class="keyword">this</span> << <span class="stringliteral">" deleting\n"</span>;);</div><div class="line"> <span class="keyword">delet e</span> <span class="keyword">this</span>;</div><div class="line"> }</div><div class="line">};</div><div class="line"></div><div class="line"><span class="comment">// Queue - round robin subscriptions</span></div><div class="line"><span class="keyword">class </span>Queue {</div><div class="line"> <a class="code" href="classproton_1_1work__queue.html">proton::work_queue</a> work_queue_;</div><div class="line"> <span class="keyword">const</span> std::string name_;</div><div class="line"> std::deque<proton::message> messages_;</div><div class="line"> <span class="keyword">typedef</span> std::map<Sender*, int> subscriptions; <span class="comment">// With credit</span></div><div class="line"> subscriptions subscriptions_;</div><div class="line"> subscriptions::iterator current_;</div><div class="line"></div><div class="line"> <span class="keywordtype">void</span> tryToSend() {</div><div class="line"> DOUT(std::cerr << <span class="stringl iteral">"Queue: "</span> << <span class="keyword">this</span> << <span class="stringliteral">" tryToSend: "</span> << subscriptions_.size(););</div><div class="line"> <span class="comment">// Starting at current_, send messages to subscriptions with credit:</span></div><div class="line"> <span class="comment">// After each send try to find another subscription; Wrap around;</span></div><div class="line"> <span class="comment">// Finish when we run out of messages or credit.</span></div><div class="line"> <span class="keywordtype">size_t</span> outOfCredit = 0;</div><div class="line"> <span class="keywordflow">while</span> (!messages_.empty() && outOfCredit<subscriptions_.size()) {</div><div class="line"> <span class="comment">// If we got the end (or haven't started yet) start at the beginning</span></div><div class="line"> <span class="keywordflow">if</span> (current_==s ubscriptions_.end()) {</div><div class="line"> current_=subscriptions_.begin();</div><div class="line"> }</div><div class="line"> <span class="comment">// If we have credit send the message</span></div><div class="line"> DOUT(std::cerr << <span class="stringliteral">"("</span> << current_->second << <span class="stringliteral">") "</span>;);</div><div class="line"> <span class="keywordflow">if</span> (current_->second>0) {</div><div class="line"> DOUT(std::cerr << current_->first << <span class="stringliteral">" "</span>;);</div><div class="line"> <a name="a9"></a><a class="code" href="namespaceproton.html#ae77a8638d584f5fc99a1bbaf218d2481">proton::schedule_work</a>(current_->first, &Sender::sendMsg, current_->first, messages_.front());</div><div class="line"> messages_.pop_front();</div><div class=" line"> --current_->second;</div><div class="line"> ++current_;</div><div class="line"> } <span class="keywordflow">else</span> {</div><div class="line"> ++outOfCredit;</div><div class="line"> }</div><div class="line"> }</div><div class="line"> DOUT(std::cerr << <span class="stringliteral">"\n"</span>;);</div><div class="line"> }</div><div class="line"></div><div class="line"><span class="keyword">public</span>:</div><div class="line"> Queue(<a name="_a10"></a><a class="code" href="classproton_1_1container.html">proton::container</a>& c, <span class="keyword">const</span> std::string& n) :</div><div class="line"> work_queue_(c), name_(n), current_(subscriptions_.end())</div><div class="line"> {}</div><div class="line"></div><div class="line"> <span class="keywordtype">bool</span> add(<a class="code" href="classproton_1_1work.html">proton::work</a> f) {</div>< div class="line"> <span class="keywordflow">return</span> work_queue_.<a class="code" href="classproton_1_1work__queue.html#a59dae2153455bc095477a3b66a0b681e">add</a>(f);</div><div class="line"> }</div><div class="line"></div><div class="line"> <span class="keywordtype">void</span> queueMsg(<a class="code" href="classproton_1_1message.html">proton::message</a> m) {</div><div class="line"> DOUT(std::cerr << <span class="stringliteral">"Queue: "</span> << <span class="keyword">this</span> << <span class="stringliteral">"("</span> << name_ << <span class="stringliteral">") queueMsg\n"</span>;);</div><div class="line"> messages_.push_back(m);</div><div class="line"> tryToSend();</div><div class="line"> }</div><div class="line"> <span class="keywordtype">void</span> flow(Sender* s, <span class="keywordtype">int</span> c) {</div><div class="line"> DOUT(std::cerr << <span cla ss="stringliteral">"Queue: "</span> << <span class="keyword">this</span> << <span class="stringliteral">"("</span> << name_ << <span class="stringliteral">") flow: "</span> << c << <span class="stringliteral">" to "</span> << s << <span class="stringliteral">"\n"</span>;);</div><div class="line"> subscriptions_[s] = c;</div><div class="line"> tryToSend();</div><div class="line"> }</div><div class="line"> <span class="keywordtype">void</span> subscribe(Sender* s) {</div><div class="line"> DOUT(std::cerr << <span class="stringliteral">"Queue: "</span> << <span class="keyword">this</span> << <span class="stringliteral">"("</span> << name_ << <span class="stringliteral">") subscribe Sender: "</span> << s << <span class="stringliteral">"\n"</span>;);</div><div class="line"> subscriptions_[s] = 0;</div><div class="line"> }</div><div class="line"> <span class="keywordtype">void</span> unsubscribe(Sender* s) {</div><div class="line"> DOUT(std::cerr << <span class="stringliteral">"Queue: "</span> << <span class="keyword">this</span> << <span class="stringliteral">"("</span> << name_ << <span class="stringliteral">") unsubscribe Sender: "</span> << s << <span class="stringliteral">"\n"</span>;);</div><div class="line"> <span class="comment">// If we're about to erase the current subscription move on</span></div><div class="line"> <span class="keywordflow">if</span> (current_ != subscriptions_.end() && current_->first==s) ++current_;</div><div class="line"> subscriptions_.erase(s);</div><div class="line"> <a class="code" href="namespaceproton.html#ae77a8638d584f5fc99a1bbaf218d2481">proton::schedule_work</a>(s, &S ender::unsubscribed, s);</div><div class="line"> }</div><div class="line">};</div><div class="line"></div><div class="line"><span class="comment">// We have credit to send a message.</span></div><div class="line"><span class="keywordtype">void</span> Sender::on_sendable(<a class="code" href="classproton_1_1sender.html">proton::sender</a> &sender) {</div><div class="line"> <span class="keywordflow">if</span> (queue_) {</div><div class="line"> <a class="code" href="namespaceproton.html#ae77a8638d584f5fc99a1bbaf218d2481">proton::schedule_work</a>(queue_, &Queue::flow, queue_, <span class="keyword">this</span>, sender.<a name="a11"></a><a class="code" href="classproton_1_1link.html#afd27bd11ba72d7df51c44f71b15749eb">credit</a>());</div><div class="line"> } <span class="keywordflow">else</span> {</div><div class="line"> pending_credit_ = sender.<a class="code" href="classproton_1_1link.html#afd27bd11ba72d7df51c44f71b15749eb">credit</a>();</div><div class="l ine"> }</div><div class="line">}</div><div class="line"></div><div class="line"><span class="keywordtype">void</span> Sender::on_sender_close(<a class="code" href="classproton_1_1sender.html">proton::sender</a> &sender) {</div><div class="line"> <span class="keywordflow">if</span> (queue_) {</div><div class="line"> <a class="code" href="namespaceproton.html#ae77a8638d584f5fc99a1bbaf218d2481">proton::schedule_work</a>(queue_, &Queue::unsubscribe, queue_, <span class="keyword">this</span>);</div><div class="line"> } <span class="keywordflow">else</span> {</div><div class="line"> <span class="comment">// TODO: Is it possible to be closed before we get the queue allocated?</span></div><div class="line"> <span class="comment">// If so, we should have a way to mark the sender deleted, so we can delete</span></div><div class="line"> <span class="comment">// on queue binding</span></div><div class="line"> }</div><div class="line"> senders_. erase(sender);</div><div class="line">}</div><div class="line"></div><div class="line"><span class="keywordtype">void</span> Sender::boundQueue(Queue* q, std::string qn) {</div><div class="line"> DOUT(std::cerr << <span class="stringliteral">"Sender: "</span> << <span class="keyword">this</span> << <span class="stringliteral">" bound to Queue: "</span> << q <<<span class="stringliteral">"("</span> << qn << <span class="stringliteral">")\n"</span>;);</div><div class="line"> queue_ = q;</div><div class="line"> queue_name_ = qn;</div><div class="line"></div><div class="line"> <a class="code" href="namespaceproton.html#ae77a8638d584f5fc99a1bbaf218d2481">proton::schedule_work</a>(q, &Queue::subscribe, q, <span class="keyword">this</span>);</div><div class="line"> sender_.<a name="a12"></a><a class="code" href="classproton_1_1sender.html#a9e8555112049fc2b4945120b3c45f8ab">open</a>(<a nam e="_a13"></a><a class="code" href="classproton_1_1sender__options.html">proton::sender_options</a>()</div><div class="line"> .source((<a name="_a14"></a><a class="code" href="classproton_1_1source__options.html">proton::source_options</a>().address(queue_name_)))</div><div class="line"> .handler(*<span class="keyword">this</span>));</div><div class="line"> <span class="keywordflow">if</span> (pending_credit_>0) {</div><div class="line"> <a class="code" href="namespaceproton.html#ae77a8638d584f5fc99a1bbaf218d2481">proton::schedule_work</a>(queue_, &Queue::flow, queue_, <span class="keyword">this</span>, pending_credit_);</div><div class="line"> }</div><div class="line"> std::cout << <span class="stringliteral">"sending from "</span> << queue_name_ << std::endl;</div><div class="line">}</div><div class="line"></div><div class="line"><span class="keyword">class </span>Receiver : <span class="keyword">public</span> <a clas s="code" href="classproton_1_1messaging__handler.html">proton::messaging_handler</a> {</div><div class="line"> <span class="keyword">friend</span> <span class="keyword">class </span>connection_handler;</div><div class="line"></div><div class="line"> <a name="_a15"></a><a class="code" href="classproton_1_1receiver.html">proton::receiver</a> receiver_;</div><div class="line"> <a class="code" href="classproton_1_1work__queue.html">proton::work_queue</a>& work_queue_;</div><div class="line"> Queue* queue_;</div><div class="line"> std::deque<proton::message> messages_;</div><div class="line"></div><div class="line"> <span class="comment">// A message is received.</span></div><div class="line"> <span class="keywordtype">void</span> on_message(<a name="_a16"></a><a class="code" href="classproton_1_1delivery.html">proton::delivery</a> &, <a class="code" href="classproton_1_1message.html">proton::message</a> &m) OVERRIDE {</div><div class="line"> messages_.push_back(m);</div><div class="line"></div><div class="line"> <span class="keywordflow">if</span> (queue_) {</div><div class="line"> queueMsgs();</div><div class="line"> }</div><div class="line"> }</div><div class="line"></div><div class="line"> <span class="keywordtype">void</span> queueMsgs() {</div><div class="line"> DOUT(std::cerr << <span class="stringliteral">"Receiver: "</span> << <span class="keyword">this</span> << <span class="stringliteral">" queueing "</span> << messages_.size() << <span class="stringliteral">" msgs to: "</span> << queue_ << <span class="stringliteral">"\n"</span>;);</div><div class="line"> <span class="keywordflow">while</span> (!messages_.empty()) {</div><div class="line"> <a class="code" href="namespaceproton.html#ae77a8638d584f5fc99a1bbaf218d2481">proton::schedule_work</a>(queue_, &Queue::queueMsg, q ueue_, messages_.front());</div><div class="line"> messages_.pop_front();</div><div class="line"> }</div><div class="line"> }</div><div class="line"></div><div class="line"><span class="keyword">public</span>:</div><div class="line"> Receiver(<a class="code" href="classproton_1_1receiver.html">proton::receiver</a> r) :</div><div class="line"> receiver_(r), work_queue_(r.work_queue()), queue_(0)</div><div class="line"> {}</div><div class="line"></div><div class="line"> <span class="keywordtype">bool</span> add(<a class="code" href="classproton_1_1work.html">proton::work</a> f) {</div><div class="line"> <span class="keywordflow">return</span> work_queue_.<a class="code" href="classproton_1_1work__queue.html#a59dae2153455bc095477a3b66a0b681e">add</a>(f);</div><div class="line"> }</div><div class="line"></div><div class="line"> <span class="keywordtype">void</span> boundQueue(Queue* q, std::string qn) {</div><div class="line"> DOU T(std::cerr << <span class="stringliteral">"Receiver: "</span> << <span class="keyword">this</span> << <span class="stringliteral">" bound to Queue: "</span> << q << <span class="stringliteral">"("</span> << qn << <span class="stringliteral">")\n"</span>;);</div><div class="line"> queue_ = q;</div><div class="line"> receiver_.<a name="a17"></a><a class="code" href="classproton_1_1receiver.html#a9e8555112049fc2b4945120b3c45f8ab">open</a>(<a name="_a18"></a><a class="code" href="classproton_1_1receiver__options.html">proton::receiver_options</a>()</div><div class="line"> .source((<a class="code" href="classproton_1_1source__options.html">proton::source_options</a>().address(qn)))</div><div class="line"> .handler(*<span class="keyword">this</span>));</div><div class="line"> std::cout << <span class="stringliteral">"receiving to "</span> << q n << std::endl;</div><div class="line"></div><div class="line"> queueMsgs();</div><div class="line"> }</div><div class="line">};</div><div class="line"></div><div class="line"><span class="keyword">class </span>QueueManager {</div><div class="line"> <a class="code" href="classproton_1_1container.html">proton::container</a>& container_;</div><div class="line"> <a class="code" href="classproton_1_1work__queue.html">proton::work_queue</a> work_queue_;</div><div class="line"> <span class="keyword">typedef</span> std::map<std::string, Queue*> queues;</div><div class="line"> queues queues_;</div><div class="line"> <span class="keywordtype">int</span> next_id_; <span class="comment">// Use to generate unique queue IDs.</span></div><div class="line"></div><div class="line"><span class="keyword">public</span>:</div><div class="line"> QueueManager(<a class="code" href="classproton_1_1container.html">proton::container</a>& c) :</div><div class="l ine"> container_(c), work_queue_(c), next_id_(0)</div><div class="line"> {}</div><div class="line"></div><div class="line"> <span class="keywordtype">bool</span> add(<a class="code" href="classproton_1_1work.html">proton::work</a> f) {</div><div class="line"> <span class="keywordflow">return</span> work_queue_.<a class="code" href="classproton_1_1work__queue.html#a59dae2153455bc095477a3b66a0b681e">add</a>(f);</div><div class="line"> }</div><div class="line"></div><div class="line"> <span class="keyword">template</span> <<span class="keyword">class</span> T></div><div class="line"> <span class="keywordtype">void</span> findQueue(T& connection, std::string& qn) {</div><div class="line"> <span class="keywordflow">if</span> (qn.empty()) {</div><div class="line"> <span class="comment">// Dynamic queue creation</span></div><div class="line"> std::ostringstream os;</div><div class="line"> os << <span c lass="stringliteral">"_dynamic_"</span> << next_id_++;</div><div class="line"> qn = os.str();</div><div class="line"> }</div><div class="line"> Queue* q = 0;</div><div class="line"> queues::iterator i = queues_.find(qn);</div><div class="line"> <span class="keywordflow">if</span> (i==queues_.end()) {</div><div class="line"> q = <span class="keyword">new</span> Queue(container_, qn);</div><div class="line"> queues_[qn] = q;</div><div class="line"> } <span class="keywordflow">else</span> {</div><div class="line"> q = i->second;</div><div class="line"> }</div><div class="line"> <a class="code" href="namespaceproton.html#ae77a8638d584f5fc99a1bbaf218d2481">proton::schedule_work</a>(&connection, &T::boundQueue, &connection, q, qn);</div><div class="line"> }</div><div class="line"></div><div class="line"> <span class="keywordtype">void</span> findQueueSender(Sen der* s, std::string qn) {</div><div class="line"> findQueue(*s, qn);</div><div class="line"> }</div><div class="line"></div><div class="line"> <span class="keywordtype">void</span> findQueueReceiver(Receiver* r, std::string qn) {</div><div class="line"> findQueue(*r, qn);</div><div class="line"> }</div><div class="line">};</div><div class="line"></div><div class="line"><span class="keyword">class </span>connection_handler : <span class="keyword">public</span> <a class="code" href="classproton_1_1messaging__handler.html">proton::messaging_handler</a> {</div><div class="line"> QueueManager& queue_manager_;</div><div class="line"> senders senders_;</div><div class="line"></div><div class="line"><span class="keyword">public</span>:</div><div class="line"> connection_handler(QueueManager& qm) :</div><div class="line"> queue_manager_(qm)</div><div class="line"> {}</div><div class="line"></div><div class="line"> <span class="keywordtype" >void</span> on_connection_open(<a name="_a19"></a><a class="code" >href="classproton_1_1connection.html">proton::connection</a>& c) OVERRIDE >{</div><div class="line"> c.open(); <span >class="comment">// Accept the connection</span></div><div class="line"> >}</div><div class="line"></div><div class="line"> <span class="comment">// >A sender sends messages from a queue to a subscriber.</span></div><div >class="line"> <span class="keywordtype">void</span> on_sender_open(<a >class="code" href="classproton_1_1sender.html">proton::sender</a> >&sender) OVERRIDE {</div><div class="line"> std::string qn = >sender.<a name="a20"></a><a class="code" >href="classproton_1_1sender.html#a91a9e8a9445b29d83dd0514cd76503ae">source</a>().<a > name="a21"></a><a class="code" >href="classproton_1_1terminus.html#adcc4ef24adb8478230018c519aa636ec">dynamic</a>() > ? <span class="stringliteral">""</span> : sender.<a class="code" >href="classproton_1_1sender.html#a91a9e8a9 445b29d83dd0514cd76503ae">source</a>().<a name="a22"></a><a class="code" href="classproton_1_1source.html#a4744eb217c976c199b678bb5a0d55acf">address</a>();</div><div class="line"> Sender* s = <span class="keyword">new</span> Sender(sender, senders_);</div><div class="line"> senders_[sender] = s;</div><div class="line"> <a class="code" href="namespaceproton.html#ae77a8638d584f5fc99a1bbaf218d2481">proton::schedule_work</a>(&queue_manager_, &QueueManager::findQueueSender, &queue_manager_, s, qn);</div><div class="line"> }</div><div class="line"></div><div class="line"> <span class="comment">// A receiver receives messages from a publisher to a queue.</span></div><div class="line"> <span class="keywordtype">void</span> on_receiver_open(<a class="code" href="classproton_1_1receiver.html">proton::receiver</a> &receiver) OVERRIDE {</div><div class="line"> std::string qname = receiver.target().address();</div><div class="line"> <sp an class="keywordflow">if</span> (qname == <span class="stringliteral">"shutdown"</span>) {</div><div class="line"> std::cout << <span class="stringliteral">"broker shutting down"</span> << std::endl;</div><div class="line"> <span class="comment">// Sending to the special "shutdown" queue stops the broker.</span></div><div class="line"> receiver.connection().container().stop(</div><div class="line"> <a name="_a23"></a><a class="code" href="classproton_1_1error__condition.html">proton::error_condition</a>(<span class="stringliteral">"shutdown"</span>, <span class="stringliteral">"stop broker"</span>));</div><div class="line"> } <span class="keywordflow">else</span> {</div><div class="line"> <span class="keywordflow">if</span> (qname.empty()) {</div><div class="line"> DOUT(std::cerr << <span class="stringliteral">"ODD - trying to att ach to a empty address\n"</span>;);</div><div class="line"> }</div><div class="line"> Receiver* r = <span class="keyword">new</span> Receiver(receiver);</div><div class="line"> <a class="code" href="namespaceproton.html#ae77a8638d584f5fc99a1bbaf218d2481">proton::schedule_work</a>(&queue_manager_, &QueueManager::findQueueReceiver, &queue_manager_, r, qname);</div><div class="line"> }</div><div class="line"> }</div><div class="line"></div><div class="line"> <span class="keywordtype">void</span> on_session_close(<a name="_a24"></a><a class="code" href="classproton_1_1session.html">proton::session</a> &session) OVERRIDE {</div><div class="line"> <span class="comment">// Unsubscribe all senders that belong to session.</span></div><div class="line"> <span class="keywordflow">for</span> (proton::sender_iterator i = session.senders().begin(); i != session.senders().end(); ++i) {</div><div class="line"> senders::iterator j = senders_.find(*i);</div><div class="line"> <span class="keywordflow">if</span> (j == senders_.end()) <span class="keywordflow">continue</span>;</div><div class="line"> Sender* s = j->second;</div><div class="line"> <span class="keywordflow">if</span> (s->queue_) {</div><div class="line"> <a class="code" href="namespaceproton.html#ae77a8638d584f5fc99a1bbaf218d2481">proton::schedule_work</a>(s->queue_, &Queue::unsubscribe, s->queue_, s);</div><div class="line"> }</div><div class="line"> senders_.erase(j);</div><div class="line"> }</div><div class="line"> }</div><div class="line"></div><div class="line"> <span class="keywordtype">void</span> on_error(<span class="keyword">const</span> <a class="code" href="classproton_1_1error__condition.html">proton::error_condition</a>& e) OVERRIDE {</div><div class="line"> std::cerr << <span class="stringlitera l">"error: "</span> << e.what() << std::endl;</div><div class="line"> }</div><div class="line"></div><div class="line"> <span class="comment">// The container calls on_transport_close() last.</span></div><div class="line"> <span class="keywordtype">void</span> on_transport_close(<a name="_a25"></a><a class="code" href="classproton_1_1transport.html">proton::transport</a>& t) OVERRIDE {</div><div class="line"> <span class="comment">// Unsubscribe all senders.</span></div><div class="line"> <span class="keywordflow">for</span> (proton::sender_iterator i = t.connection().senders().begin(); i != t.connection().senders().end(); ++i) {</div><div class="line"> senders::iterator j = senders_.find(*i);</div><div class="line"> <span class="keywordflow">if</span> (j == senders_.end()) <span class="keywordflow">continue</span>;</div><div class="line"> Sender* s = j->second;</div><div class="line"> <s pan class="keywordflow">if</span> (s->queue_) {</div><div class="line"> <a class="code" href="namespaceproton.html#ae77a8638d584f5fc99a1bbaf218d2481">proton::schedule_work</a>(s->queue_, &Queue::unsubscribe, s->queue_, s);</div><div class="line"> }</div><div class="line"> }</div><div class="line"> <span class="keyword">delete</span> <span class="keyword">this</span>; <span class="comment">// All done.</span></div><div class="line"> }</div><div class="line">};</div><div class="line"></div><div class="line"><span class="keyword">class </span>broker {</div><div class="line"> <span class="keyword">public</span>:</div><div class="line"> broker(<span class="keyword">const</span> std::string addr) :</div><div class="line"> container_(<span class="stringliteral">"broker"</span>), queues_(container_), listener_(queues_)</div><div class="line"> {</div><div class="line"> container_.<a name="a26">< /a><a class="code" href="classproton_1_1container.html#a9e138b28e9589583915cf5c5e0e7a524">listen</a>(addr, listener_);</div><div class="line"> std::cout << <span class="stringliteral">"broker listening on "</span> << addr << std::endl;</div><div class="line"> }</div><div class="line"></div><div class="line"> <span class="keywordtype">void</span> run() {</div><div class="line"><span class="preprocessor">#if PN_CPP_SUPPORTS_THREADS</span></div><div class="line"> std::cout << <span class="stringliteral">"starting "</span> << std::thread::hardware_concurrency() << <span class="stringliteral">" listening threads\n"</span>;</div><div class="line"> container_.<a name="a27"></a><a class="code" href="classproton_1_1container.html#a13a43e6d814de94978c515cb084873b1">run</a>(std::thread::hardware_concurrency());</div><div class="line"><span class="preprocessor">#else</span></div><div class="line"> container_.<a class="code" href="classproton_1_1container.html#a13a43e6d814de94978c515cb084873b1">run</a>();</div><div class="line"><span class="preprocessor">#endif</span></div><div class="line"> }</div><div class="line"></div><div class="line"> <span class="keyword">private</span>:</div><div class="line"> <span class="keyword">struct </span>listener : <span class="keyword">public</span> <a name="_a28"></a><a class="code" href="classproton_1_1listen__handler.html">proton::listen_handler</a> {</div><div class="line"> listener(QueueManager& c) : queues_(c) {}</div><div class="line"></div><div class="line"> <a name="_a29"></a><a class="code" href="classproton_1_1connection__options.html">proton::connection_options</a> on_accept(<a name="_a30"></a><a class="code" href="classproton_1_1listener.html">proton::listener</a>&) OVERRIDE{</div><div class="line"> <span class="keywordflow">return</span> <a class="code" href="classproton_1_1connection__op tions.html">proton::connection_options</a>().<a name="a31"></a><a class="code" href="classproton_1_1connection__options.html#a7bca23aeb3455378ef2d35975758e504">handler</a>(*(<span class="keyword">new</span> connection_handler(queues_)));</div><div class="line"> }</div><div class="line"></div><div class="line"> <span class="keywordtype">void</span> on_error(<a class="code" href="classproton_1_1listener.html">proton::listener</a>&, <span class="keyword">const</span> std::string& s) OVERRIDE {</div><div class="line"> std::cerr << <span class="stringliteral">"listen error: "</span> << s << std::endl;</div><div class="line"> <span class="keywordflow">throw</span> std::runtime_error(s);</div><div class="line"> }</div><div class="line"> QueueManager& queues_;</div><div class="line"> };</div><div class="line"></div><div class="line"> <a class="code" href="classproton_1_1container.html">proton::co ntainer</a> container_;</div><div class="line"> QueueManager queues_;</div><div class="line"> listener listener_;</div><div class="line">};</div><div class="line"></div><div class="line"><span class="keywordtype">int</span> main(<span class="keywordtype">int</span> argc, <span class="keywordtype">char</span> **argv) {</div><div class="line"> <span class="comment">// Command line options</span></div><div class="line"> std::string address(<span class="stringliteral">"0.0.0.0"</span>);</div><div class="line"> example::options opts(argc, argv);</div><div class="line"></div><div class="line"> opts.add_flag(verbose, <span class="charliteral">'v'</span>, <span class="stringliteral">"verbose"</span>, <span class="stringliteral">"verbose (debugging) output"</span>);</div><div class="line"> opts.add_value(address, <span class="charliteral">'a'</span>, <span class="stringliteral">"address"</span>, <span class="stringl iteral">"listen on URL"</span>, <span class="stringliteral">"URL"</span>);</div><div class="line"></div><div class="line"> <span class="keywordflow">try</span> {</div><div class="line"> verbose = <span class="keyword">false</span>;</div><div class="line"> opts.parse();</div><div class="line"> broker(address).run();</div><div class="line"> <span class="keywordflow">return</span> 0;</div><div class="line"> } <span class="keywordflow">catch</span> (<span class="keyword">const</span> example::bad_option& e) {</div><div class="line"> std::cout << opts << std::endl << e.what() << std::endl;</div><div class="line"> } <span class="keywordflow">catch</span> (<span class="keyword">const</span> std::exception& e) {</div><div class="line"> std::cerr << <span class="stringliteral">"broker shutdown: "</span> << e.what() << std::endl;</div><div class="line"> }</div> <div class="line"> <span class="keywordflow">return</span> 1;</div><div class="line">}</div></div><!-- fragment --> </div><!-- contents --> +<p>A broker using the <code><a class="el" href="classproton_1_1container.html" title="A top-level container of connections, sessions, and links. ">proton::container</a></code>. You can use this to run other examples that require an intermediary, or you can use any AMQP 1.0 broker. This broker creates queues automatically when a client tries to send or subscribe.</p> +<div class="fragment"><div class="line"><span class="comment">/*</span></div><div class="line"><span class="comment"> * Licensed to the Apache Software Foundation (ASF) under one</span></div><div class="line"><span class="comment"> * or more contributor license agreements. See the NOTICE file</span></div><div class="line"><span class="comment"> * distributed with this work for additional information</span></div><div class="line"><span class="comment"> * regarding copyright ownership. The ASF licenses this file</span></div><div class="line"><span class="comment"> * to you under the Apache License, Version 2.0 (the</span></div><div class="line"><span class="comment"> * "License"); you may not use this file except in compliance</span></div><div class="line"><span class="comment"> * with the License. You may obtain a copy of the License at</span></div><div class="line"><span class="comment"> *</span></div><div class="line"><span class="comment"> * http://www.apache.org/li censes/LICENSE-2.0</span></div><div class="line"><span class="comment"> *</span></div><div class="line"><span class="comment"> * Unless required by applicable law or agreed to in writing,</span></div><div class="line"><span class="comment"> * software distributed under the License is distributed on an</span></div><div class="line"><span class="comment"> * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY</span></div><div class="line"><span class="comment"> * KIND, either express or implied. See the License for the</span></div><div class="line"><span class="comment"> * specific language governing permissions and limitations</span></div><div class="line"><span class="comment"> * under the License.</span></div><div class="line"><span class="comment"> */</span></div><div class="line"></div><div class="line"><span class="preprocessor">#include "options.hpp"</span></div><div class="line"></div><div class="line"><span class="preprocessor">#include <<a class="co de" href="connection_8hpp.html">proton/connection.hpp</a>></span></div><div class="line"><span class="preprocessor">#include <<a class="code" href="connection__options_8hpp.html">proton/connection_options.hpp</a>></span></div><div class="line"><span class="preprocessor">#include <<a class="code" href="container_8hpp.html">proton/container.hpp</a>></span></div><div class="line"><span class="preprocessor">#include <<a class="code" href="delivery_8hpp.html">proton/delivery.hpp</a>></span></div><div class="line"><span class="preprocessor">#include <<a class="code" href="error__condition_8hpp.html">proton/error_condition.hpp</a>></span></div><div class="line"><span class="preprocessor">#include <<a class="code" href="listen__handler_8hpp.html">proton/listen_handler.hpp</a>></span></div><div class="line"><span class="preprocessor">#include <<a class="code" href="listener_8hpp.html">proton/listener.hpp</a>></span></div><div class="line"><span class="p reprocessor">#include <<a class="code" href="message_8hpp.html">proton/message.hpp</a>></span></div><div class="line"><span class="preprocessor">#include <<a class="code" href="messaging__handler_8hpp.html">proton/messaging_handler.hpp</a>></span></div><div class="line"><span class="preprocessor">#include <<a class="code" href="receiver__options_8hpp.html">proton/receiver_options.hpp</a>></span></div><div class="line"><span class="preprocessor">#include <<a class="code" href="sender__options_8hpp.html">proton/sender_options.hpp</a>></span></div><div class="line"><span class="preprocessor">#include <<a class="code" href="source__options_8hpp.html">proton/source_options.hpp</a>></span></div><div class="line"><span class="preprocessor">#include <<a class="code" href="target_8hpp.html">proton/target.hpp</a>></span></div><div class="line"><span class="preprocessor">#include <<a class="code" href="target__options_8hpp.html">proton/target_options.hpp< /a>></span></div><div class="line"><span class="preprocessor">#include <<a class="code" href="tracker_8hpp.html">proton/tracker.hpp</a>></span></div><div class="line"><span class="preprocessor">#include <<a class="code" href="transport_8hpp.html">proton/transport.hpp</a>></span></div><div class="line"><span class="preprocessor">#include <<a class="code" href="work__queue_8hpp.html">proton/work_queue.hpp</a>></span></div><div class="line"></div><div class="line"><span class="preprocessor">#include <deque></span></div><div class="line"><span class="preprocessor">#include <iostream></span></div><div class="line"><span class="preprocessor">#include <map></span></div><div class="line"><span class="preprocessor">#include <string></span></div><div class="line"></div><div class="line"><span class="preprocessor">#if PN_CPP_HAS_STD_THREAD</span></div><div class="line"><span class="preprocessor">#include <thread></span></div><div class="line"> </div><div class="line"><span class="keywordtype">int</span> hardware_concurrency() {<span class="keywordflow">return</span> std::thread::hardware_concurrency();}</div><div class="line"><span class="preprocessor">#else</span></div><div class="line"><span class="keywordtype">int</span> hardware_concurrency() {<span class="keywordflow">return</span> 1;}</div><div class="line"><span class="preprocessor">#endif</span></div><div class="line"></div><div class="line"><span class="preprocessor">#include "fake_cpp11.hpp"</span></div><div class="line"></div><div class="line"><span class="comment">// This is a simplified model for a message broker, that only allows for</span></div><div class="line"><span class="comment">// messages to go to a single receiver.</span></div><div class="line"><span class="comment">//</span></div><div class="line"><span class="comment">// This broker is multithread safe and if compiled with C++11 with a multithreaded Proton</span></div><div class="line">< span class="comment">// binding library will use as many threads as there are thread resources available (usually</span></div><div class="line"><span class="comment">// cores)</span></div><div class="line"><span class="comment">//</span></div><div class="line"><span class="comment">// Queues are only created and never destroyed</span></div><div class="line"><span class="comment">//</span></div><div class="line"><span class="comment">// Broker Entities (that need to be individually serialised)</span></div><div class="line"><span class="comment">// QueueManager - Creates new queues, finds queues</span></div><div class="line"><span class="comment">// Queue - Queues msgs, records subscribers, sends msgs to subscribers</span></div><div class="line"><span class="comment">// Connection - Receives Messages from network, sends messages to network.</span></div><div class="line"></div><div class="line"><span class="comment">// Work</span></div><div class="line"><span class="comment">/ / FindQueue(queueName, connection) - From a Connection to the QueueManager</span></div><div class="line"><span class="comment">// This will create the queue if it doesn't already exist and send a BoundQueue</span></div><div class="line"><span class="comment">// message back to the connection.</span></div><div class="line"><span class="comment">// BoundQueue(queue) - From the QueueManager to a Connection</span></div><div class="line"><span class="comment">//</span></div><div class="line"><span class="comment">// QueueMsg(msg) - From a Connection (receiver) to a Queue</span></div><div class="line"><span class="comment">// Subscribe(sender) - From a Connection (sender) to a Queue</span></div><div class="line"><span class="comment">// Flow(sender, credit) - From a Connection (sender) to a Queue</span></div><div class="line"><span class="comment">// Unsubscribe(sender) - From a Connection (sender) to a Queue</span></div><div class="line"><span class="comment">//</s pan></div><div class="line"><span class="comment">// SendMsg(msg) - From a Queue to a Connection (sender)</span></div><div class="line"><span class="comment">// Unsubscribed() - From a Queue to a Connection (sender)</span></div><div class="line"></div><div class="line"></div><div class="line"><span class="comment">// Simple debug output</span></div><div class="line"><span class="keywordtype">bool</span> verbose;</div><div class="line"><span class="preprocessor">#define DOUT(x) do {if (verbose) {x};} while (false)</span></div><div class="line"></div><div class="line"><span class="keyword">class </span>Queue;</div><div class="line"><span class="keyword">class </span>Sender;</div><div class="line"></div><div class="line"><span class="keyword">typedef</span> std::map<proton::sender, Sender*> senders;</div><div class="line"></div><div class="line"><span class="keyword">class </span>Sender : <span class="keyword">public</span> <a name="_a0"></a><a class="code" href="classproton_1_ 1messaging__handler.html">proton::messaging_handler</a> {</div><div class="line"> <span class="keyword">friend</span> <span class="keyword">class </span>connection_handler;</div><div class="line"></div><div class="line"> <a name="_a1"></a><a class="code" href="classproton_1_1sender.html">proton::sender</a> sender_;</div><div class="line"> senders& senders_;</div><div class="line"> <a name="_a2"></a><a class="code" href="classproton_1_1work__queue.html">proton::work_queue</a>& work_queue_;</div><div class="line"> std::string queue_name_;</div><div class="line"> Queue* queue_;</div><div class="line"> <span class="keywordtype">int</span> pending_credit_;</div><div class="line"></div><div class="line"> <span class="comment">// Messaging handlers</span></div><div class="line"> <span class="keywordtype">void</span> <a name="a3"></a><a class="code" href="classproton_1_1messaging__handler.html#a86c9509ba3ce1925150c5b7a9a937c94">on_sendable</a>(<a class="co de" href="classproton_1_1sender.html">proton::sender</a> &sender) OVERRIDE;</div><div class="line"> <span class="keywordtype">void</span> <a name="a4"></a><a class="code" href="classproton_1_1messaging__handler.html#aa604b2d3dad610ded346b0dcf3ec6f29">on_sender_close</a>(<a class="code" href="classproton_1_1sender.html">proton::sender</a> &sender) OVERRIDE;</div><div class="line"></div><div class="line"><span class="keyword">public</span>:</div><div class="line"> Sender(<a class="code" href="classproton_1_1sender.html">proton::sender</a> s, senders& ss) :</div><div class="line"> sender_(s), senders_(ss), work_queue_(s.<a name="a5"></a><a class="code" href="classproton_1_1link.html#a7c755d6ac6385e007adb61966598ba63">work_queue</a>()), queue_(0), pending_credit_(0)</div><div class="line"> {}</div><div class="line"></div><div class="line"> <span class="keywordtype">bool</span> add(proton::work f) {</div><div class="line"> <span class="keywordflow">r eturn</span> work_queue_.<a name="a6"></a><a class="code" href="classproton_1_1work__queue.html#a59dae2153455bc095477a3b66a0b681e">add</a>(f);</div><div class="line"> }</div><div class="line"></div><div class="line"></div><div class="line"> <span class="keywordtype">void</span> boundQueue(Queue* q, std::string qn);</div><div class="line"> <span class="keywordtype">void</span> sendMsg(<a name="_a7"></a><a class="code" href="classproton_1_1message.html">proton::message</a> m) {</div><div class="line"> DOUT(std::cerr << <span class="stringliteral">"Sender: "</span> << <span class="keyword">this</span> << <span class="stringliteral">" sending\n"</span>;);</div><div class="line"> sender_.<a name="a8"></a><a class="code" href="classproton_1_1sender.html#a214eb30b24e6831d016a47b9dddda830">send</a>(m);</div><div class="line"> }</div><div class="line"> <span class="keywordtype">void</span> unsubscribed() {</div><div class=" line"> DOUT(std::cerr << <span class="stringliteral">"Sender: "</span> << <span class="keyword">this</span> << <span class="stringliteral">" deleting\n"</span>;);</div><div class="line"> <span class="keyword">delete</span> <span class="keyword">this</span>;</div><div class="line"> }</div><div class="line">};</div><div class="line"></div><div class="line"><span class="comment">// Queue - round robin subscriptions</span></div><div class="line"><span class="keyword">class </span>Queue {</div><div class="line"> <a class="code" href="classproton_1_1work__queue.html">proton::work_queue</a> work_queue_;</div><div class="line"> <span class="keyword">const</span> std::string name_;</div><div class="line"> std::deque<proton::message> messages_;</div><div class="line"> <span class="keyword">typedef</span> std::map<Sender*, int> subscriptions; <span class="comment">// With credit</span></div><div class="line"> s ubscriptions subscriptions_;</div><div class="line"> subscriptions::iterator current_;</div><div class="line"></div><div class="line"> <span class="keywordtype">void</span> tryToSend() {</div><div class="line"> DOUT(std::cerr << <span class="stringliteral">"Queue: "</span> << <span class="keyword">this</span> << <span class="stringliteral">" tryToSend: "</span> << subscriptions_.size(););</div><div class="line"> <span class="comment">// Starting at current_, send messages to subscriptions with credit:</span></div><div class="line"> <span class="comment">// After each send try to find another subscription; Wrap around;</span></div><div class="line"> <span class="comment">// Finish when we run out of messages or credit.</span></div><div class="line"> <span class="keywordtype">size_t</span> outOfCredit = 0;</div><div class="line"> <span class="keywordflow">while</span> (!messages_.empty() &am p;& outOfCredit<subscriptions_.size()) {</div><div class="line"> <span class="comment">// If we got the end (or haven't started yet) start at the beginning</span></div><div class="line"> <span class="keywordflow">if</span> (current_==subscriptions_.end()) {</div><div class="line"> current_=subscriptions_.begin();</div><div class="line"> }</div><div class="line"> <span class="comment">// If we have credit send the message</span></div><div class="line"> DOUT(std::cerr << <span class="stringliteral">"("</span> << current_->second << <span class="stringliteral">") "</span>;);</div><div class="line"> <span class="keywordflow">if</span> (current_->second>0) {</div><div class="line"> DOUT(std::cerr << current_->first << <span class="stringliteral">" "</span>;);</div><div class="line"> current_->f irst->add(make_work(&Sender::sendMsg, current_->first, messages_.front()));</div><div class="line"> messages_.pop_front();</div><div class="line"> --current_->second;</div><div class="line"> ++current_;</div><div class="line"> } <span class="keywordflow">else</span> {</div><div class="line"> ++outOfCredit;</div><div class="line"> }</div><div class="line"> }</div><div class="line"> DOUT(std::cerr << <span class="stringliteral">"\n"</span>;);</div><div class="line"> }</div><div class="line"></div><div class="line"><span class="keyword">public</span>:</div><div class="line"> Queue(<a name="_a9"></a><a class="code" href="classproton_1_1container.html">proton::container</a>& c, <span class="keyword">const</span> std::string& n) :</div><div class="line"> work_queue_(c), name_(n), current_(subscriptions_.end())</div><div class="line"> {}</div> <div class="line"></div><div class="line"> <span class="keywordtype">bool</span> add(proton::work f) {</div><div class="line"> <span class="keywordflow">return</span> work_queue_.<a class="code" href="classproton_1_1work__queue.html#a59dae2153455bc095477a3b66a0b681e">add</a>(f);</div><div class="line"> }</div><div class="line"></div><div class="line"> <span class="keywordtype">void</span> queueMsg(<a class="code" href="classproton_1_1message.html">proton::message</a> m) {</div><div class="line"> DOUT(std::cerr << <span class="stringliteral">"Queue: "</span> << <span class="keyword">this</span> << <span class="stringliteral">"("</span> << name_ << <span class="stringliteral">") queueMsg\n"</span>;);</div><div class="line"> messages_.push_back(m);</div><div class="line"> tryToSend();</div><div class="line"> }</div><div class="line"> <span class="keywordtype">void</span> flow(Sende r* s, <span class="keywordtype">int</span> c) {</div><div class="line"> DOUT(std::cerr << <span class="stringliteral">"Queue: "</span> << <span class="keyword">this</span> << <span class="stringliteral">"("</span> << name_ << <span class="stringliteral">") flow: "</span> << c << <span class="stringliteral">" to "</span> << s << <span class="stringliteral">"\n"</span>;);</div><div class="line"> subscriptions_[s] = c;</div><div class="line"> tryToSend();</div><div class="line"> }</div><div class="line"> <span class="keywordtype">void</span> subscribe(Sender* s) {</div><div class="line"> DOUT(std::cerr << <span class="stringliteral">"Queue: "</span> << <span class="keyword">this</span> << <span class="stringliteral">"("</span> << name_ << <span class="stringliteral">") subscribe Sender: &qu ot;</span> << s << <span class="stringliteral">"\n"</span>;);</div><div class="line"> subscriptions_[s] = 0;</div><div class="line"> }</div><div class="line"> <span class="keywordtype">void</span> unsubscribe(Sender* s) {</div><div class="line"> DOUT(std::cerr << <span class="stringliteral">"Queue: "</span> << <span class="keyword">this</span> << <span class="stringliteral">"("</span> << name_ << <span class="stringliteral">") unsubscribe Sender: "</span> << s << <span class="stringliteral">"\n"</span>;);</div><div class="line"> <span class="comment">// If we're about to erase the current subscription move on</span></div><div class="line"> <span class="keywordflow">if</span> (current_ != subscriptions_.end() && current_->first==s) ++current_;</div><div class="line"> subscriptions_.erase(s);</div><div class="line"> s ->add(make_work(&Sender::unsubscribed, s));</div><div class="line"> }</div><div class="line">};</div><div class="line"></div><div class="line"><span class="comment">// We have credit to send a message.</span></div><div class="line"><span class="keywordtype">void</span> Sender::on_sendable(<a class="code" href="classproton_1_1sender.html">proton::sender</a> &sender) {</div><div class="line"> <span class="keywordflow">if</span> (queue_) {</div><div class="line"> queue_->add(make_work(&Queue::flow, queue_, <span class="keyword">this</span>, sender.<a name="a10"></a><a class="code" href="classproton_1_1link.html#afd27bd11ba72d7df51c44f71b15749eb">credit</a>()));</div><div class="line"> } <span class="keywordflow">else</span> {</div><div class="line"> pending_credit_ = sender.<a class="code" href="classproton_1_1link.html#afd27bd11ba72d7df51c44f71b15749eb">credit</a>();</div><div class="line"> }</div><div class="line">}</div><div class="line">< /div><div class="line"><span class="keywordtype">void</span> Sender::on_sender_close(<a class="code" href="classproton_1_1sender.html">proton::sender</a> &sender) {</div><div class="line"> <span class="keywordflow">if</span> (queue_) {</div><div class="line"> queue_->add(make_work(&Queue::unsubscribe, queue_, <span class="keyword">this</span>));</div><div class="line"> } <span class="keywordflow">else</span> {</div><div class="line"> <span class="comment">// TODO: Is it possible to be closed before we get the queue allocated?</span></div><div class="line"> <span class="comment">// If so, we should have a way to mark the sender deleted, so we can delete</span></div><div class="line"> <span class="comment">// on queue binding</span></div><div class="line"> }</div><div class="line"> senders_.erase(sender);</div><div class="line">}</div><div class="line"></div><div class="line"><span class="keywordtype">void</span> Sender::boundQueue(Qu eue* q, std::string qn) {</div><div class="line"> DOUT(std::cerr << <span class="stringliteral">"Sender: "</span> << <span class="keyword">this</span> << <span class="stringliteral">" bound to Queue: "</span> << q <<<span class="stringliteral">"("</span> << qn << <span class="stringliteral">")\n"</span>;);</div><div class="line"> queue_ = q;</div><div class="line"> queue_name_ = qn;</div><div class="line"></div><div class="line"> q->add(make_work(&Queue::subscribe, q, <span class="keyword">this</span>));</div><div class="line"> sender_.open(<a name="_a11"></a><a class="code" href="classproton_1_1sender__options.html">proton::sender_options</a>()</div><div class="line"> .source((<a name="_a12"></a><a class="code" href="classproton_1_1source__options.html">proton::source_options</a>().address(queue_name_)))</div><div class="line"> .handler(*<span class="keyword">this </span>));</div><div class="line"> <span class="keywordflow">if</span> (pending_credit_>0) {</div><div class="line"> queue_->add(make_work(&Queue::flow, queue_, <span class="keyword">this</span>, pending_credit_));</div><div class="line"> }</div><div class="line"> std::cout << <span class="stringliteral">"sending from "</span> << queue_name_ << std::endl;</div><div class="line">}</div><div class="line"></div><div class="line"><span class="keyword">class </span>Receiver : <span class="keyword">public</span> <a class="code" href="classproton_1_1messaging__handler.html">proton::messaging_handler</a> {</div><div class="line"> <span class="keyword">friend</span> <span class="keyword">class </span>connection_handler;</div><div class="line"></div><div class="line"> <a name="_a13"></a><a class="code" href="classproton_1_1receiver.html">proton::receiver</a> receiver_;</div><div class="line"> <a class="code" href="classproton_1_1w ork__queue.html">proton::work_queue</a>& work_queue_;</div><div class="line"> Queue* queue_;</div><div class="line"> std::deque<proton::message> messages_;</div><div class="line"></div><div class="line"> <span class="comment">// A message is received.</span></div><div class="line"> <span class="keywordtype">void</span> on_message(<a name="_a14"></a><a class="code" href="classproton_1_1delivery.html">proton::delivery</a> &, <a class="code" href="classproton_1_1message.html">proton::message</a> &m) OVERRIDE {</div><div class="line"> messages_.push_back(m);</div><div class="line"></div><div class="line"> <span class="keywordflow">if</span> (queue_) {</div><div class="line"> queueMsgs();</div><div class="line"> }</div><div class="line"> }</div><div class="line"></div><div class="line"> <span class="keywordtype">void</span> queueMsgs() {</div><div class="line"> DOUT(std::cerr << <span class="stringliteral">& quot;Receiver: "</span> << <span class="keyword">this</span> << <span class="stringliteral">" queueing "</span> << messages_.size() << <span class="stringliteral">" msgs to: "</span> << queue_ << <span class="stringliteral">"\n"</span>;);</div><div class="line"> <span class="keywordflow">while</span> (!messages_.empty()) {</div><div class="line"> queue_->add(make_work(&Queue::queueMsg, queue_, messages_.front()));</div><div class="line"> messages_.pop_front();</div><div class="line"> }</div><div class="line"> }</div><div class="line"></div><div class="line"><span class="keyword">public</span>:</div><div class="line"> Receiver(<a class="code" href="classproton_1_1receiver.html">proton::receiver</a> r) :</div><div class="line"> receiver_(r), work_queue_(r.<a class="code" href="classproton_1_1link.html#a7c755d6ac6385e007adb61966598ba63">work_queue</a>()), queue _(0)</div><div class="line"> {}</div><div class="line"></div><div class="line"> <span class="keywordtype">bool</span> add(proton::work f) {</div><div class="line"> <span class="keywordflow">return</span> work_queue_.<a class="code" href="classproton_1_1work__queue.html#a59dae2153455bc095477a3b66a0b681e">add</a>(f);</div><div class="line"> }</div><div class="line"></div><div class="line"> <span class="keywordtype">void</span> boundQueue(Queue* q, std::string qn) {</div><div class="line"> DOUT(std::cerr << <span class="stringliteral">"Receiver: "</span> << <span class="keyword">this</span> << <span class="stringliteral">" bound to Queue: "</span> << q << <span class="stringliteral">"("</span> << qn << <span class="stringliteral">")\n"</span>;);</div><div class="line"> queue_ = q;</div><div class="line"> receiver_.<a name="a15"></a><a class="code" href="classproton_ 1_1receiver.html#a9e8555112049fc2b4945120b3c45f8ab">open</a>(<a name="_a16"></a><a class="code" href="classproton_1_1receiver__options.html">proton::receiver_options</a>()</div><div class="line"> .source((<a class="code" href="classproton_1_1source__options.html">proton::source_options</a>().address(qn)))</div><div class="line"> .handler(*<span class="keyword">this</span>));</div><div class="line"> std::cout << <span class="stringliteral">"receiving to "</span> << qn << std::endl;</div><div class="line"></div><div class="line"> queueMsgs();</div><div class="line"> }</div><div class="line">};</div><div class="line"></div><div class="line"><span class="keyword">class </span>QueueManager {</div><div class="line"> <a class="code" href="classproton_1_1container.html">proton::container</a>& container_;</div><div class="line"> <a class="code" href="classproton_1_1work__queue.html">proton::work_queue</a> work_queue _;</div><div class="line"> <span class="keyword">typedef</span> std::map<std::string, Queue*> queues;</div><div class="line"> queues queues_;</div><div class="line"> <span class="keywordtype">int</span> next_id_; <span class="comment">// Use to generate unique queue IDs.</span></div><div class="line"></div><div class="line"><span class="keyword">public</span>:</div><div class="line"> QueueManager(<a class="code" href="classproton_1_1container.html">proton::container</a>& c) :</div><div class="line"> container_(c), work_queue_(c), next_id_(0)</div><div class="line"> {}</div><div class="line"></div><div class="line"> <span class="keywordtype">bool</span> add(proton::work f) {</div><div class="line"> <span class="keywordflow">return</span> work_queue_.<a class="code" href="classproton_1_1work__queue.html#a59dae2153455bc095477a3b66a0b681e">add</a>(f);</div><div class="line"> }</div><div class="line"></div><div class="line"> <span class="ke yword">template</span> <<span class="keyword">class</span> T></div><div class="line"> <span class="keywordtype">void</span> findQueue(T& connection, std::string& qn) {</div><div class="line"> <span class="keywordflow">if</span> (qn.empty()) {</div><div class="line"> <span class="comment">// Dynamic queue creation</span></div><div class="line"> std::ostringstream os;</div><div class="line"> os << <span class="stringliteral">"_dynamic_"</span> << next_id_++;</div><div class="line"> qn = os.str();</div><div class="line"> }</div><div class="line"> Queue* q = 0;</div><div class="line"> queues::iterator i = queues_.find(qn);</div><div class="line"> <span class="keywordflow">if</span> (i==queues_.end()) {</div><div class="line"> q = <span class="keyword">new</span> Queue(container_, qn);</div><div class="line"> queues_[qn] = q;</div><div class="line"> } <span class="keywordflow">else</span> {</div><div class="line"> q = i->second;</div><div class="line"> }</div><div class="line"> connection.add(make_work(&T::boundQueue, &connection, q, qn));</div><div class="line"> }</div><div class="line"></div><div class="line"> <span class="keywordtype">void</span> findQueueSender(Sender* s, std::string qn) {</div><div class="line"> findQueue(*s, qn);</div><div class="line"> }</div><div class="line"></div><div class="line"> <span class="keywordtype">void</span> findQueueReceiver(Receiver* r, std::string qn) {</div><div class="line"> findQueue(*r, qn);</div><div class="line"> }</div><div class="line">};</div><div class="line"></div><div class="line"><span class="keyword">class </span>connection_handler : <span class="keyword">public</span> <a class="code" href="classproton_1_1messaging__handler.html">proton::messaging_handler</a> {</div><div class="line"> QueueManager&am p; queue_manager_;</div><div class="line"> senders senders_;</div><div class="line"></div><div class="line"><span class="keyword">public</span>:</div><div class="line"> connection_handler(QueueManager& qm) :</div><div class="line"> queue_manager_(qm)</div><div class="line"> {}</div><div class="line"></div><div class="line"> <span class="keywordtype">void</span> on_connection_open(<a name="_a17"></a><a class="code" href="classproton_1_1connection.html">proton::connection</a>& c) OVERRIDE {</div><div class="line"> c.<a name="a18"></a><a class="code" href="classproton_1_1connection.html#a9e8555112049fc2b4945120b3c45f8ab">open</a>(); <span class="comment">// Accept the connection</span></div><div class="line"> }</div><div class="line"></div><div class="line"> <span class="comment">// A sender sends messages from a queue to a subscriber.</span></div><div class="line"> <span class="keywordtype">void</span> on_sender_open(<a class="code " href="classproton_1_1sender.html">proton::sender</a> &sender) OVERRIDE {</div><div class="line"> std::string qn = sender.<a name="a19"></a><a class="code" href="classproton_1_1sender.html#a91a9e8a9445b29d83dd0514cd76503ae">source</a>().<a name="a20"></a><a class="code" href="classproton_1_1terminus.html#adcc4ef24adb8478230018c519aa636ec">dynamic</a>() ? <span class="stringliteral">""</span> : sender.<a class="code" href="classproton_1_1sender.html#a91a9e8a9445b29d83dd0514cd76503ae">source</a>().<a name="a21"></a><a class="code" href="classproton_1_1source.html#a4744eb217c976c199b678bb5a0d55acf">address</a>();</div><div class="line"> Sender* s = <span class="keyword">new</span> Sender(sender, senders_);</div><div class="line"> senders_[sender] = s;</div><div class="line"> queue_manager_.add(make_work(&QueueManager::findQueueSender, &queue_manager_, s, qn));</div><div class="line"> }</div><div class="line"></div><div class="line"> <span class="comment">// A receiver receives messages from a publisher to a queue.</span></div><div class="line"> <span class="keywordtype">void</span> on_receiver_open(<a class="code" href="classproton_1_1receiver.html">proton::receiver</a> &receiver) OVERRIDE {</div><div class="line"> std::string qname = receiver.<a name="a22"></a><a class="code" href="classproton_1_1receiver.html#a5f0b815f331997411a6794f8628592ba">target</a>().<a name="a23"></a><a class="code" href="classproton_1_1target.html#a4744eb217c976c199b678bb5a0d55acf">address</a>();</div><div class="line"> <span class="keywordflow">if</span> (qname == <span class="stringliteral">"shutdown"</span>) {</div><div class="line"> std::cout << <span class="stringliteral">"broker shutting down"</span> << std::endl;</div><div class="line"> <span class="comment">// Sending to the special "shutdown" queue stops the broker.</span></div><div class ="line"> receiver.<a name="a24"></a><a class="code" href="classproton_1_1link.html#aff302bb6016f2ae29f01bb4e07389a52">connection</a>().<a name="a25"></a><a class="code" href="classproton_1_1connection.html#a2f632b80247e868b8264b8a025b84baf">container</a>().<a name="a26"></a><a class="code" href="classproton_1_1container.html#ade42a19caab198f0fd1d28e03cf9ba0e">stop</a>(</div><div class="line"> <a name="_a27"></a><a class="code" href="classproton_1_1error__condition.html">proton::error_condition</a>(<span class="stringliteral">"shutdown"</span>, <span class="stringliteral">"stop broker"</span>));</div><div class="line"> } <span class="keywordflow">else</span> {</div><div class="line"> <span class="keywordflow">if</span> (qname.empty()) {</div><div class="line"> DOUT(std::cerr << <span class="stringliteral">"ODD - trying to attach to a empty address\n"</span>;);</div><div class="line"> }</div><div class="line"> Receiver* r = <span class="keyword">new</span> Receiver(receiver);</div><div class="line"> queue_manager_.add(make_work(&QueueManager::findQueueReceiver, &queue_manager_, r, qname));</div><div class="line"> }</div><div class="line"> }</div><div class="line"></div><div class="line"> <span class="keywordtype">void</span> on_session_close(<a name="_a28"></a><a class="code" href="classproton_1_1session.html">proton::session</a> &session) OVERRIDE {</div><div class="line"> <span class="comment">// Unsubscribe all senders that belong to session.</span></div><div class="line"> <span class="keywordflow">for</span> (proton::sender_iterator i = session.<a name="a29"></a><a class="code" href="classproton_1_1session.html#a2e6d8395032b1d590a9e1586c7de0fb5">senders</a>().begin(); i != session.<a class="code" href="classproton_1_1session.html#a2e6d8395032b1d590a9e1586c7de0fb5">senders</a>().end(); ++i) {</d iv><div class="line"> senders::iterator j = senders_.find(*i);</div><div class="line"> <span class="keywordflow">if</span> (j == senders_.end()) <span class="keywordflow">continue</span>;</div><div class="line"> Sender* s = j->second;</div><div class="line"> <span class="keywordflow">if</span> (s->queue_) {</div><div class="line"> s->queue_->add(make_work(&Queue::unsubscribe, s->queue_, s));</div><div class="line"> }</div><div class="line"> senders_.erase(j);</div><div class="line"> }</div><div class="line"> }</div><div class="line"></div><div class="line"> <span class="keywordtype">void</span> on_error(<span class="keyword">const</span> <a class="code" href="classproton_1_1error__condition.html">proton::error_condition</a>& e) OVERRIDE {</div><div class="line"> std::cerr << <span class="stringliteral">"error: "</span> << e.<a name="a30">< /a><a class="code" href="classproton_1_1error__condition.html#ad2e53bd52dbb1d59081ee7f9f1efc2ac">what</a>() << std::endl;</div><div class="line"> }</div><div class="line"></div><div class="line"> <span class="comment">// The container calls on_transport_close() last.</span></div><div class="line"> <span class="keywordtype">void</span> on_transport_close(<a name="_a31"></a><a class="code" href="classproton_1_1transport.html">proton::transport</a>& t) OVERRIDE {</div><div class="line"> <span class="comment">// Unsubscribe all senders.</span></div><div class="line"> <span class="keywordflow">for</span> (proton::sender_iterator i = t.<a name="a32"></a><a class="code" href="classproton_1_1transport.html#aff302bb6016f2ae29f01bb4e07389a52">connection</a>().<a name="a33"></a><a class="code" href="classproton_1_1connection.html#a2e6d8395032b1d590a9e1586c7de0fb5">senders</a>().begin(); i != t.<a class="code" href="classproton_1_1transport.html#aff302bb6016f2ae 29f01bb4e07389a52">connection</a>().<a class="code" href="classproton_1_1connection.html#a2e6d8395032b1d590a9e1586c7de0fb5">senders</a>().end(); ++i) {</div><div class="line"> senders::iterator j = senders_.find(*i);</div><div class="line"> <span class="keywordflow">if</span> (j == senders_.end()) <span class="keywordflow">continue</span>;</div><div class="line"> Sender* s = j->second;</div><div class="line"> <span class="keywordflow">if</span> (s->queue_) {</div><div class="line"> s->queue_->add(make_work(&Queue::unsubscribe, s->queue_, s));</div><div class="line"> }</div><div class="line"> }</div><div class="line"> <span class="keyword">delete</span> <span class="keyword">this</span>; <span class="comment">// All done.</span></div><div class="line"> }</div><div class="line">};</div><div class="line"></div><div class="line"><span class="keyword">class </span>broker { </div><div class="line"> <span class="keyword">public</span>:</div><div class="line"> broker(<span class="keyword">const</span> std::string addr) :</div><div class="line"> container_(<span class="stringliteral">"broker"</span>), queues_(container_), listener_(queues_)</div><div class="line"> {</div><div class="line"> container_.listen(addr, listener_);</div><div class="line"> }</div><div class="line"></div><div class="line"> <span class="keywordtype">void</span> run() {</div><div class="line"><span class="preprocessor">#if PN_CPP_SUPPORTS_THREADS</span></div><div class="line"> <span class="keywordtype">int</span> threads = hardware_concurrency();</div><div class="line"> std::cout << <span class="stringliteral">"starting "</span> << threads << <span class="stringliteral">" listening threads\n"</span>;</div><div class="line"> std::cout.flush();</div><div class="line"> container_.run( threads);</div><div class="line"><span class="preprocessor">#else</span></div><div class="line"> std::cout << <span class="stringliteral">"no thread support - starting 1 listening thread\n"</span>;</div><div class="line"> std::cout.flush();</div><div class="line"> container_.run();</div><div class="line"><span class="preprocessor">#endif</span></div><div class="line"> }</div><div class="line"></div><div class="line"> <span class="keyword">private</span>:</div><div class="line"> <span class="keyword">struct </span>listener : <span class="keyword">public</span> <a name="_a34"></a><a class="code" href="classproton_1_1listen__handler.html">proton::listen_handler</a> {</div><div class="line"> listener(QueueManager& c) : queues_(c) {}</div><div class="line"></div><div class="line"> <a name="_a35"></a><a class="code" href="classproton_1_1connection__options.html">proton::connection_options</a> on_accept(<a name="_a36"></a><a cla ss="code" href="classproton_1_1listener.html">proton::listener</a>&) OVERRIDE{</div><div class="line"> <span class="keywordflow">return</span> <a class="code" href="classproton_1_1connection__options.html">proton::connection_options</a>().<a name="a37"></a><a class="code" href="classproton_1_1connection__options.html#a7bca23aeb3455378ef2d35975758e504">handler</a>(*(<span class="keyword">new</span> connection_handler(queues_)));</div><div class="line"> }</div><div class="line"></div><div class="line"> <span class="keywordtype">void</span> on_open(<a class="code" href="classproton_1_1listener.html">proton::listener</a>& l) OVERRIDE {</div><div class="line"> std::cout << <span class="stringliteral">"broker listening on "</span> << l.<a name="a38"></a><a class="code" href="classproton_1_1listener.html#a0a3b88007d7cb5fa8d890376c5a8b102">port</a>() << std::endl;</div><div class="line"> }</div><div class="line"> </div><div class="line"> <span class="keywordtype">void</span> on_error(<a class="code" href="classproton_1_1listener.html">proton::listener</a>&, <span class="keyword">const</span> std::string& s) OVERRIDE {</div><div class="line"> std::cerr << <span class="stringliteral">"listen error: "</span> << s << std::endl;</div><div class="line"> <span class="keywordflow">throw</span> std::runtime_error(s);</div><div class="line"> }</div><div class="line"> QueueManager& queues_;</div><div class="line"> };</div><div class="line"></div><div class="line"> <a class="code" href="classproton_1_1container.html">proton::container</a> container_;</div><div class="line"> QueueManager queues_;</div><div class="line"> listener listener_;</div><div class="line">};</div><div class="line"></div><div class="line"><span class="keywordtype">int</span> main(<span class="keywordtype">int</span> argc, <span class="key wordtype">char</span> **argv) {</div><div class="line"> <span class="comment">// Command line options</span></div><div class="line"> std::string address(<span class="stringliteral">"0.0.0.0"</span>);</div><div class="line"> example::options opts(argc, argv);</div><div class="line"></div><div class="line"> opts.add_flag(verbose, <span class="charliteral">'v'</span>, <span class="stringliteral">"verbose"</span>, <span class="stringliteral">"verbose (debugging) output"</span>);</div><div class="line"> opts.add_value(address, <span class="charliteral">'a'</span>, <span class="stringliteral">"address"</span>, <span class="stringliteral">"listen on URL"</span>, <span class="stringliteral">"URL"</span>);</div><div class="line"></div><div class="line"> <span class="keywordflow">try</span> {</div><div class="line"> verbose = <span class="keyword">false</span>;</div><div class="line"> o pts.parse();</div><div class="line"> broker(address).run();</div><div class="line"> <span class="keywordflow">return</span> 0;</div><div class="line"> } <span class="keywordflow">catch</span> (<span class="keyword">const</span> example::bad_option& e) {</div><div class="line"> std::cout << opts << std::endl << e.what() << std::endl;</div><div class="line"> } <span class="keywordflow">catch</span> (<span class="keyword">const</span> std::exception& e) {</div><div class="line"> std::cerr << <span class="stringliteral">"broker shutdown: "</span> << e.what() << std::endl;</div><div class="line"> }</div><div class="line"> <span class="keywordflow">return</span> 1;</div><div class="line">}</div></div><!-- fragment --> </div><!-- contents --> </div><!-- doc-content --> <!-- start footer part --> <div id="nav-path" class="navpath"><!-- id is needed for treeview function! -->
--------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
