| <!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.0 Transitional//EN" "http://www.w3.org/TR/xhtml1/DTD/xhtml1-transitional.dtd"> |
| <html xmlns="http://www.w3.org/1999/xhtml"> |
| <head> |
| <meta http-equiv="Content-Type" content="text/xhtml;charset=UTF-8"/> |
| <meta http-equiv="X-UA-Compatible" content="IE=9"/> |
| <meta name="generator" content="Doxygen 1.8.13"/> |
| <meta name="viewport" content="width=device-width, initial-scale=1"/> |
| <title>Qpid Proton C++ API: broker.cpp</title> |
| <link href="tabs.css" rel="stylesheet" type="text/css"/> |
| <script type="text/javascript" src="jquery.js"></script> |
| <script type="text/javascript" src="dynsections.js"></script> |
| <link href="navtree.css" rel="stylesheet" type="text/css"/> |
| <script type="text/javascript" src="resize.js"></script> |
| <script type="text/javascript" src="navtreedata.js"></script> |
| <script type="text/javascript" src="navtree.js"></script> |
| <script type="text/javascript"> |
| $(document).ready(initResizable); |
| </script> |
| <link href="search/search.css" rel="stylesheet" type="text/css"/> |
| <script type="text/javascript" src="search/searchdata.js"></script> |
| <script type="text/javascript" src="search/search.js"></script> |
| <script type="text/javascript"> |
| $(document).ready(function() { init_search(); }); |
| </script> |
| <link href="doxygen.css" rel="stylesheet" type="text/css" /> |
| </head> |
| <body> |
| <div id="top"><!-- do not remove this div, it is closed by doxygen! --> |
| <div id="titlearea"> |
| <table cellspacing="0" cellpadding="0"> |
| <tbody> |
| <tr style="height: 56px;"> |
| <td id="projectalign" style="padding-left: 0.5em;"> |
| <div id="projectname">Qpid Proton C++ API |
|  <span id="projectnumber">0.25.0</span> |
| </div> |
| </td> |
| <td> <div id="MSearchBox" class="MSearchBoxInactive"> |
| <span class="left"> |
| <img id="MSearchSelect" src="search/mag_sel.png" |
| onmouseover="return searchBox.OnSearchSelectShow()" |
| onmouseout="return searchBox.OnSearchSelectHide()" |
| alt=""/> |
| <input type="text" id="MSearchField" value="Search" accesskey="S" |
| onfocus="searchBox.OnSearchFieldFocus(true)" |
| onblur="searchBox.OnSearchFieldFocus(false)" |
| onkeyup="searchBox.OnSearchFieldChange(event)"/> |
| </span><span class="right"> |
| <a id="MSearchClose" href="javascript:searchBox.CloseResultsWindow()"><img id="MSearchCloseImg" border="0" src="search/close.png" alt=""/></a> |
| </span> |
| </div> |
| </td> |
| </tr> |
| </tbody> |
| </table> |
| </div> |
| <!-- end header part --> |
| <!-- Generated by Doxygen 1.8.13 --> |
| <script type="text/javascript"> |
| var searchBox = new SearchBox("searchBox", "search",false,'Search'); |
| </script> |
| </div><!-- top --> |
| <div id="side-nav" class="ui-resizable side-nav-resizable"> |
| <div id="nav-tree"> |
| <div id="nav-tree-contents"> |
| <div id="nav-sync" class="sync"></div> |
| </div> |
| </div> |
| <div id="splitbar" style="-moz-user-select:none;" |
| class="ui-resizable-handle"> |
| </div> |
| </div> |
| <script type="text/javascript"> |
| $(document).ready(function(){initNavTree('broker_8cpp-example.html','');}); |
| </script> |
| <div id="doc-content"> |
| <!-- window showing the filter options --> |
| <div id="MSearchSelectWindow" |
| onmouseover="return searchBox.OnSearchSelectShow()" |
| onmouseout="return searchBox.OnSearchSelectHide()" |
| onkeydown="return searchBox.OnSearchSelectKey(event)"> |
| </div> |
| |
| <!-- iframe showing the search results (closed by default) --> |
| <div id="MSearchResultsWindow"> |
| <iframe src="javascript:void(0)" frameborder="0" |
| name="MSearchResults" id="MSearchResults"> |
| </iframe> |
| </div> |
| |
| <div class="header"> |
| <div class="headertitle"> |
| <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 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/licenses/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="code" 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="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="preprocessor">#if PN_CPP_HAS_STD_THREAD</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</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">//</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#a86c9509ba3ce1925150c5b7a9a937c94">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#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.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(proton::work f) {</div><div class="line"> <span class="keywordflow">return</span> work_queue_.<a name="a5"></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="_a6"></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="a7"></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"> 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="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() && 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_->first->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="_a8"></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(Sender* 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: "</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="a9"></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(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"> q->add(make_work(&Queue::subscribe, q, <span class="keyword">this</span>));</div><div class="line"> sender_.<a name="a10"></a><a class="code" href="classproton_1_1sender.html#a9e8555112049fc2b4945120b3c45f8ab">open</a>(<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_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="_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">"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.work_queue()), 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="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 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& 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.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="a18"></a><a class="code" href="classproton_1_1sender.html#a91a9e8a9445b29d83dd0514cd76503ae">source</a>().<a name="a19"></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="a20"></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.target().address();</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.connection().container().stop(</div><div class="line"> <a name="_a21"></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="_a22"></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"> 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.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="_a23"></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"> <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_.<a name="a24"></a><a class="code" href="classproton_1_1container.html#a9e138b28e9589583915cf5c5e0e7a524">listen</a>(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_HAS_STD_THREAD</span></div><div class="line"> container_.<a name="a25"></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="_a26"></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="_a27"></a><a class="code" href="classproton_1_1connection__options.html">proton::connection_options</a> on_accept(<a name="_a28"></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__options.html">proton::connection_options</a>().<a name="a29"></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.port() << 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="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="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"> 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 --> |
| </div><!-- doc-content --> |
| <!-- start footer part --> |
| <div id="nav-path" class="navpath"><!-- id is needed for treeview function! --> |
| <ul> |
| <li class="footer">Generated by |
| <a href="http://www.doxygen.org/index.html"> |
| <img class="footer" src="doxygen.png" alt="doxygen"/></a> 1.8.13 </li> |
| </ul> |
| </div> |
| </body> |
| </html> |
