You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by jr...@apache.org on 2018/05/01 13:45:10 UTC

[20/51] [abbrv] [partial] qpid-site git commit: QPID-8154: Refresh the proton doc snapshot

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"> * &quot;License&quot;); 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"> * &quot;AS IS&quot; 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 &quot;options.hpp&quot;</span></div><div class="line"></div><div class="line"><span class="preprocessor">#include &lt;<a class="co
 de" href="connection_8hpp.html">proton/connection.hpp</a>&gt;</span></div><div class="line"><span class="preprocessor">#include &lt;<a class="code" href="connection__options_8hpp.html">proton/connection_options.hpp</a>&gt;</span></div><div class="line"><span class="preprocessor">#include &lt;<a class="code" href="container_8hpp.html">proton/container.hpp</a>&gt;</span></div><div class="line"><span class="preprocessor">#include &lt;<a class="code" href="delivery_8hpp.html">proton/delivery.hpp</a>&gt;</span></div><div class="line"><span class="preprocessor">#include &lt;<a class="code" href="error__condition_8hpp.html">proton/error_condition.hpp</a>&gt;</span></div><div class="line"><span class="preprocessor">#include &lt;<a class="code" href="function_8hpp.html">proton/function.hpp</a>&gt;</span></div><div class="line"><span class="preprocessor">#include &lt;<a class="code" href="listen__handler_8hpp.html">proton/listen_handler.hpp</a>&gt;</span></div><div class="line"><span class="p
 reprocessor">#include &lt;<a class="code" href="listener_8hpp.html">proton/listener.hpp</a>&gt;</span></div><div class="line"><span class="preprocessor">#include &lt;<a class="code" href="message_8hpp.html">proton/message.hpp</a>&gt;</span></div><div class="line"><span class="preprocessor">#include &lt;<a class="code" href="messaging__handler_8hpp.html">proton/messaging_handler.hpp</a>&gt;</span></div><div class="line"><span class="preprocessor">#include &lt;<a class="code" href="receiver__options_8hpp.html">proton/receiver_options.hpp</a>&gt;</span></div><div class="line"><span class="preprocessor">#include &lt;<a class="code" href="sender__options_8hpp.html">proton/sender_options.hpp</a>&gt;</span></div><div class="line"><span class="preprocessor">#include &lt;<a class="code" href="source__options_8hpp.html">proton/source_options.hpp</a>&gt;</span></div><div class="line"><span class="preprocessor">#include &lt;<a class="code" href="target_8hpp.html">proton/target.hpp</a>&gt;</span
 ></div><div class="line"><span class="preprocessor">#include &lt;<a class="code" href="target__options_8hpp.html">proton/target_options.hpp</a>&gt;</span></div><div class="line"><span class="preprocessor">#include &lt;<a class="code" href="tracker_8hpp.html">proton/tracker.hpp</a>&gt;</span></div><div class="line"><span class="preprocessor">#include &lt;<a class="code" href="transport_8hpp.html">proton/transport.hpp</a>&gt;</span></div><div class="line"><span class="preprocessor">#include &lt;<a class="code" href="work__queue_8hpp.html">proton/work_queue.hpp</a>&gt;</span></div><div class="line"></div><div class="line"><span class="preprocessor">#include &lt;deque&gt;</span></div><div class="line"><span class="preprocessor">#include &lt;iostream&gt;</span></div><div class="line"><span class="preprocessor">#include &lt;map&gt;</span></div><div class="line"><span class="preprocessor">#include &lt;string&gt;</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 &lt;thread&gt;</span></div><div class="line"><span class="preprocessor">#endif</span></div><div class="line"></div><div class="line"><span class="preprocessor">#include &quot;fake_cpp11.hpp&quot;</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&#39;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&lt;proton::sender, Sender*&gt; 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&amp; senders_;</div><div class="line">    <a name="_a2"></a><a class="code" href="classproton_1_1work__queue.html">proton::work_queue</a>&amp; 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> &amp;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> &amp;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&amp; 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 &lt;&lt; <span class="stringliteral">&quot;Sender:   &quot;</span> &lt;&lt; <span class="keyword">this</span> &lt;&lt; <span class="stringliteral">&quot; sending\n&quot;</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 &lt;&lt; <span class="stringliteral">&quot;Sender:   &quot;</span> &lt;&lt; <span class="keyword">this</span> &lt;&lt; <span class="stringliteral">&quot; deleting\n&quot;</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&lt;proton::message&gt; messages_;</div><div class="line">    <span class="keyword">typedef</span> std::map&lt;Sender*, int&gt; 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 &lt;&lt; <span class="stringl
 iteral">&quot;Queue:    &quot;</span> &lt;&lt; <span class="keyword">this</span> &lt;&lt; <span class="stringliteral">&quot; tryToSend: &quot;</span> &lt;&lt; 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() &amp;&amp; outOfCredit&lt;subscriptions_.size()) {</div><div class="line">            <span class="comment">// If we got the end (or haven&#39;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 &lt;&lt; <span class="stringliteral">&quot;(&quot;</span> &lt;&lt; current_-&gt;second &lt;&lt; <span class="stringliteral">&quot;) &quot;</span>;);</div><div class="line">            <span class="keywordflow">if</span> (current_-&gt;second&gt;0) {</div><div class="line">                DOUT(std::cerr &lt;&lt; current_-&gt;first &lt;&lt; <span class="stringliteral">&quot; &quot;</span>;);</div><div class="line">                <a name="a9"></a><a class="code" href="namespaceproton.html#ae77a8638d584f5fc99a1bbaf218d2481">proton::schedule_work</a>(current_-&gt;first, &amp;Sender::sendMsg, current_-&gt;first, messages_.front());</div><div class="line">                messages_.pop_front();</div><div class="
 line">                --current_-&gt;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 &lt;&lt; <span class="stringliteral">&quot;\n&quot;</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>&amp; c, <span class="keyword">const</span> std::string&amp; 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 &lt;&lt; <span class="stringliteral">&quot;Queue:    &quot;</span> &lt;&lt; <span class="keyword">this</span> &lt;&lt; <span class="stringliteral">&quot;(&quot;</span> &lt;&lt; name_ &lt;&lt; <span class="stringliteral">&quot;) queueMsg\n&quot;</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 &lt;&lt; <span cla
 ss="stringliteral">&quot;Queue:    &quot;</span> &lt;&lt; <span class="keyword">this</span> &lt;&lt; <span class="stringliteral">&quot;(&quot;</span> &lt;&lt; name_ &lt;&lt; <span class="stringliteral">&quot;) flow: &quot;</span> &lt;&lt; c &lt;&lt; <span class="stringliteral">&quot; to &quot;</span> &lt;&lt; s &lt;&lt; <span class="stringliteral">&quot;\n&quot;</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 &lt;&lt; <span class="stringliteral">&quot;Queue:    &quot;</span> &lt;&lt; <span class="keyword">this</span> &lt;&lt; <span class="stringliteral">&quot;(&quot;</span> &lt;&lt; name_ &lt;&lt; <span class="stringliteral">&quot;) subscribe Sender: &quot;</span> &lt;&lt; s &lt;&lt; <span class="stringliteral">&quot;\n&quot;</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 &lt;&lt; <span class="stringliteral">&quot;Queue:    &quot;</span> &lt;&lt; <span class="keyword">this</span> &lt;&lt; <span class="stringliteral">&quot;(&quot;</span> &lt;&lt; name_ &lt;&lt; <span class="stringliteral">&quot;) unsubscribe Sender: &quot;</span> &lt;&lt; s &lt;&lt; <span class="stringliteral">&quot;\n&quot;</span>;);</div><div class="line">        <span class="comment">// If we&#39;re about to erase the current subscription move on</span></div><div class="line">        <span class="keywordflow">if</span> (current_ != subscriptions_.end() &amp;&amp; current_-&gt;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, &amp;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> &amp;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_, &amp;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> &amp;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_, &amp;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 &lt;&lt; <span class="stringliteral">&quot;Sender:   &quot;</span> &lt;&lt; <span class="keyword">this</span> &lt;&lt; <span class="stringliteral">&quot; bound to Queue: &quot;</span> &lt;&lt; q &lt;&lt;<span class="stringliteral">&quot;(&quot;</span> &lt;&lt; qn &lt;&lt; <span class="stringliteral">&quot;)\n&quot;</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, &amp;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_&gt;0) {</div><div class="line">        <a class="code" href="namespaceproton.html#ae77a8638d584f5fc99a1bbaf218d2481">proton::schedule_work</a>(queue_, &amp;Queue::flow, queue_, <span class="keyword">this</span>, pending_credit_);</div><div class="line">    }</div><div class="line">    std::cout &lt;&lt; <span class="stringliteral">&quot;sending from &quot;</span> &lt;&lt; queue_name_ &lt;&lt; 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>&amp; work_queue_;</div><div class="line">    Queue* queue_;</div><div class="line">    std::deque&lt;proton::message&gt; 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> &amp;, <a class="code" href="classproton_1_1message.html">proton::message</a> &amp;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 &lt;&lt; <span class="stringliteral">&quot;Receiver: &quot;</span> &lt;&lt; <span class="keyword">this</span> &lt;&lt; <span class="stringliteral">&quot; queueing &quot;</span> &lt;&lt; messages_.size() &lt;&lt; <span class="stringliteral">&quot; msgs to: &quot;</span> &lt;&lt; queue_ &lt;&lt; <span class="stringliteral">&quot;\n&quot;</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_, &amp;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 &lt;&lt; <span class="stringliteral">&quot;Receiver: &quot;</span> &lt;&lt; <span class="keyword">this</span> &lt;&lt; <span class="stringliteral">&quot; bound to Queue: &quot;</span> &lt;&lt; q &lt;&lt; <span class="stringliteral">&quot;(&quot;</span> &lt;&lt; qn &lt;&lt; <span class="stringliteral">&quot;)\n&quot;</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 &lt;&lt; <span class="stringliteral">&quot;receiving to &quot;</span> &lt;&lt; q
 n &lt;&lt; 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>&amp; 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&lt;std::string, Queue*&gt; 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>&amp; 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> &lt;<span class="keyword">class</span> T&gt;</div><div class="line">    <span class="keywordtype">void</span> findQueue(T&amp; connection, std::string&amp; 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 &lt;&lt; <span c
 lass="stringliteral">&quot;_dynamic_&quot;</span> &lt;&lt; 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-&gt;second;</div><div class="line">        }</div><div class="line">        <a class="code" href="namespaceproton.html#ae77a8638d584f5fc99a1bbaf218d2481">proton::schedule_work</a>(&amp;connection, &amp;T::boundQueue, &amp;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&amp; 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&amp; 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>&amp; 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> &amp;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">&quot;&quot;</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>(&amp;queue_manager_, &amp;QueueManager::findQueueSender, &amp;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> &amp;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">&quot;shutdown&quot;</span>) {</div><div class="line">            std::cout &lt;&lt; <span class="stringliteral">&quot;broker shutting down&quot;</span> &lt;&lt; std::endl;</div><div class="line">            <span class="comment">// Sending to the special &quot;shutdown&quot; 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">&quot;shutdown&quot;</span>, <span class="stringliteral">&quot;stop broker&quot;</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 &lt;&lt; <span class="stringliteral">&quot;ODD - trying to att
 ach to a empty address\n&quot;</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>(&amp;queue_manager_, &amp;QueueManager::findQueueReceiver, &amp;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> &amp;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-&gt;second;</div><div class="line">            <span class="keywordflow">if</span> (s-&gt;queue_) {</div><div class="line">                <a class="code" href="namespaceproton.html#ae77a8638d584f5fc99a1bbaf218d2481">proton::schedule_work</a>(s-&gt;queue_, &amp;Queue::unsubscribe, s-&gt;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>&amp; e) OVERRIDE {</div><div class="line">        std::cerr &lt;&lt; <span class="stringlitera
 l">&quot;error: &quot;</span> &lt;&lt; e.what() &lt;&lt; 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>&amp; 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-&gt;second;</div><div class="line">            <s
 pan class="keywordflow">if</span> (s-&gt;queue_) {</div><div class="line">                <a class="code" href="namespaceproton.html#ae77a8638d584f5fc99a1bbaf218d2481">proton::schedule_work</a>(s-&gt;queue_, &amp;Queue::unsubscribe, s-&gt;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">&quot;broker&quot;</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 &lt;&lt; <span class="stringliteral">&quot;broker listening on &quot;</span> &lt;&lt; addr &lt;&lt; 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 &lt;&lt; <span class="stringliteral">&quot;starting &quot;</span> &lt;&lt; std::thread::hardware_concurrency() &lt;&lt; <span class="stringliteral">&quot; listening threads\n&quot;</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&amp; 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>&amp;) 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>&amp;, <span class="keyword">const</span> std::string&amp; s) OVERRIDE {</div><div class="line">            std::cerr &lt;&lt; <span class="stringliteral">&quot;listen error: &quot;</span> &lt;&lt; s &lt;&lt; std::endl;</div><div class="line">            <span class="keywordflow">throw</span> std::runtime_error(s);</div><div class="line">        }</div><div class="line">        QueueManager&amp; 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">&quot;0.0.0.0&quot;</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">&#39;v&#39;</span>, <span class="stringliteral">&quot;verbose&quot;</span>, <span class="stringliteral">&quot;verbose (debugging) output&quot;</span>);</div><div class="line">    opts.add_value(address, <span class="charliteral">&#39;a&#39;</span>, <span class="stringliteral">&quot;address&quot;</span>, <span class="stringl
 iteral">&quot;listen on URL&quot;</span>, <span class="stringliteral">&quot;URL&quot;</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&amp; e) {</div><div class="line">        std::cout &lt;&lt; opts &lt;&lt; std::endl &lt;&lt; e.what() &lt;&lt; std::endl;</div><div class="line">    } <span class="keywordflow">catch</span> (<span class="keyword">const</span> std::exception&amp; e) {</div><div class="line">        std::cerr &lt;&lt; <span class="stringliteral">&quot;broker shutdown: &quot;</span> &lt;&lt; e.what() &lt;&lt; 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"> * &quot;License&quot;); 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"> * &quot;AS IS&quot; 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 &quot;options.hpp&quot;</span></div><div class="line"></div><div class="line"><span class="preprocessor">#include &lt;<a class="co
 de" href="connection_8hpp.html">proton/connection.hpp</a>&gt;</span></div><div class="line"><span class="preprocessor">#include &lt;<a class="code" href="connection__options_8hpp.html">proton/connection_options.hpp</a>&gt;</span></div><div class="line"><span class="preprocessor">#include &lt;<a class="code" href="container_8hpp.html">proton/container.hpp</a>&gt;</span></div><div class="line"><span class="preprocessor">#include &lt;<a class="code" href="delivery_8hpp.html">proton/delivery.hpp</a>&gt;</span></div><div class="line"><span class="preprocessor">#include &lt;<a class="code" href="error__condition_8hpp.html">proton/error_condition.hpp</a>&gt;</span></div><div class="line"><span class="preprocessor">#include &lt;<a class="code" href="listen__handler_8hpp.html">proton/listen_handler.hpp</a>&gt;</span></div><div class="line"><span class="preprocessor">#include &lt;<a class="code" href="listener_8hpp.html">proton/listener.hpp</a>&gt;</span></div><div class="line"><span class="p
 reprocessor">#include &lt;<a class="code" href="message_8hpp.html">proton/message.hpp</a>&gt;</span></div><div class="line"><span class="preprocessor">#include &lt;<a class="code" href="messaging__handler_8hpp.html">proton/messaging_handler.hpp</a>&gt;</span></div><div class="line"><span class="preprocessor">#include &lt;<a class="code" href="receiver__options_8hpp.html">proton/receiver_options.hpp</a>&gt;</span></div><div class="line"><span class="preprocessor">#include &lt;<a class="code" href="sender__options_8hpp.html">proton/sender_options.hpp</a>&gt;</span></div><div class="line"><span class="preprocessor">#include &lt;<a class="code" href="source__options_8hpp.html">proton/source_options.hpp</a>&gt;</span></div><div class="line"><span class="preprocessor">#include &lt;<a class="code" href="target_8hpp.html">proton/target.hpp</a>&gt;</span></div><div class="line"><span class="preprocessor">#include &lt;<a class="code" href="target__options_8hpp.html">proton/target_options.hpp<
 /a>&gt;</span></div><div class="line"><span class="preprocessor">#include &lt;<a class="code" href="tracker_8hpp.html">proton/tracker.hpp</a>&gt;</span></div><div class="line"><span class="preprocessor">#include &lt;<a class="code" href="transport_8hpp.html">proton/transport.hpp</a>&gt;</span></div><div class="line"><span class="preprocessor">#include &lt;<a class="code" href="work__queue_8hpp.html">proton/work_queue.hpp</a>&gt;</span></div><div class="line"></div><div class="line"><span class="preprocessor">#include &lt;deque&gt;</span></div><div class="line"><span class="preprocessor">#include &lt;iostream&gt;</span></div><div class="line"><span class="preprocessor">#include &lt;map&gt;</span></div><div class="line"><span class="preprocessor">#include &lt;string&gt;</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 &lt;thread&gt;</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 &quot;fake_cpp11.hpp&quot;</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&#39;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&lt;proton::sender, Sender*&gt; 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&amp; senders_;</div><div class="line">    <a name="_a2"></a><a class="code" href="classproton_1_1work__queue.html">proton::work_queue</a>&amp; 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> &amp;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> &amp;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&amp; 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 &lt;&lt; <span class="stringliteral">&quot;Sender:   &quot;</span> &lt;&lt; <span class="keyword">this</span> &lt;&lt; <span class="stringliteral">&quot; sending\n&quot;</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 &lt;&lt; <span class="stringliteral">&quot;Sender:   &quot;</span> &lt;&lt; <span class="keyword">this</span> &lt;&lt; <span class="stringliteral">&quot; deleting\n&quot;</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&lt;proton::message&gt; messages_;</div><div class="line">    <span class="keyword">typedef</span> std::map&lt;Sender*, int&gt; 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 &lt;&lt; <span class="stringliteral">&quot;Queue:    &quot;</span> &lt;&lt; <span class="keyword">this</span> &lt;&lt; <span class="stringliteral">&quot; tryToSend: &quot;</span> &lt;&lt; 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;&amp; outOfCredit&lt;subscriptions_.size()) {</div><div class="line">            <span class="comment">// If we got the end (or haven&#39;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 &lt;&lt; <span class="stringliteral">&quot;(&quot;</span> &lt;&lt; current_-&gt;second &lt;&lt; <span class="stringliteral">&quot;) &quot;</span>;);</div><div class="line">            <span class="keywordflow">if</span> (current_-&gt;second&gt;0) {</div><div class="line">                DOUT(std::cerr &lt;&lt; current_-&gt;first &lt;&lt; <span class="stringliteral">&quot; &quot;</span>;);</div><div class="line">                current_-&gt;f
 irst-&gt;add(make_work(&amp;Sender::sendMsg, current_-&gt;first, messages_.front()));</div><div class="line">                messages_.pop_front();</div><div class="line">                --current_-&gt;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 &lt;&lt; <span class="stringliteral">&quot;\n&quot;</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>&amp; c, <span class="keyword">const</span> std::string&amp; 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 &lt;&lt; <span class="stringliteral">&quot;Queue:    &quot;</span> &lt;&lt; <span class="keyword">this</span> &lt;&lt; <span class="stringliteral">&quot;(&quot;</span> &lt;&lt; name_ &lt;&lt; <span class="stringliteral">&quot;) queueMsg\n&quot;</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 &lt;&lt; <span class="stringliteral">&quot;Queue:    &quot;</span> &lt;&lt; <span class="keyword">this</span> &lt;&lt; <span class="stringliteral">&quot;(&quot;</span> &lt;&lt; name_ &lt;&lt; <span class="stringliteral">&quot;) flow: &quot;</span> &lt;&lt; c &lt;&lt; <span class="stringliteral">&quot; to &quot;</span> &lt;&lt; s &lt;&lt; <span class="stringliteral">&quot;\n&quot;</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 &lt;&lt; <span class="stringliteral">&quot;Queue:    &quot;</span> &lt;&lt; <span class="keyword">this</span> &lt;&lt; <span class="stringliteral">&quot;(&quot;</span> &lt;&lt; name_ &lt;&lt; <span class="stringliteral">&quot;) subscribe Sender: &qu
 ot;</span> &lt;&lt; s &lt;&lt; <span class="stringliteral">&quot;\n&quot;</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 &lt;&lt; <span class="stringliteral">&quot;Queue:    &quot;</span> &lt;&lt; <span class="keyword">this</span> &lt;&lt; <span class="stringliteral">&quot;(&quot;</span> &lt;&lt; name_ &lt;&lt; <span class="stringliteral">&quot;) unsubscribe Sender: &quot;</span> &lt;&lt; s &lt;&lt; <span class="stringliteral">&quot;\n&quot;</span>;);</div><div class="line">        <span class="comment">// If we&#39;re about to erase the current subscription move on</span></div><div class="line">        <span class="keywordflow">if</span> (current_ != subscriptions_.end() &amp;&amp; current_-&gt;first==s) ++current_;</div><div class="line">        subscriptions_.erase(s);</div><div class="line">        s
 -&gt;add(make_work(&amp;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> &amp;sender) {</div><div class="line">    <span class="keywordflow">if</span> (queue_) {</div><div class="line">        queue_-&gt;add(make_work(&amp;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> &amp;sender) {</div><div class="line">    <span class="keywordflow">if</span> (queue_) {</div><div class="line">        queue_-&gt;add(make_work(&amp;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 &lt;&lt; <span class="stringliteral">&quot;Sender:   &quot;</span> &lt;&lt; <span class="keyword">this</span> &lt;&lt; <span class="stringliteral">&quot; bound to Queue: &quot;</span> &lt;&lt; q &lt;&lt;<span class="stringliteral">&quot;(&quot;</span> &lt;&lt; qn &lt;&lt; <span class="stringliteral">&quot;)\n&quot;</span>;);</div><div class="line">    queue_ = q;</div><div class="line">    queue_name_ = qn;</div><div class="line"></div><div class="line">    q-&gt;add(make_work(&amp;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_&gt;0) {</div><div class="line">        queue_-&gt;add(make_work(&amp;Queue::flow, queue_, <span class="keyword">this</span>, pending_credit_));</div><div class="line">    }</div><div class="line">    std::cout &lt;&lt; <span class="stringliteral">&quot;sending from &quot;</span> &lt;&lt; queue_name_ &lt;&lt; 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>&amp; work_queue_;</div><div class="line">    Queue* queue_;</div><div class="line">    std::deque&lt;proton::message&gt; 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> &amp;, <a class="code" href="classproton_1_1message.html">proton::message</a> &amp;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 &lt;&lt; <span class="stringliteral">&
 quot;Receiver: &quot;</span> &lt;&lt; <span class="keyword">this</span> &lt;&lt; <span class="stringliteral">&quot; queueing &quot;</span> &lt;&lt; messages_.size() &lt;&lt; <span class="stringliteral">&quot; msgs to: &quot;</span> &lt;&lt; queue_ &lt;&lt; <span class="stringliteral">&quot;\n&quot;</span>;);</div><div class="line">        <span class="keywordflow">while</span> (!messages_.empty()) {</div><div class="line">            queue_-&gt;add(make_work(&amp;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 &lt;&lt; <span class="stringliteral">&quot;Receiver: &quot;</span> &lt;&lt; <span class="keyword">this</span> &lt;&lt; <span class="stringliteral">&quot; bound to Queue: &quot;</span> &lt;&lt; q &lt;&lt; <span class="stringliteral">&quot;(&quot;</span> &lt;&lt; qn &lt;&lt; <span class="stringliteral">&quot;)\n&quot;</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 &lt;&lt; <span class="stringliteral">&quot;receiving to &quot;</span> &lt;&lt; qn &lt;&lt; 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>&amp; 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&lt;std::string, Queue*&gt; 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>&amp; 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> &lt;<span class="keyword">class</span> T&gt;</div><div class="line">    <span class="keywordtype">void</span> findQueue(T&amp; connection, std::string&amp; 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 &lt;&lt; <span class="stringliteral">&quot;_dynamic_&quot;</span> &lt;&lt; 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-&gt;second;</div><div class="line">        }</div><div class="line">        connection.add(make_work(&amp;T::boundQueue, &amp;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&amp; 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>&amp; 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> &amp;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">&quot;&quot;</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(&amp;QueueManager::findQueueSender, &amp;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> &amp;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">&quot;shutdown&quot;</span>) {</div><div class="line">            std::cout &lt;&lt; <span class="stringliteral">&quot;broker shutting down&quot;</span> &lt;&lt; std::endl;</div><div class="line">            <span class="comment">// Sending to the special &quot;shutdown&quot; 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">&quot;shutdown&quot;</span>, <span class="stringliteral">&quot;stop broker&quot;</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 &lt;&lt; <span class="stringliteral">&quot;ODD - trying to attach to a empty address\n&quot;</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(&amp;QueueManager::findQueueReceiver, &amp;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> &amp;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-&gt;second;</div><div class="line">            <span class="keywordflow">if</span> (s-&gt;queue_) {</div><div class="line">                s-&gt;queue_-&gt;add(make_work(&amp;Queue::unsubscribe, s-&gt;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>&amp; e) OVERRIDE {</div><div class="line">        std::cerr &lt;&lt; <span class="stringliteral">&quot;error: &quot;</span> &lt;&lt; e.<a name="a30"><
 /a><a class="code" href="classproton_1_1error__condition.html#ad2e53bd52dbb1d59081ee7f9f1efc2ac">what</a>() &lt;&lt; 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>&amp; 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-&gt;second;</div><div class="line">            <span class="keywordflow">if</span> (s-&gt;queue_) {</div><div class="line">                s-&gt;queue_-&gt;add(make_work(&amp;Queue::unsubscribe, s-&gt;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">&quot;broker&quot;</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 &lt;&lt; <span class="stringliteral">&quot;starting &quot;</span> &lt;&lt; threads &lt;&lt; <span class="stringliteral">&quot; listening threads\n&quot;</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 &lt;&lt; <span class="stringliteral">&quot;no thread support - starting 1 listening thread\n&quot;</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&amp; 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>&amp;) 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>&amp; l) OVERRIDE {</div><div class="line">            std::cout &lt;&lt; <span class="stringliteral">&quot;broker listening on &quot;</span> &lt;&lt; l.<a name="a38"></a><a class="code" href="classproton_1_1listener.html#a0a3b88007d7cb5fa8d890376c5a8b102">port</a>() &lt;&lt; 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>&amp;, <span class="keyword">const</span> std::string&amp; s) OVERRIDE {</div><div class="line">            std::cerr &lt;&lt; <span class="stringliteral">&quot;listen error: &quot;</span> &lt;&lt; s &lt;&lt; std::endl;</div><div class="line">            <span class="keywordflow">throw</span> std::runtime_error(s);</div><div class="line">        }</div><div class="line">        QueueManager&amp; 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">&quot;0.0.0.0&quot;</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">&#39;v&#39;</span>, <span class="stringliteral">&quot;verbose&quot;</span>, <span class="stringliteral">&quot;verbose (debugging) output&quot;</span>);</div><div class="line">    opts.add_value(address, <span class="charliteral">&#39;a&#39;</span>, <span class="stringliteral">&quot;address&quot;</span>, <span class="stringliteral">&quot;listen on URL&quot;</span>, <span class="stringliteral">&quot;URL&quot;</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&amp; e) {</div><div class="line">        std::cout &lt;&lt; opts &lt;&lt; std::endl &lt;&lt; e.what() &lt;&lt; std::endl;</div><div class="line">    } <span class="keywordflow">catch</span> (<span class="keyword">const</span> std::exception&amp; e) {</div><div class="line">        std::cerr &lt;&lt; <span class="stringliteral">&quot;broker shutdown: &quot;</span> &lt;&lt; e.what() &lt;&lt; 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: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org