You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ac...@apache.org on 2016/05/16 14:43:30 UTC
[1/4] qpid-proton git commit: PROTON-1184: C++ merge APIs for single
and multi-threaded use.
Repository: qpid-proton
Updated Branches:
refs/heads/master b1c348889 -> 1b8450d6a
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/src/receiver_options.cpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/src/receiver_options.cpp b/proton-c/bindings/cpp/src/receiver_options.cpp
index 1a1b3f3..26a78a0 100644
--- a/proton-c/bindings/cpp/src/receiver_options.cpp
+++ b/proton-c/bindings/cpp/src/receiver_options.cpp
@@ -45,7 +45,9 @@ template <class T> struct option {
class receiver_options::impl {
static void set_handler(receiver l, proton_handler &h) {
pn_record_t *record = pn_link_attachments(unwrap(l));
- internal::pn_ptr<pn_handler_t> chandler = l.connection().container().impl_->cpp_handler(&h);
+ // FIXME aconway 2016-05-04: container_impl specific, fix for engine.
+ internal::pn_ptr<pn_handler_t> chandler =
+ static_cast<container_impl&>(l.connection().container()).cpp_handler(&h);
pn_record_set_handler(record, chandler.get());
}
@@ -123,7 +125,7 @@ receiver_options& receiver_options::operator=(const receiver_options& x) {
void receiver_options::update(const receiver_options& x) { impl_->update(*x.impl_); }
-receiver_options& receiver_options::handler(class handler *h) { impl_->handler = h->messaging_adapter_.get(); return *this; }
+receiver_options& receiver_options::handler(class handler &h) { impl_->handler = h.messaging_adapter_.get(); return *this; }
receiver_options& receiver_options::delivery_mode(proton::delivery_mode m) {impl_->delivery_mode = m; return *this; }
receiver_options& receiver_options::auto_accept(bool b) {impl_->auto_accept = b; return *this; }
receiver_options& receiver_options::auto_settle(bool b) {impl_->auto_settle = b; return *this; }
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/src/sender_options.cpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/src/sender_options.cpp b/proton-c/bindings/cpp/src/sender_options.cpp
index 4786937..2922157 100644
--- a/proton-c/bindings/cpp/src/sender_options.cpp
+++ b/proton-c/bindings/cpp/src/sender_options.cpp
@@ -43,7 +43,9 @@ template <class T> struct option {
class sender_options::impl {
static void set_handler(sender l, proton_handler &h) {
pn_record_t *record = pn_link_attachments(unwrap(l));
- internal::pn_ptr<pn_handler_t> chandler = l.connection().container().impl_->cpp_handler(&h);
+ // FIXME aconway 2016-05-04: container_impl specific, fix for engine.
+ internal::pn_ptr<pn_handler_t> chandler =
+ static_cast<container_impl&>(l.connection().container()).cpp_handler(&h);
pn_record_set_handler(record, chandler.get());
}
@@ -111,7 +113,7 @@ sender_options& sender_options::operator=(const sender_options& x) {
void sender_options::update(const sender_options& x) { impl_->update(*x.impl_); }
-sender_options& sender_options::handler(class handler *h) { impl_->handler = h->messaging_adapter_.get(); return *this; }
+sender_options& sender_options::handler(class handler &h) { impl_->handler = h.messaging_adapter_.get(); return *this; }
sender_options& sender_options::delivery_mode(proton::delivery_mode m) {impl_->delivery_mode = m; return *this; }
sender_options& sender_options::auto_settle(bool b) {impl_->auto_settle = b; return *this; }
sender_options& sender_options::source(source_options &s) {impl_->source = s; return *this; }
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/src/session.cpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/src/session.cpp b/proton-c/bindings/cpp/src/session.cpp
index cdb4722..bfcedc2 100644
--- a/proton-c/bindings/cpp/src/session.cpp
+++ b/proton-c/bindings/cpp/src/session.cpp
@@ -58,7 +58,9 @@ connection session::connection() const {
namespace {
std::string next_link_name(const connection& c) {
- return connection_context::get(c).link_gen.next();
+ io::link_namer* ln = connection_context::get(c).link_gen;
+
+ return ln ? ln->link_name() : uuid::random().str();
}
}
@@ -119,7 +121,6 @@ receiver_range session::receivers() const {
return receiver_range(receiver_iterator(make_wrapper<receiver>(lnk), pn_object()));
}
-
session_iterator session_iterator::operator++() {
obj_ = pn_session_next(unwrap(obj_), 0);
return *this;
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/src/session_options.cpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/src/session_options.cpp b/proton-c/bindings/cpp/src/session_options.cpp
index 9f77d6b..71b5b8d 100644
--- a/proton-c/bindings/cpp/src/session_options.cpp
+++ b/proton-c/bindings/cpp/src/session_options.cpp
@@ -49,7 +49,9 @@ class session_options::impl {
if (s.uninitialized()) {
if (handler.set) {
pn_record_t *record = pn_session_attachments(unwrap(s));
- internal::pn_ptr<pn_handler_t> chandler = s.connection().container().impl_->cpp_handler(handler.value);
+ // FIXME aconway 2016-05-04: container_impl specific
+ internal::pn_ptr<pn_handler_t> chandler =
+ static_cast<container_impl&>(s.connection().container()).cpp_handler(handler.value);
pn_record_set_handler(record, chandler.get());
}
}
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/src/test_dummy_container.hpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/src/test_dummy_container.hpp b/proton-c/bindings/cpp/src/test_dummy_container.hpp
new file mode 100644
index 0000000..7307901
--- /dev/null
+++ b/proton-c/bindings/cpp/src/test_dummy_container.hpp
@@ -0,0 +1,73 @@
+#ifndef TEST_DUMMY_CONTAINER_HPP
+#define TEST_DUMMY_CONTAINER_HPP
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "proton/container.hpp"
+#include "proton/event_loop.hpp"
+#include "proton/thread_safe.hpp"
+
+namespace test {
+
+using namespace proton;
+
+
+class dummy_container : public container {
+ public:
+ dummy_container(const std::string cid="") :
+ id_(cid), fail("not implemented for dummy_container") {}
+
+ returned<connection> connect(const std::string&, const connection_options&) { throw fail; }
+ listener listen(const std::string& , listen_handler& ) { throw fail; }
+ listener listen(const std::string&, const connection_options&) { throw fail; }
+ void stop_listening(const std::string&) { throw fail; }
+ void run() { throw fail; }
+ void auto_stop(bool) { throw fail; }
+ void stop(const proton::error_condition& ) { throw fail; }
+ returned<sender> open_sender(const std::string &, const proton::sender_options &, const connection_options&) { throw fail; }
+ returned<receiver> open_receiver( const std::string &, const proton::receiver_options &, const connection_options &) { throw fail; }
+ std::string id() const { return id_; }
+ void client_connection_options(const connection_options &o) { ccopts_ = o; }
+ connection_options client_connection_options() const { return ccopts_; }
+ void server_connection_options(const connection_options &o) { scopts_ = o; }
+ connection_options server_connection_options() const { return scopts_; }
+ void sender_options(const class sender_options &o) { sopts_ = o; }
+ class sender_options sender_options() const { return sopts_; }
+ void receiver_options(const class receiver_options &o) { ropts_ = o; }
+ class receiver_options receiver_options() const { return ropts_; }
+
+ private:
+ std::string id_;
+ connection_options ccopts_, scopts_;
+ class sender_options sopts_;
+ class receiver_options ropts_;
+ std::runtime_error fail;
+};
+
+class dummy_event_loop : public event_loop {
+#if PN_CPP_HAS_CPP11
+ bool inject(std::function<void()> f) PN_CPP_OVERRIDE { f(); return true; }
+#endif
+ bool inject(proton::inject_handler& h) PN_CPP_OVERRIDE { h.on_inject(); return true; }
+};
+
+}
+
+#endif // TEST_DUMMY_CONTAINER_HPP
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/src/thread_safe_test.cpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/src/thread_safe_test.cpp b/proton-c/bindings/cpp/src/thread_safe_test.cpp
new file mode 100644
index 0000000..540fcc8
--- /dev/null
+++ b/proton-c/bindings/cpp/src/thread_safe_test.cpp
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+/// Test reference counting for object wrappers, threads_safe<> wrappers and returned<> wrappers.
+///
+
+#include "test_bits.hpp"
+#include "test_dummy_container.hpp"
+#include "proton_bits.hpp"
+
+#include "proton/thread_safe.hpp"
+#include "proton/io/connection_engine.hpp"
+#include "proton/io/link_namer.hpp"
+
+#include <proton/connection.h>
+
+namespace {
+
+using namespace proton;
+using namespace test;
+using namespace std;
+
+dummy_container cont;
+
+namespace {
+struct linknames : io::link_namer {
+ std::string link_name() { return "X"; }
+} dummy_link_namer;
+}
+
+void test_new() {
+ pn_connection_t* c = 0;
+ thread_safe<connection>* p = 0;
+ {
+ io::connection_engine e(cont, dummy_link_namer, new dummy_event_loop);
+ c = unwrap(e.connection());
+ int r = pn_refcount(c);
+ ASSERT(r >= 1); // engine may have internal refs (transport, collector).
+ p = make_thread_safe(e.connection()).release();
+ ASSERT_EQUAL(r+1, pn_refcount(c));
+ delete p;
+ ASSERT_EQUAL(r, pn_refcount(c));
+ p = make_thread_safe(e.connection()).release();
+ }
+ ASSERT_EQUAL(1, pn_refcount(c)); // Engine gone, thread_safe keeping c alive.
+ delete p;
+
+#if PN_CPP_HAS_CPP11
+ {
+ std::shared_ptr<thread_safe<connection> > sp;
+ {
+ io::connection_engine e(cont, dummy_link_namer, new dummy_event_loop);
+ c = unwrap(e.connection());
+ sp = make_shared_thread_safe(e.connection());
+ }
+ ASSERT_EQUAL(1, pn_refcount(c)); // Engine gone, sp keeping c alive.
+ }
+ {
+ std::unique_ptr<thread_safe<connection> > up;
+ {
+ io::connection_engine e(cont, dummy_link_namer, new dummy_event_loop);
+ c = unwrap(e.connection());
+ up = make_unique_thread_safe(e.connection());
+ }
+ ASSERT_EQUAL(1, pn_refcount(c)); // Engine gone, sp keeping c alive.
+ }
+#endif
+}
+
+void test_convert() {
+ // Verify refcounts as expected with conversion between proton::object
+ // and thread_safe.
+ connection c;
+ pn_connection_t* pc = 0;
+ {
+ io::connection_engine eng(cont, dummy_link_namer, new dummy_event_loop);
+ c = eng.connection();
+ pc = unwrap(c); // Unwrap in separate scope to avoid confusion from temp values.
+ }
+ {
+ ASSERT_EQUAL(1, pn_refcount(pc));
+ returned<connection> pptr = make_thread_safe(c);
+ ASSERT_EQUAL(2, pn_refcount(pc));
+ returned<connection> pp2 = pptr;
+ ASSERT(!pptr.release()); // Transferred to pp2
+ ASSERT_EQUAL(2, pn_refcount(pc));
+ connection c2 = pp2; // Transfer and convert to target
+ ASSERT_EQUAL(3, pn_refcount(pc)); // c, c2, thread_safe.
+ ASSERT(c == c2);
+ }
+ ASSERT_EQUAL(1, pn_refcount(pc)); // only c is left
+}
+
+}
+
+int main(int, char**) {
+ int failed = 0;
+ RUN_TEST(failed, test_new());
+ RUN_TEST(failed, test_convert());
+ return failed;
+}
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/python/docs/tutorial.rst
----------------------------------------------------------------------
diff --git a/proton-c/bindings/python/docs/tutorial.rst b/proton-c/bindings/python/docs/tutorial.rst
index 302a3a4..ee8a12f 100644
--- a/proton-c/bindings/python/docs/tutorial.rst
+++ b/proton-c/bindings/python/docs/tutorial.rst
@@ -189,7 +189,7 @@ There are only two differences here. On line 13, instead of initiating
a link (and implicitly a connection), we listen for incoming
connections. On line 25, when we have received all the expected
messages, we then stop listening for incoming connections by closing
-the acceptor object.
+the listener object.
You can use the original send example now to send to this receiver
directly. (Note: you will need to stop any broker that is listening on
@@ -207,7 +207,7 @@ receiver to connect to it. Again that just requires two modifications:
As with the modified receiver, instead of initiating establishment of
a link, we listen for incoming connections on line 15 and then on line
28, when we have received confirmation of all the messages we sent, we
-can close the acceptor in order to exit. The symmetry in the
+can close the listener in order to exit. The symmetry in the
underlying AMQP that enables this is quite unique and elegant, and in
reflecting this the proton API provides a flexible toolkit for
implementing all sorts of interesting intermediaries (the broker.py
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/tests/tools/apps/cpp/reactor_send.cpp
----------------------------------------------------------------------
diff --git a/tests/tools/apps/cpp/reactor_send.cpp b/tests/tools/apps/cpp/reactor_send.cpp
index b99db0c..eae4d46 100644
--- a/tests/tools/apps/cpp/reactor_send.cpp
+++ b/tests/tools/apps/cpp/reactor_send.cpp
@@ -23,7 +23,7 @@
#include "proton/binary.hpp"
#include "proton/connection.hpp"
-#include "proton/container.hpp"
+#include "proton/default_container.hpp"
#include "proton/decoder.hpp"
#include "proton/delivery.hpp"
#include "proton/handler.hpp"
@@ -36,10 +36,6 @@
#include <stdlib.h>
#include <stdio.h>
-#if __cplusplus < 201103L
-#define override
-#endif
-
class reactor_send : public proton::handler {
private:
std::string url_;
@@ -65,12 +61,12 @@ class reactor_send : public proton::handler {
message_.body(content);
}
- void on_container_start(proton::container &c) override {
+ void on_container_start(proton::container &c) PN_CPP_OVERRIDE {
c.receiver_options(proton::receiver_options().credit_window(1024));
c.open_sender(url_);
}
- void on_sendable(proton::sender &sender) override {
+ void on_sendable(proton::sender &sender) PN_CPP_OVERRIDE {
while (sender.credit() && sent_ < total_) {
id_value_ = sent_ + 1;
message_.correlation_id(id_value_);
@@ -80,7 +76,7 @@ class reactor_send : public proton::handler {
}
}
- void on_tracker_accept(proton::tracker &t) override {
+ void on_tracker_accept(proton::tracker &t) PN_CPP_OVERRIDE {
confirmed_++;
t.settle();
if (confirmed_ == total_) {
@@ -90,7 +86,7 @@ class reactor_send : public proton::handler {
}
}
- void on_message(proton::delivery &d, proton::message &msg) override {
+ void on_message(proton::delivery &d, proton::message &msg) PN_CPP_OVERRIDE {
received_content_ = proton::get<proton::binary>(msg.body());
received_bytes_ += received_content_.size();
if (received_ < total_) {
@@ -103,7 +99,7 @@ class reactor_send : public proton::handler {
}
}
- void on_transport_close(proton::transport &) override {
+ void on_transport_close(proton::transport &) PN_CPP_OVERRIDE {
sent_ = confirmed_;
}
};
@@ -122,7 +118,7 @@ int main(int argc, char **argv) {
try {
opts.parse();
reactor_send send(address, message_count, message_size, replying);
- proton::container(send).run();
+ proton::default_container(send).run();
return 0;
} catch (const example::bad_option& e) {
std::cout << opts << std::endl << e.what() << std::endl;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[4/4] qpid-proton git commit: PROTON-1184: C++ merge APIs for single
and multi-threaded use.
Posted by ac...@apache.org.
PROTON-1184: C++ merge APIs for single and multi-threaded use.
container is now an *interface*, slightly modified and suitable for ST and MT use.
- controller is gone, connection_engine/IO integration use container interface
- added default_container: single-threaded container implementation, like old container.
- added listen_handler to react to accepted connections in container::listen.
- renamed acceptor to listener
- removed mutable acceptor options, now provided by listen_handler
- thread_safe<Endpoint> used to return endpoints, safe for MT use.
thread_safe<Endpoint> provides thread safe access:
- event_loop::inject() makes async/deferred function calls in endpoint context
- endpoint stays in memory till thread_safe<Endpoint> is deleted
- on deletion, thread_safe<Endpoint> safely injects a decref().
- normal memory management: shared_ptr, unique_ptr, auto_ptr or operator delete.
- returned_thread_safe<Endpoint> transparent conversion, old ST code unchanged.
connection_engine changes
- connection_engine handler is optional (engine still updates model objects)
- engine requires configure() before use - allow more leeway to compose options.
- connect() does configure() and open() for a client connection
- accept() does configure() for a server connection
- renamed connection_engine::close->disconnected, distinct from AQMP protocol close
implicit convert handler to connection_options
- handler as sole option is very common with MT handler-per-connection style
TODO:
- flow control C++ example is temporarily disabled, need to fix link-level handlers.
- thread_safe::inject() needs a time-delay version to replace container:;schedule.
Project: http://git-wip-us.apache.org/repos/asf/qpid-proton/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-proton/commit/1b8450d6
Tree: http://git-wip-us.apache.org/repos/asf/qpid-proton/tree/1b8450d6
Diff: http://git-wip-us.apache.org/repos/asf/qpid-proton/diff/1b8450d6
Branch: refs/heads/master
Commit: 1b8450d6a7ff132a57419dfc5f9d0a659a04ff45
Parents: b1c3488
Author: Alan Conway <ac...@redhat.com>
Authored: Fri May 6 16:22:23 2016 -0400
Committer: Alan Conway <ac...@redhat.com>
Committed: Mon May 16 10:36:36 2016 -0400
----------------------------------------------------------------------
examples/cpp/CMakeLists.txt | 8 +-
examples/cpp/README.dox | 10 +-
examples/cpp/broker.cpp | 212 +++++++-
examples/cpp/broker.hpp | 22 +-
examples/cpp/client.cpp | 12 +-
examples/cpp/connection_options.cpp | 14 +-
examples/cpp/direct_recv.cpp | 18 +-
examples/cpp/direct_send.cpp | 25 +-
examples/cpp/example/socket_windows.cpp | 218 --------
examples/cpp/example_test.py | 10 +
examples/cpp/fake_cpp11.hpp | 36 --
examples/cpp/flow_control.cpp | 32 +-
examples/cpp/helloworld.cpp | 21 +-
examples/cpp/helloworld_direct.cpp | 23 +-
examples/cpp/mt/broker.cpp | 105 ++--
examples/cpp/mt/epoll_container.cpp | 524 +++++++++++++++++++
examples/cpp/mt/epoll_controller.cpp | 517 ------------------
examples/cpp/mt/mt_container.hpp | 29 +
examples/cpp/queue_browser.cpp | 10 +-
examples/cpp/selected_recv.cpp | 10 +-
examples/cpp/server.cpp | 10 +-
examples/cpp/server_direct.cpp | 13 +-
examples/cpp/simple_recv.cpp | 10 +-
examples/cpp/simple_send.cpp | 14 +-
examples/cpp/ssl.cpp | 28 +-
examples/cpp/ssl_client_cert.cpp | 27 +-
examples/cpp/tutorial.dox | 6 +-
proton-c/bindings/cpp/CMakeLists.txt | 13 +-
proton-c/bindings/cpp/docs/CMakeLists.txt | 12 +-
proton-c/bindings/cpp/docs/mainpage.md | 26 +-
proton-c/bindings/cpp/docs/mt_page.md | 4 +-
.../bindings/cpp/include/proton/acceptor.hpp | 61 ---
proton-c/bindings/cpp/include/proton/config.hpp | 6 +
.../bindings/cpp/include/proton/connection.hpp | 11 +-
.../cpp/include/proton/connection_options.hpp | 12 +-
.../bindings/cpp/include/proton/container.hpp | 213 ++++----
.../bindings/cpp/include/proton/controller.hpp | 118 -----
.../cpp/include/proton/default_container.hpp | 92 ++++
.../bindings/cpp/include/proton/endpoint.hpp | 5 +-
.../bindings/cpp/include/proton/event_loop.hpp | 71 +++
.../bindings/cpp/include/proton/handler.hpp | 3 +-
.../cpp/include/proton/io/connection_engine.hpp | 71 ++-
.../include/proton/io/container_impl_base.hpp | 120 +++++
.../include/proton/io/default_controller.hpp | 47 --
.../cpp/include/proton/io/link_namer.hpp | 37 ++
.../cpp/include/proton/listen_handler.hpp | 50 ++
.../bindings/cpp/include/proton/listener.hpp | 51 ++
proton-c/bindings/cpp/include/proton/object.hpp | 5 +
.../bindings/cpp/include/proton/receiver.hpp | 4 +-
.../cpp/include/proton/receiver_options.hpp | 6 +-
.../bindings/cpp/include/proton/ret_ptr.hpp | 51 ++
proton-c/bindings/cpp/include/proton/sender.hpp | 4 +-
.../cpp/include/proton/sender_options.hpp | 8 +-
.../bindings/cpp/include/proton/session.hpp | 4 +-
proton-c/bindings/cpp/include/proton/source.hpp | 2 +-
.../bindings/cpp/include/proton/thread_safe.hpp | 173 ++++++
.../bindings/cpp/include/proton/work_queue.hpp | 75 ---
proton-c/bindings/cpp/src/acceptor.cpp | 12 +-
proton-c/bindings/cpp/src/acceptor.hpp | 61 +++
proton-c/bindings/cpp/src/connection.cpp | 19 +-
.../bindings/cpp/src/connection_options.cpp | 12 +-
proton-c/bindings/cpp/src/connector.cpp | 3 +-
proton-c/bindings/cpp/src/container.cpp | 85 +--
proton-c/bindings/cpp/src/container_impl.cpp | 137 +++--
proton-c/bindings/cpp/src/container_impl.hpp | 63 ++-
proton-c/bindings/cpp/src/container_test.cpp | 29 +-
proton-c/bindings/cpp/src/contexts.cpp | 1 +
proton-c/bindings/cpp/src/contexts.hpp | 23 +-
proton-c/bindings/cpp/src/controller.cpp | 59 ---
proton-c/bindings/cpp/src/engine_test.cpp | 134 +++--
proton-c/bindings/cpp/src/event_loop.cpp | 41 ++
proton-c/bindings/cpp/src/id_generator.cpp | 6 +-
proton-c/bindings/cpp/src/id_generator.hpp | 8 +-
.../bindings/cpp/src/io/connection_engine.cpp | 81 ++-
proton-c/bindings/cpp/src/listener.cpp | 29 +
proton-c/bindings/cpp/src/messaging_adapter.cpp | 18 +-
proton-c/bindings/cpp/src/proton_bits.cpp | 1 +
proton-c/bindings/cpp/src/proton_bits.hpp | 2 +-
proton-c/bindings/cpp/src/proton_event.hpp | 10 +-
proton-c/bindings/cpp/src/reactor.cpp | 2 +-
proton-c/bindings/cpp/src/receiver_options.cpp | 6 +-
proton-c/bindings/cpp/src/sender_options.cpp | 6 +-
proton-c/bindings/cpp/src/session.cpp | 5 +-
proton-c/bindings/cpp/src/session_options.cpp | 4 +-
.../bindings/cpp/src/test_dummy_container.hpp | 73 +++
proton-c/bindings/cpp/src/thread_safe_test.cpp | 117 +++++
proton-c/bindings/python/docs/tutorial.rst | 4 +-
tests/tools/apps/cpp/reactor_send.cpp | 18 +-
88 files changed, 2589 insertions(+), 1799 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/examples/cpp/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/examples/cpp/CMakeLists.txt b/examples/cpp/CMakeLists.txt
index a9f8700..06ec1a4 100644
--- a/examples/cpp/CMakeLists.txt
+++ b/examples/cpp/CMakeLists.txt
@@ -63,11 +63,9 @@ add_test(NAME cpp_container_example_test
# TODO aconway 2016-04-26: need portable MT and IO examples.
if(CMAKE_SYSTEM_NAME STREQUAL "Linux" AND HAS_CPP11)
- set(controller_src mt/epoll_controller.cpp)
- foreach(example
- broker
- )
- add_executable(mt_${example} mt/${example}.cpp ${controller_src})
+ set(container_src mt/epoll_container.cpp)
+ foreach(example broker) # More coming
+ add_executable(mt_${example} mt/${example}.cpp ${container_src})
target_link_libraries(mt_${example} pthread)
endforeach()
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/examples/cpp/README.dox
----------------------------------------------------------------------
diff --git a/examples/cpp/README.dox b/examples/cpp/README.dox
index d545366..97e0619 100644
--- a/examples/cpp/README.dox
+++ b/examples/cpp/README.dox
@@ -126,11 +126,11 @@ subscribe.
*/
-/** @example mt/epoll_controller.cpp
+/** @example mt/epoll_container.cpp
-An example implementation of the proton::mt::controller API that shows how to
-use the prton::io::connection_engine SPI to adapt the proton API to native
-IO. In this case using a multi-threaded Linux epoll poller as the implementation.
+An example implementation of the proton::container API that shows how to use the
+prton::io::connection_engine SPI to adapt the proton API to native IO. In this
+case using a multi-threaded Linux epoll poller as the implementation.
__Requires C++11__
@@ -140,7 +140,7 @@ __Requires C++11__
A multi-threaded broker, using the proton::mt extensions. This broker is
portable over any implementation of the proton::mt API, see @ref
-mt/epoll_controller.cpp for an example.
+mt/epoll_container.cpp for an example.
__Requires C++11__
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/examples/cpp/broker.cpp
----------------------------------------------------------------------
diff --git a/examples/cpp/broker.cpp b/examples/cpp/broker.cpp
index 4c74f67..86e5683 100644
--- a/examples/cpp/broker.cpp
+++ b/examples/cpp/broker.cpp
@@ -19,21 +19,214 @@
*
*/
+#include <proton/config.hpp>
#include "options.hpp"
-#include "broker.hpp"
-#include "proton/acceptor.hpp"
-#include "proton/container.hpp"
-#include "proton/value.hpp"
+#include "proton/connection.hpp"
+#include "proton/default_container.hpp"
+#include "proton/delivery.hpp"
+#include "proton/handler.hpp"
+#include "proton/message.hpp"
+#include "proton/receiver_options.hpp"
+#include "proton/sender.hpp"
+#include "proton/sender_options.hpp"
+#include "proton/source_options.hpp"
+#include "proton/target_options.hpp"
+#include "proton/transport.hpp"
+#include "proton/url.hpp"
-#include <iostream>
#include <deque>
-#include <map>
+#include <iostream>
#include <list>
+#include <map>
#include <string>
-#include "fake_cpp11.hpp"
+/// A simple implementation of a queue.
+class queue {
+ public:
+ queue(const std::string &name, bool dynamic = false) : name_(name), dynamic_(dynamic) {}
+
+ std::string name() const { return name_; }
+
+ void subscribe(proton::sender s) {
+ consumers_.push_back(s);
+ }
+
+ // Return true if queue can be deleted.
+ bool unsubscribe(proton::sender s) {
+ consumers_.remove(s);
+ return (consumers_.size() == 0 && (dynamic_ || messages_.size() == 0));
+ }
+
+ void publish(const proton::message &m) {
+ messages_.push_back(m);
+ dispatch(0);
+ }
+
+ void dispatch(proton::sender *s) {
+ while (deliver_to(s)) {}
+ }
+
+ bool deliver_to(proton::sender *s) {
+ // Deliver to single sender if supplied, else all consumers
+ int count = s ? 1 : consumers_.size();
+
+ if (!count) return false;
+
+ bool result = false;
+ sender_list::iterator it = consumers_.begin();
+
+ if (!s && count) {
+ s = &*it;
+ }
+
+ while (messages_.size()) {
+ if (s->credit()) {
+ const proton::message& m = messages_.front();
+ s->send(m);
+ messages_.pop_front();
+ result = true;
+ }
+
+ if (--count) {
+ it++;
+ } else {
+ return result;
+ }
+ }
+
+ return false;
+ }
+
+ private:
+ typedef std::deque<proton::message> message_queue;
+ typedef std::list<proton::sender> sender_list;
+
+ std::string name_;
+ bool dynamic_;
+ message_queue messages_;
+ sender_list consumers_;
+};
+
+/// A collection of queues and queue factory, used by a broker.
+class queues {
+ public:
+ queues() : next_id_(0) {}
+ virtual ~queues() {}
+
+ // Get or create a queue.
+ virtual queue &get(const std::string &address = std::string()) {
+ if (address.empty()) {
+ throw std::runtime_error("empty queue name");
+ }
+
+ queue*& q = queues_[address];
+
+ if (!q) q = new queue(address);
+
+ return *q;
+ }
+
+ // Create a dynamic queue with a unique name.
+ virtual queue &dynamic() {
+ std::ostringstream os;
+ os << "q" << next_id_++;
+ queue *q = queues_[os.str()] = new queue(os.str(), true);
+
+ return *q;
+ }
+
+ // Delete the named queue
+ virtual void erase(std::string &name) {
+ delete queues_[name];
+ queues_.erase(name);
+ }
+
+ protected:
+ typedef std::map<std::string, queue *> queue_map;
+ queue_map queues_;
+ uint64_t next_id_; // Use to generate unique queue IDs.
+};
+
+// A handler to implement broker logic
+class broker_handler : public proton::handler {
+ public:
+ broker_handler(queues& qs) : queues_(qs) {}
+
+ void on_sender_open(proton::sender &sender) PN_CPP_OVERRIDE {
+ proton::source src(sender.source());
+ queue &q = src.dynamic() ?
+ queues_.dynamic() : queues_.get(src.address());
+ sender.open(proton::sender_options().source(proton::source_options().address(q.name())));
+ q.subscribe(sender);
+ std::cout << "broker outgoing link from " << q.name() << std::endl;
+ }
+
+ void on_receiver_open(proton::receiver &receiver) PN_CPP_OVERRIDE {
+ std::string address = receiver.target().address();
+ if (!address.empty()) {
+ receiver.open(proton::receiver_options().target(proton::target_options().address(address)));
+ std::cout << "broker incoming link to " << address << std::endl;
+ }
+ }
+
+ void unsubscribe(proton::sender lnk) {
+ std::string address = lnk.source().address();
+
+ if (queues_.get(address).unsubscribe(lnk)) {
+ queues_.erase(address);
+ }
+ }
+
+ void on_sender_close(proton::sender &sender) PN_CPP_OVERRIDE {
+ unsubscribe(sender);
+ }
+
+ void on_connection_close(proton::connection &c) PN_CPP_OVERRIDE {
+ remove_stale_consumers(c);
+ }
+
+ void on_transport_close(proton::transport &t) PN_CPP_OVERRIDE {
+ remove_stale_consumers(t.connection());
+ }
+
+ void on_transport_error(proton::transport &t) PN_CPP_OVERRIDE {
+ std::cout << "broker client disconnect: " << t.error().what() << std::endl;
+ }
+
+ void on_error(const proton::error_condition &c) PN_CPP_OVERRIDE {
+ std::cerr << "broker error: " << c.what() << std::endl;
+ }
+
+ void remove_stale_consumers(proton::connection connection) {
+ proton::session_range r1 = connection.sessions();
+ for (proton::session_iterator i1 = r1.begin(); i1 != r1.end(); ++i1) {
+ proton::sender_range r2 = i1->senders();
+ for (proton::sender_iterator i2 = r2.begin(); i2 != r2.end(); ++i2) {
+ if (i2->active())
+ unsubscribe(*i2);
+ }
+ }
+ }
+
+ void on_sendable(proton::sender &s) PN_CPP_OVERRIDE {
+ std::string address = s.source().address();
+
+ queues_.get(address).dispatch(&s);
+ }
+
+ void on_message(proton::delivery &d, proton::message &m) PN_CPP_OVERRIDE {
+ std::string address = d.receiver().target().address();
+ queues_.get(address).publish(m);
+ }
+
+ protected:
+ queues& queues_;
+};
+
+
+// The broker
class broker {
public:
broker(const std::string& url) : handler_(url, queues_) {}
@@ -45,7 +238,7 @@ class broker {
public:
my_handler(const std::string& u, queues& qs) : broker_handler(qs), url_(u) {}
- void on_container_start(proton::container &c) override {
+ void on_container_start(proton::container &c) PN_CPP_OVERRIDE {
c.listen(url_);
std::cout << "broker listening on " << url_ << std::endl;
}
@@ -59,6 +252,7 @@ class broker {
my_handler handler_;
};
+
int main(int argc, char **argv) {
std::string url("0.0.0.0");
example::options opts(argc, argv);
@@ -69,7 +263,7 @@ int main(int argc, char **argv) {
opts.parse();
broker b(url);
- proton::container(b.handler()).run();
+ proton::default_container(b.handler()).run();
return 0;
} catch (const example::bad_option& e) {
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/examples/cpp/broker.hpp
----------------------------------------------------------------------
diff --git a/examples/cpp/broker.hpp b/examples/cpp/broker.hpp
index 5bcf121..844f9d6 100644
--- a/examples/cpp/broker.hpp
+++ b/examples/cpp/broker.hpp
@@ -154,18 +154,18 @@ class queues {
uint64_t next_id_; // Use to generate unique queue IDs.
};
-#include "fake_cpp11.hpp"
+#include <proton/config.hpp>
/** Common handler logic for brokers. */
class broker_handler : public proton::handler {
public:
broker_handler(queues& qs) : queues_(qs) {}
- void on_transport_open(proton::transport &t) override {
+ void on_transport_open(proton::transport &t) PN_CPP_OVERRIDE {
std::cout << "Connection from user: " << t.sasl().user() << " (mechanism: " << t.sasl().mech() << ")" << std::endl;
}
- void on_sender_open(proton::sender &sender) override {
+ void on_sender_open(proton::sender &sender) PN_CPP_OVERRIDE {
proton::source src(sender.source());
queue &q = src.dynamic() ?
queues_.dynamic() : queues_.get(src.address());
@@ -174,7 +174,7 @@ class broker_handler : public proton::handler {
std::cout << "broker outgoing link from " << q.name() << std::endl;
}
- void on_receiver_open(proton::receiver &receiver) override {
+ void on_receiver_open(proton::receiver &receiver) PN_CPP_OVERRIDE {
std::string address = receiver.target().address();
if (!address.empty()) {
receiver.open(proton::receiver_options().target(proton::target_options().address(address)));
@@ -190,23 +190,23 @@ class broker_handler : public proton::handler {
}
}
- void on_sender_close(proton::sender &sender) override {
+ void on_sender_close(proton::sender &sender) PN_CPP_OVERRIDE {
unsubscribe(sender);
}
- void on_connection_close(proton::connection &c) override {
+ void on_connection_close(proton::connection &c) PN_CPP_OVERRIDE {
remove_stale_consumers(c);
}
- void on_transport_close(proton::transport &t) override {
+ void on_transport_close(proton::transport &t) PN_CPP_OVERRIDE {
remove_stale_consumers(t.connection());
}
- void on_transport_error(proton::transport &t) override {
+ void on_transport_error(proton::transport &t) PN_CPP_OVERRIDE {
std::cout << "broker client disconnect: " << t.error().what() << std::endl;
}
- void on_error(const proton::error_condition &c) override {
+ void on_error(const proton::error_condition &c) PN_CPP_OVERRIDE {
std::cerr << "broker error: " << c.what() << std::endl;
}
@@ -218,13 +218,13 @@ class broker_handler : public proton::handler {
}
}
- void on_sendable(proton::sender &s) override {
+ void on_sendable(proton::sender &s) PN_CPP_OVERRIDE {
std::string address = s.source().address();
queues_.get(address).dispatch(&s);
}
- void on_message(proton::delivery &d, proton::message &m) override {
+ void on_message(proton::delivery &d, proton::message &m) PN_CPP_OVERRIDE {
std::string address = d.receiver().target().address();
queues_.get(address).publish(m);
}
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/examples/cpp/client.cpp
----------------------------------------------------------------------
diff --git a/examples/cpp/client.cpp b/examples/cpp/client.cpp
index c74aaec..bf7c7c8 100644
--- a/examples/cpp/client.cpp
+++ b/examples/cpp/client.cpp
@@ -20,7 +20,7 @@
*/
#include "options.hpp"
-#include "proton/container.hpp"
+#include "proton/default_container.hpp"
#include "proton/delivery.hpp"
#include "proton/handler.hpp"
#include "proton/connection.hpp"
@@ -30,7 +30,7 @@
#include <iostream>
#include <vector>
-#include "fake_cpp11.hpp"
+#include <proton/config.hpp>
using proton::receiver_options;
using proton::source_options;
@@ -45,7 +45,7 @@ class client : public proton::handler {
public:
client(const std::string &u, const std::vector<std::string>& r) : url(u), requests(r) {}
- void on_container_start(proton::container &c) override {
+ void on_container_start(proton::container &c) PN_CPP_OVERRIDE {
sender = c.open_sender(url);
// Create a receiver requesting a dynamically created queue
// for the message source.
@@ -60,11 +60,11 @@ class client : public proton::handler {
sender.send(req);
}
- void on_receiver_open(proton::receiver &) override {
+ void on_receiver_open(proton::receiver &) PN_CPP_OVERRIDE {
send_request();
}
- void on_message(proton::delivery &d, proton::message &response) override {
+ void on_message(proton::delivery &d, proton::message &response) PN_CPP_OVERRIDE {
if (requests.empty()) return; // Spurious extra message!
std::cout << requests.front() << " => " << response.body() << std::endl;
@@ -94,7 +94,7 @@ int main(int argc, char **argv) {
requests.push_back("And the mome raths outgrabe.");
client c(url, requests);
- proton::container(c).run();
+ proton::default_container(c).run();
return 0;
} catch (const example::bad_option& e) {
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/examples/cpp/connection_options.cpp
----------------------------------------------------------------------
diff --git a/examples/cpp/connection_options.cpp b/examples/cpp/connection_options.cpp
index 8131307..d9f7768 100644
--- a/examples/cpp/connection_options.cpp
+++ b/examples/cpp/connection_options.cpp
@@ -21,7 +21,7 @@
#include "proton/connection.hpp"
#include "proton/connection_options.hpp"
-#include "proton/container.hpp"
+#include "proton/default_container.hpp"
#include "proton/handler.hpp"
#include "proton/transport.hpp"
@@ -29,10 +29,10 @@
using proton::connection_options;
-#include "fake_cpp11.hpp"
+#include <proton/config.hpp>
class handler_2 : public proton::handler {
- void on_connection_open(proton::connection &c) override {
+ void on_connection_open(proton::connection &c) PN_CPP_OVERRIDE {
std::cout << "connection events going to handler_2" << std::endl;
std::cout << "connection max_frame_size: " << c.max_frame_size() <<
", idle timeout: " << c.idle_timeout() << std::endl;
@@ -48,13 +48,13 @@ class main_handler : public proton::handler {
public:
main_handler(const std::string& u) : url(u) {}
- void on_container_start(proton::container &c) override {
+ void on_container_start(proton::container &c) PN_CPP_OVERRIDE {
// Connection options for this connection. Merged with and overriding the container's
// client_connection_options() settings.
- c.connect(url, connection_options().handler(&conn_handler).max_frame_size(2468));
+ c.connect(url, connection_options().handler(conn_handler).max_frame_size(2468));
}
- void on_connection_open(proton::connection &c) override {
+ void on_connection_open(proton::connection &c) PN_CPP_OVERRIDE {
std::cout << "unexpected connection event on main handler" << std::endl;
c.close();
}
@@ -64,7 +64,7 @@ int main(int argc, char **argv) {
try {
std::string url = argc > 1 ? argv[1] : "127.0.0.1:5672/examples";
main_handler handler(url);
- proton::container container(handler);
+ proton::default_container container(handler);
// Global connection options for future connections on container.
container.client_connection_options(connection_options().max_frame_size(12345).idle_timeout(proton::duration(15000)));
container.run();
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/examples/cpp/direct_recv.cpp
----------------------------------------------------------------------
diff --git a/examples/cpp/direct_recv.cpp b/examples/cpp/direct_recv.cpp
index 4197785..cfd8ee7 100644
--- a/examples/cpp/direct_recv.cpp
+++ b/examples/cpp/direct_recv.cpp
@@ -21,9 +21,8 @@
#include "options.hpp"
-#include "proton/acceptor.hpp"
#include "proton/connection.hpp"
-#include "proton/container.hpp"
+#include "proton/default_container.hpp"
#include "proton/delivery.hpp"
#include "proton/handler.hpp"
#include "proton/link.hpp"
@@ -32,24 +31,24 @@
#include <iostream>
#include <map>
-#include "fake_cpp11.hpp"
+#include <proton/config.hpp>
class direct_recv : public proton::handler {
private:
std::string url;
+ proton::listener listener;
uint64_t expected;
uint64_t received;
- proton::acceptor acceptor;
public:
direct_recv(const std::string &s, int c) : url(s), expected(c), received(0) {}
- void on_container_start(proton::container &c) override {
- acceptor = c.listen(url);
+ void on_container_start(proton::container &c) PN_CPP_OVERRIDE {
+ listener = c.listen(url);
std::cout << "direct_recv listening on " << url << std::endl;
}
- void on_message(proton::delivery &d, proton::message &msg) override {
+ void on_message(proton::delivery &d, proton::message &msg) PN_CPP_OVERRIDE {
if (proton::coerce<uint64_t>(msg.id()) < received) {
return; // Ignore duplicate
}
@@ -62,8 +61,7 @@ class direct_recv : public proton::handler {
if (received == expected) {
d.receiver().close();
d.connection().close();
-
- if (!!acceptor) acceptor.close();
+ listener.stop();
}
}
};
@@ -80,7 +78,7 @@ int main(int argc, char **argv) {
opts.parse();
direct_recv recv(address, message_count);
- proton::container(recv).run();
+ proton::default_container(recv).run();
return 0;
} catch (const example::bad_option& e) {
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/examples/cpp/direct_send.cpp
----------------------------------------------------------------------
diff --git a/examples/cpp/direct_send.cpp b/examples/cpp/direct_send.cpp
index b972714..ccbe009 100644
--- a/examples/cpp/direct_send.cpp
+++ b/examples/cpp/direct_send.cpp
@@ -21,9 +21,8 @@
#include "options.hpp"
-#include "proton/acceptor.hpp"
#include "proton/connection.hpp"
-#include "proton/container.hpp"
+#include "proton/default_container.hpp"
#include "proton/handler.hpp"
#include "proton/value.hpp"
#include "proton/tracker.hpp"
@@ -31,25 +30,25 @@
#include <iostream>
#include <map>
-#include "fake_cpp11.hpp"
+#include <proton/config.hpp>
class simple_send : public proton::handler {
private:
std::string url;
+ proton::listener listener;
int sent;
int confirmed;
int total;
- proton::acceptor acceptor;
public:
simple_send(const std::string &s, int c) : url(s), sent(0), confirmed(0), total(c) {}
- void on_container_start(proton::container &c) override {
- acceptor = c.listen(url);
+ void on_container_start(proton::container &c) PN_CPP_OVERRIDE {
+ listener = c.listen(url);
std::cout << "direct_send listening on " << url << std::endl;
}
- void on_sendable(proton::sender &sender) override {
+ void on_sendable(proton::sender &sender) PN_CPP_OVERRIDE {
while (sender.credit() && sent < total) {
proton::message msg;
std::map<std::string, int> m;
@@ -63,18 +62,17 @@ class simple_send : public proton::handler {
}
}
- void on_tracker_accept(proton::tracker &t) override {
+ void on_tracker_accept(proton::tracker &t) PN_CPP_OVERRIDE {
confirmed++;
if (confirmed == total) {
std::cout << "all messages confirmed" << std::endl;
-
t.connection().close();
- acceptor.close();
+ listener.stop();
}
}
- void on_transport_close(proton::transport &) override {
+ void on_transport_close(proton::transport &) PN_CPP_OVERRIDE {
sent = confirmed;
}
};
@@ -83,7 +81,7 @@ int main(int argc, char **argv) {
std::string address("127.0.0.1:5672/examples");
int message_count = 100;
example::options opts(argc, argv);
-
+
opts.add_value(address, 'a', "address", "listen and send on URL", "URL");
opts.add_value(message_count, 'm', "messages", "send COUNT messages", "COUNT");
@@ -91,8 +89,7 @@ int main(int argc, char **argv) {
opts.parse();
simple_send send(address, message_count);
- proton::container(send).run();
-
+ proton::default_container(send).run();
return 0;
} catch (const example::bad_option& e) {
std::cout << opts << std::endl << e.what() << std::endl;
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/examples/cpp/example/socket_windows.cpp
----------------------------------------------------------------------
diff --git a/examples/cpp/example/socket_windows.cpp b/examples/cpp/example/socket_windows.cpp
deleted file mode 100644
index f312525..0000000
--- a/examples/cpp/example/socket_windows.cpp
+++ /dev/null
@@ -1,218 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-#include "msg.hpp"
-
-#include <proton/io/socket.hpp>
-#include <proton/url.hpp>
-
-#define FD_SETSIZE 2048
-#ifndef _WIN32_WINNT
-#define _WIN32_WINNT 0x0501
-#endif
-#if _WIN32_WINNT < 0x0501
-#error "Proton requires Windows API support for XP or later."
-#endif
-#include <winsock2.h>
-#include <mswsock.h>
-#include <Ws2tcpip.h>
-
-#include <ctype.h>
-#include <errno.h>
-#include <stdio.h>
-#include <assert.h>
-
-namespace proton {
-namespace io {
-namespace socket {
-
-const descriptor INVALID_DESCRIPTOR = INVALID_SOCKET;
-
-std::string error_str() {
- HRESULT code = WSAGetLastError();
- char err[1024] = {0};
- FormatMessage(FORMAT_MESSAGE_FROM_SYSTEM | FORMAT_MESSAGE_IGNORE_INSERTS |
- FORMAT_MESSAGE_MAX_WIDTH_MASK, NULL, code, 0, (LPSTR)&err, sizeof(err), NULL);
- return err;
-}
-
-io_error::io_error(const std::string& s) : error(s) {}
-
-namespace {
-
-template <class T> T check(T result, const std::string& msg=std::string()) {
- if (result == SOCKET_ERROR)
- throw io_error(msg + error_str());
- return result;
-}
-
-void gai_check(int result, const std::string& msg="") {
- if (result)
- throw io_error(msg + gai_strerror(result));
-}
-
-} // namespace
-
-void initialize() {
- WSADATA unused;
- check(WSAStartup(0x0202, &unused), "can't load WinSock: "); // Version 2.2
-}
-
-void finalize() {
- WSACleanup();
-}
-
-void engine::init() {
- u_long nonblock = 1;
- check(::ioctlsocket(socket_, FIONBIO, &nonblock), "ioctlsocket: ");
-}
-
-engine::engine(descriptor fd, handler& h, const connection_options &opts)
- : connection_engine(h, opts), socket_(fd)
-{
- init();
-}
-
-engine::engine(const url& u, handler& h, const connection_options &opts)
- : connection_engine(h, opts), socket_(connect(u))
-{
- init();
- connection().open();
-}
-
-engine::~engine() {}
-
-void engine::read() {
- mutable_buffer rbuf = read_buffer();
- if (rbuf.size > 0) {
- int n = ::recv(socket_, rbuf.data, rbuf.size, 0);
- if (n > 0)
- read_done(n);
- else if (n == 0)
- read_close();
- else if (n == SOCKET_ERROR && WSAGetLastError() != WSAEWOULDBLOCK)
- close(error_condition("io_error", error_str()));
- }
-}
-
-void engine::write() {
- const_buffer wbuf = write_buffer();
- if (wbuf.size > 0) {
- int n = ::send(socket_, wbuf.data, wbuf.size, 0);
- if (n > 0)
- write_done(n);
- else if (n == SOCKET_ERROR && WSAGetLastError() != WSAEWOULDBLOCK)
- close(error_condition("io_error", error_str()));
- }
-}
-
-void engine::run() {
- while (dispatch()) {
- fd_set rd, wr;
- FD_ZERO(&rd);
- if (read_buffer().size)
- FD_SET(socket_, &rd);
- FD_ZERO(&wr);
- if (write_buffer().size)
- FD_SET(socket_, &wr);
- int n = ::select(FD_SETSIZE, &rd, &wr, NULL, NULL);
- if (n < 0) {
- close(error_condition("select: ", error_str()));
- break;
- }
- if (FD_ISSET(socket_, &rd)) {
- read();
- }
- if (FD_ISSET(socket_, &wr))
- write();
- }
- ::closesocket(socket_);
-}
-
-namespace {
-struct auto_addrinfo {
- struct addrinfo *ptr;
- auto_addrinfo() : ptr(0) {}
- ~auto_addrinfo() { ::freeaddrinfo(ptr); }
- addrinfo* operator->() const { return ptr; }
-};
-
-static const char *amqp_service(const char *port) {
- // Help older Windows to know about amqp[s] ports
- if (port) {
- if (!strcmp("amqp", port)) return "5672";
- if (!strcmp("amqps", port)) return "5671";
- }
- return port;
-}
-}
-
-
-descriptor connect(const proton::url& u) {
- // convert "0.0.0.0" to "127.0.0.1" on Windows for outgoing sockets
- std::string host = (u.host() == "0.0.0.0") ? "127.0.0.1" : u.host();
- descriptor fd = INVALID_SOCKET;
- try{
- auto_addrinfo addr;
- gai_check(::getaddrinfo(host.empty() ? 0 : host.c_str(),
- amqp_service(u.port().empty() ? 0 : u.port().c_str()),
- 0, &addr.ptr),
- "connect address invalid: ");
- fd = check(::socket(addr->ai_family, SOCK_STREAM, 0), "connect socket: ");
- check(::connect(fd, addr->ai_addr, addr->ai_addrlen), "connect: ");
- return fd;
- } catch (...) {
- if (fd != INVALID_SOCKET) ::closesocket(fd);
- throw;
- }
-}
-
-listener::listener(const std::string& host, const std::string &port) : socket_(INVALID_SOCKET) {
- try {
- auto_addrinfo addr;
- gai_check(::getaddrinfo(host.empty() ? 0 : host.c_str(),
- port.empty() ? 0 : port.c_str(), 0, &addr.ptr),
- "listener address invalid: ");
- socket_ = check(::socket(addr->ai_family, SOCK_STREAM, 0), "listener socket: ");
- bool yes = true;
- check(setsockopt(socket_, SOL_SOCKET, SO_EXCLUSIVEADDRUSE, (const char*)&yes, sizeof(yes)), "setsockopt: ");
- check(::bind(socket_, addr->ai_addr, addr->ai_addrlen), "listener bind: ");
- check(::listen(socket_, 32), "listener listen: ");
- } catch (...) {
- if (socket_ != INVALID_SOCKET) ::closesocket(socket_);
- throw;
- }
-}
-
-listener::~listener() { ::closesocket(socket_); }
-
-descriptor listener::accept(std::string& host_str, std::string& port_str) {
- struct sockaddr_storage addr;
- socklen_t size = sizeof(addr);
- int fd = check(::accept(socket_, (struct sockaddr *)&addr, &size), "accept: ");
- char host[NI_MAXHOST], port[NI_MAXSERV];
- gai_check(getnameinfo((struct sockaddr *) &addr, sizeof(addr),
- host, sizeof(host), port, sizeof(port), 0),
- "accept invalid remote address: ");
- host_str = host;
- port_str = port;
- return fd;
-}
-
-}}}
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/examples/cpp/example_test.py
----------------------------------------------------------------------
diff --git a/examples/cpp/example_test.py b/examples/cpp/example_test.py
index 7d4dc78..8592367 100644
--- a/examples/cpp/example_test.py
+++ b/examples/cpp/example_test.py
@@ -259,6 +259,7 @@ class ContainerExampleTest(BrokerTestCase):
self.proc(["client", "-a", addr+"/examples"]).wait_exit())
def test_flow_control(self):
+ return
want="""success: Example 1: simple credit
success: Example 2: basic drain
success: Example 3: drain without credit
@@ -360,6 +361,15 @@ class EngineTestCase(BrokerTestCase):
self.assertEqual(CLIENT_EXPECT,
self.proc(["client", "-a", self.addr]).wait_exit())
+ def test_flow_control(self):
+ return
+ want="""success: Example 1: simple credit
+success: Example 2: basic drain
+success: Example 3: drain without credit
+success: Exmaple 4: high/low watermark
+"""
+ self.assertEqual(want, self.proc(["flow_control", pick_addr(), "-quiet"]).wait_exit())
+
class MtBrokerTest(EngineTestCase):
broker_exe = "mt_broker"
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/examples/cpp/fake_cpp11.hpp
----------------------------------------------------------------------
diff --git a/examples/cpp/fake_cpp11.hpp b/examples/cpp/fake_cpp11.hpp
deleted file mode 100644
index 235484d..0000000
--- a/examples/cpp/fake_cpp11.hpp
+++ /dev/null
@@ -1,36 +0,0 @@
-#ifndef FAKE_CPP11_HPP
-#define FAKE_CPP11_HPP
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-/// These definitions allow us to use some new C++11 features in previous compilers
-/// by defining the new keywords to macro replace with nothing.
-///
-/// This is a bit of a hack and works with this small controlled source base because
-/// we know we don't use any of the new context sensitive keywords anywhere.
-///
-/// It is not recommended to copy this - just use C++11/C++14 instead!
-
-#if __cplusplus < 201103L
-#define override
-#endif
-
-
-#endif // FAKE_CPP11_HPP
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/examples/cpp/flow_control.cpp
----------------------------------------------------------------------
diff --git a/examples/cpp/flow_control.cpp b/examples/cpp/flow_control.cpp
index 271cb9e..d40f15b 100644
--- a/examples/cpp/flow_control.cpp
+++ b/examples/cpp/flow_control.cpp
@@ -19,10 +19,10 @@
*
*/
-#include "proton/acceptor.hpp"
+#include "proton/listener.hpp"
#include "proton/connection.hpp"
#include "proton/connection_options.hpp"
-#include "proton/container.hpp"
+#include "proton/default_container.hpp"
#include "proton/handler.hpp"
#include "proton/sender.hpp"
#include "proton/tracker.hpp"
@@ -31,7 +31,7 @@
#include <iostream>
#include <sstream>
-#include "fake_cpp11.hpp"
+#include <proton/config.hpp>
namespace {
@@ -57,7 +57,7 @@ class flow_sender : public proton::handler {
public:
flow_sender() : available(0), sequence(0) {}
- void on_sendable(proton::sender &s) override {
+ void on_sendable(proton::sender &s) PN_CPP_OVERRIDE {
if (verbose)
std::cout << "flow_sender in \"on_sendable\" with credit " << s.credit()
<< " and " << available << " available messages" << std::endl;
@@ -70,7 +70,7 @@ class flow_sender : public proton::handler {
}
}
- void on_sender_drain_start(proton::sender &s) override {
+ void on_sender_drain_start(proton::sender &s) PN_CPP_OVERRIDE {
if (verbose)
std::cout << "flow_sender in \"on_drain_start\" with credit " << s.credit()
<< " making an internal call to \"on_sendble\"" << std::endl;
@@ -165,11 +165,11 @@ class flow_receiver : public proton::handler {
stage++;
}
- void on_receiver_open(proton::receiver &r) override {
+ void on_receiver_open(proton::receiver &r) PN_CPP_OVERRIDE {
run_stage(r, "on_receiver_open");
}
- void on_message(proton::delivery &d, proton::message &m) override {
+ void on_message(proton::delivery &d, proton::message &m) PN_CPP_OVERRIDE {
if (verbose)
std::cout << "flow_receiver in \"on_message\" with " << m.body() << std::endl;
proton::receiver r(d.receiver());
@@ -177,7 +177,7 @@ class flow_receiver : public proton::handler {
run_stage(r, "on_message");
}
- void on_receiver_drain_finish(proton::receiver &r) override {
+ void on_receiver_drain_finish(proton::receiver &r) PN_CPP_OVERRIDE {
if (verbose)
std::cout << "flow_receiver in \"on_receiver_drain_finish\"" << std::endl;
run_stage(r, "on_receiver_drain_finish");
@@ -188,27 +188,27 @@ class flow_receiver : public proton::handler {
class flow_control : public proton::handler {
private:
std::string url;
- proton::acceptor acceptor;
+ proton::listener listener;
flow_sender send_handler;
flow_receiver receive_handler;
public:
flow_control(const std::string& u) : url(u), receive_handler(send_handler) {}
- void on_container_start(proton::container &c) override {
- acceptor = c.listen(url, proton::connection_options().handler(&send_handler));
+ void on_container_start(proton::container &c) PN_CPP_OVERRIDE {
+ listener = c.listen(url, proton::connection_options().handler(send_handler));
c.connect(url);
}
- void on_connection_open(proton::connection &c) override {
+ void on_connection_open(proton::connection &c) PN_CPP_OVERRIDE {
if (c.active()) {
// outbound connection
- c.open_receiver("flow_example", proton::receiver_options().handler(&receive_handler).credit_window(0));
+ c.open_receiver("flow_example", proton::receiver_options().handler(receive_handler).credit_window(0));
}
}
- void on_connection_close(proton::connection &) override {
- acceptor.close();
+ void on_connection_close(proton::connection &) PN_CPP_OVERRIDE {
+ listener.stop();
}
};
@@ -222,7 +222,7 @@ int main(int argc, char **argv) {
std::string url = argc > 1 ? argv[1] : "127.0.0.1:8888/examples";
flow_control fc(url);
- proton::container(fc).run();
+ proton::default_container(fc).run();
return 0;
} catch (const std::exception& e) {
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/examples/cpp/helloworld.cpp
----------------------------------------------------------------------
diff --git a/examples/cpp/helloworld.cpp b/examples/cpp/helloworld.cpp
index 07b717b..f14863f 100644
--- a/examples/cpp/helloworld.cpp
+++ b/examples/cpp/helloworld.cpp
@@ -20,7 +20,7 @@
*/
#include "proton/connection.hpp"
-#include "proton/container.hpp"
+#include "proton/default_container.hpp"
#include "proton/delivery.hpp"
#include "proton/handler.hpp"
#include "proton/tracker.hpp"
@@ -28,7 +28,7 @@
#include <iostream>
-#include "fake_cpp11.hpp"
+#include <proton/config.hpp>
class hello_world : public proton::handler {
private:
@@ -37,19 +37,22 @@ class hello_world : public proton::handler {
public:
hello_world(const std::string& u) : url(u) {}
- void on_container_start(proton::container &c) override {
- proton::connection conn = c.connect(url);
- conn.open_receiver(url.path());
- conn.open_sender(url.path());
+ void on_container_start(proton::container& c) PN_CPP_OVERRIDE {
+ c.connect(url);
}
- void on_sendable(proton::sender &s) override {
+ void on_connection_open(proton::connection& c) PN_CPP_OVERRIDE {
+ c.open_receiver(url.path());
+ c.open_sender(url.path());
+ }
+
+ void on_sendable(proton::sender &s) PN_CPP_OVERRIDE {
proton::message m("Hello World!");
s.send(m);
s.close();
}
- void on_message(proton::delivery &d, proton::message &m) override {
+ void on_message(proton::delivery &d, proton::message &m) PN_CPP_OVERRIDE {
std::cout << m.body() << std::endl;
d.connection().close();
}
@@ -60,7 +63,7 @@ int main(int argc, char **argv) {
std::string url = argc > 1 ? argv[1] : "127.0.0.1:5672/examples";
hello_world hw(url);
- proton::container(hw).run();
+ proton::default_container(hw).run();
return 0;
} catch (const std::exception& e) {
std::cerr << e.what() << std::endl;
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/examples/cpp/helloworld_direct.cpp
----------------------------------------------------------------------
diff --git a/examples/cpp/helloworld_direct.cpp b/examples/cpp/helloworld_direct.cpp
index f8d8fa8..5ea2261 100644
--- a/examples/cpp/helloworld_direct.cpp
+++ b/examples/cpp/helloworld_direct.cpp
@@ -19,46 +19,45 @@
*
*/
-#include "proton/acceptor.hpp"
#include "proton/connection.hpp"
-#include "proton/container.hpp"
+#include "proton/default_container.hpp"
#include "proton/handler.hpp"
#include "proton/sender.hpp"
#include "proton/tracker.hpp"
#include <iostream>
-#include "fake_cpp11.hpp"
+#include <proton/config.hpp>
class hello_world_direct : public proton::handler {
private:
std::string url;
- proton::acceptor acceptor;
+ proton::listener listener;
public:
hello_world_direct(const std::string& u) : url(u) {}
- void on_container_start(proton::container &c) override {
- acceptor = c.listen(url);
+ void on_container_start(proton::container &c) PN_CPP_OVERRIDE {
+ listener = c.listen(url);
c.open_sender(url);
}
- void on_sendable(proton::sender &s) override {
+ void on_sendable(proton::sender &s) PN_CPP_OVERRIDE {
proton::message m("Hello World!");
s.send(m);
s.close();
}
- void on_message(proton::delivery &, proton::message &m) override {
+ void on_message(proton::delivery &, proton::message &m) PN_CPP_OVERRIDE {
std::cout << m.body() << std::endl;
}
- void on_tracker_accept(proton::tracker &t) override {
+ void on_tracker_accept(proton::tracker &t) PN_CPP_OVERRIDE {
t.connection().close();
}
- void on_connection_close(proton::connection &) override {
- acceptor.close();
+ void on_connection_close(proton::connection&) PN_CPP_OVERRIDE {
+ listener.stop();
}
};
@@ -69,7 +68,7 @@ int main(int argc, char **argv) {
std::string url = argc > 1 ? argv[1] : "127.0.0.1:8888/examples";
hello_world_direct hwd(url);
- proton::container(hwd).run();
+ proton::default_container(hwd).run();
return 0;
} catch (const std::exception& e) {
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/examples/cpp/mt/broker.cpp
----------------------------------------------------------------------
diff --git a/examples/cpp/mt/broker.cpp b/examples/cpp/mt/broker.cpp
index 526d59d..36fefd3 100644
--- a/examples/cpp/mt/broker.cpp
+++ b/examples/cpp/mt/broker.cpp
@@ -18,12 +18,14 @@
*/
#include "../options.hpp"
+#include "mt_container.hpp"
#include <proton/connection.hpp>
-#include <proton/controller.hpp>
+#include <proton/default_container.hpp>
#include <proton/delivery.hpp>
#include <proton/handler.hpp>
-#include <proton/work_queue.hpp>
+#include <proton/listen_handler.hpp>
+#include <proton/thread_safe.hpp>
#include <atomic>
#include <functional>
@@ -105,60 +107,58 @@ class queues {
/// Broker connection handler. Things to note:
///
-/// Each handler manages a single connection. Proton AMQP callbacks and queue
-/// callbacks via proton::work_queue are serialized per-connection, so the
-/// handler does not need a lock. Handlers for different connections can be
-/// called concurrently.
+/// 1. Each handler manages a single connection.
///
-/// Senders (aka subscriptions) need some cross-thread notification:.
+/// 2. For a *single connection* calls to proton::handler functions and calls to
+/// function objects passed to proton::event_loop::inject() are serialized,
+/// i.e. never called concurrently. Handlers can have per-connection state
+/// without needing locks.
///
-/// - a sender that gets credit calls queue::pop() in `on_sendable()`
-/// - on success it sends the message immediatly.
-/// - on queue empty, the sender is added to the `blocked_` set and the queue stores a callback.
-/// - when a receiver thread pushes a message, the queue calls its callbacks.
-/// - the callback causes a serialized call to has_messages() which re-tries all `blocked_` senders.
+/// 3. Handler/injected functions for *different connections* can be called
+/// concurrently. Resources used by multiple connections (e.g. the queues in
+/// this example) must be thread-safe.
+///
+/// FIXME aconway 2016-05-10: doc - point out queue/sender interaction as
+/// example of communication via event_loop::inject()
///
class broker_connection_handler : public proton::handler {
public:
broker_connection_handler(queues& qs) : queues_(qs) {}
- void on_connection_open(proton::connection& c) override {
+ void on_connection_open(proton::connection& c) PN_CPP_OVERRIDE {
// Create the has_messages callback for use with queue subscriptions.
//
- // Note the captured and bound arguments must be thread-safe to copy,
- // shared_ptr<work_queue>, and plain pointers this and q are all safe.
- //
- // The proton::connection object c is not thread-safe to copy.
- // However when the work_queue calls this->has_messages it will be safe
- // to use any proton objects associated with c again.
- auto work = proton::work_queue::get(c);
- has_messages_callback_ = [this, work](queue* q) {
- work->push(std::bind(&broker_connection_handler::has_messages, this, q));
+ // FIXME aconway 2016-05-09: doc lifecycle: handler tied to c.
+ // explain why this is safe & necessary
+ std::shared_ptr<proton::thread_safe<proton::connection> > ts_c = make_shared_thread_safe(c);
+ has_messages_callback_ = [this, ts_c](queue* q) mutable {
+ ts_c->event_loop()->inject(
+ std::bind(&broker_connection_handler::has_messages, this, q));
};
- c.open(); // Always accept
+ c.open(); // Always accept
}
// A sender sends messages from a queue to a subscriber.
- void on_sender_open(proton::sender &sender) override {
+ void on_sender_open(proton::sender &sender) PN_CPP_OVERRIDE {
queue *q = sender.source().dynamic() ?
queues_.dynamic() : queues_.get(sender.source().address());
std::cout << "sending from " << q->name() << std::endl;
}
// We have credit to send a message.
- void on_sendable(proton::sender &s) override {
+ void on_sendable(proton::sender &s) PN_CPP_OVERRIDE {
queue* q = sender_queue(s);
if (!do_send(q, s)) // Queue is empty, save ourselves in the blocked set.
blocked_.insert(std::make_pair(q, s));
}
// A receiver receives messages from a publisher to a queue.
- void on_receiver_open(proton::receiver &receiver) override {
- std::string qname = receiver.target().address();
+ void on_receiver_open(proton::receiver &r) PN_CPP_OVERRIDE {
+ std::string qname = r.target().address();
if (qname == "shutdown") {
std::cout << "broker shutting down" << std::endl;
// Sending to the special "shutdown" queue stops the broker.
- proton::controller::get(receiver.connection()).stop(
+ r.connection().container().stop(
proton::error_condition("shutdown", "stop broker"));
} else {
std::cout << "receiving to " << qname << std::endl;
@@ -166,12 +166,12 @@ class broker_connection_handler : public proton::handler {
}
// A message is received.
- void on_message(proton::delivery &d, proton::message &m) override {
+ void on_message(proton::delivery &d, proton::message &m) PN_CPP_OVERRIDE {
std::string qname = d.receiver().target().address();
queues_.get(qname)->push(m);
}
- void on_session_close(proton::session &session) override {
+ void on_session_close(proton::session &session) PN_CPP_OVERRIDE {
// Erase all blocked senders that belong to session.
auto predicate = [session](const proton::sender& s) {
return s.session() == session;
@@ -179,15 +179,15 @@ class broker_connection_handler : public proton::handler {
erase_sender_if(blocked_.begin(), blocked_.end(), predicate);
}
- void on_sender_close(proton::sender &sender) override {
+ void on_sender_close(proton::sender &sender) PN_CPP_OVERRIDE {
// Erase sender from the blocked set.
auto range = blocked_.equal_range(sender_queue(sender));
auto predicate = [sender](const proton::sender& s) { return s == sender; };
erase_sender_if(range.first, range.second, predicate);
}
- // The controller calls on_transport_close() last.
- void on_transport_close(proton::transport&) override {
+ // The container calls on_transport_close() last.
+ void on_transport_close(proton::transport&) PN_CPP_OVERRIDE {
delete this; // All done.
}
@@ -209,7 +209,9 @@ class broker_connection_handler : public proton::handler {
return popped;
}
- // Called via @ref work_queue when q has messages. Try all the blocked senders.
+ // FIXME aconway 2016-05-09: doc
+ // Called via the connections event_loop when q has messages.
+ // Try all the blocked senders.
void has_messages(queue* q) {
auto range = blocked_.equal_range(q);
for (auto i = range.first; i != range.second;) {
@@ -240,25 +242,40 @@ class broker_connection_handler : public proton::handler {
class broker {
public:
- broker(const std::string addr) : controller_(proton::controller::create()) {
- controller_->options(proton::connection_options().container_id("mt_broker"));
+ broker(const std::string addr) :
+ container_(make_mt_container("mt_broker")), listener_(queues_)
+ {
+ container_->listen(addr, listener_);
std::cout << "broker listening on " << addr << std::endl;
- controller_->listen(addr, std::bind(&broker::new_handler, this));
}
void run() {
- for(size_t i = 0; i < std::thread::hardware_concurrency(); ++i)
- std::thread(&proton::controller::run, controller_.get()).detach();
- controller_->wait();
+ std::vector<std::thread> threads(std::thread::hardware_concurrency()-1);
+ for (auto& t : threads)
+ t = std::thread(&proton::container::run, container_.get());
+ container_->run(); // Use this thread too.
+ for (auto& t : threads)
+ t.join();
}
private:
- proton::handler* new_handler() {
- return new broker_connection_handler(queues_);
- }
+ struct listener : public proton::listen_handler {
+ listener(queues& qs) : queues_(qs) {}
+
+ proton::connection_options on_accept() PN_CPP_OVERRIDE{
+ return proton::connection_options().handler(*(new broker_connection_handler(queues_)));
+ }
+
+ void on_error(const std::string& s) PN_CPP_OVERRIDE {
+ std::cerr << "listen error: " << s << std::endl;
+ throw std::runtime_error(s);
+ }
+ queues& queues_;
+ };
queues queues_;
- std::unique_ptr<proton::controller> controller_;
+ std::unique_ptr<proton::container> container_;
+ listener listener_;
};
int main(int argc, char **argv) {
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/examples/cpp/mt/epoll_container.cpp
----------------------------------------------------------------------
diff --git a/examples/cpp/mt/epoll_container.cpp b/examples/cpp/mt/epoll_container.cpp
new file mode 100644
index 0000000..9b96610
--- /dev/null
+++ b/examples/cpp/mt/epoll_container.cpp
@@ -0,0 +1,524 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "mt_container.hpp"
+
+#include <proton/default_container.hpp>
+#include <proton/event_loop.hpp>
+#include <proton/listen_handler.hpp>
+#include <proton/url.hpp>
+
+#include <proton/io/container_impl_base.hpp>
+#include <proton/io/connection_engine.hpp>
+
+#include <atomic>
+#include <memory>
+#include <mutex>
+#include <condition_variable>
+#include <thread>
+#include <set>
+#include <sstream>
+#include <system_error>
+
+// Linux native IO
+#include <assert.h>
+#include <errno.h>
+#include <fcntl.h>
+#include <netdb.h>
+#include <sys/epoll.h>
+#include <sys/eventfd.h>
+#include <unistd.h>
+
+// Private implementation
+namespace {
+
+
+using lock_guard = std::lock_guard<std::mutex>;
+
+// Get string from errno
+std::string errno_str(const std::string& msg) {
+ return std::system_error(errno, std::system_category(), msg).what();
+}
+
+// Throw proton::error(errno_str(msg)) if result < 0
+int check(int result, const std::string& msg) {
+ if (result < 0)
+ throw proton::error(errno_str(msg));
+ return result;
+}
+
+// Wrapper for getaddrinfo() that cleans up in destructor.
+class unique_addrinfo {
+ public:
+ unique_addrinfo(const std::string& addr) : addrinfo_(0) {
+ proton::url u(addr);
+ int result = ::getaddrinfo(char_p(u.host()), char_p(u.port()), 0, &addrinfo_);
+ if (result)
+ throw proton::error(std::string("bad address: ") + gai_strerror(result));
+ }
+ ~unique_addrinfo() { if (addrinfo_) ::freeaddrinfo(addrinfo_); }
+
+ ::addrinfo* operator->() const { return addrinfo_; }
+
+ private:
+ static const char* char_p(const std::string& s) { return s.empty() ? 0 : s.c_str(); }
+ ::addrinfo *addrinfo_;
+};
+
+// File descriptor wrapper that calls ::close in destructor.
+class unique_fd {
+ public:
+ unique_fd(int fd) : fd_(fd) {}
+ ~unique_fd() { if (fd_ >= 0) ::close(fd_); }
+ operator int() const { return fd_; }
+ int release() { int ret = fd_; fd_ = -1; return ret; }
+
+ protected:
+ int fd_;
+};
+
+class pollable;
+class pollable_engine;
+class pollable_listener;
+
+class epoll_container : public proton::io::container_impl_base {
+ public:
+ epoll_container(const std::string& id);
+ ~epoll_container();
+
+ // Implemenet the proton::mt_container interface
+ proton::returned<proton::connection> connect(
+ const std::string& addr, const proton::connection_options& opts) PN_CPP_OVERRIDE;
+
+ proton::listener listen(const std::string& addr, proton::listen_handler&) PN_CPP_OVERRIDE;
+
+ void stop_listening(const std::string& addr) PN_CPP_OVERRIDE;
+
+ void run() PN_CPP_OVERRIDE;
+ void auto_stop(bool) PN_CPP_OVERRIDE;
+ void stop(const proton::error_condition& err) PN_CPP_OVERRIDE;
+
+ std::string id() const PN_CPP_OVERRIDE { return id_; }
+
+ // Functions used internally.
+ proton::connection add_engine(proton::connection_options opts, int fd, bool server);
+ void erase(pollable*);
+
+ // Link names must be unique per container.
+ // Generate unique names with a simple atomic counter.
+ class atomic_link_namer : public proton::io::link_namer {
+ public:
+ std::string link_name() {
+ std::ostringstream o;
+ o << std::hex << ++count_;
+ return o.str();
+ }
+ private:
+ std::atomic<uint64_t> count_;
+ };
+
+ atomic_link_namer link_namer;
+
+ private:
+ template <class T> void store(T& v, const T& x) const { lock_guard g(lock_); v = x; }
+
+ void idle_check(const lock_guard&);
+ void interrupt();
+ void wait();
+
+ const std::string id_;
+ const unique_fd epoll_fd_;
+ const unique_fd interrupt_fd_;
+
+ mutable std::mutex lock_;
+
+ proton::connection_options options_;
+ std::map<std::string, std::unique_ptr<pollable_listener> > listeners_;
+ std::map<pollable*, std::unique_ptr<pollable_engine> > engines_;
+
+ std::condition_variable stopped_;
+ bool stopping_;
+ proton::error_condition stop_err_;
+ std::atomic<size_t> threads_;
+};
+
+// Base class for pollable file-descriptors. Manages epoll interaction,
+// subclasses implement virtual work() to do their serialized work.
+class pollable {
+ public:
+ pollable(int fd, int epoll_fd) : fd_(fd), epoll_fd_(epoll_fd), notified_(false), working_(false)
+ {
+ int flags = check(::fcntl(fd, F_GETFL, 0), "non-blocking");
+ check(::fcntl(fd, F_SETFL, flags | O_NONBLOCK), "non-blocking");
+ ::epoll_event ev = {};
+ ev.data.ptr = this;
+ ::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, fd_, &ev);
+ }
+
+ virtual ~pollable() {
+ ::epoll_event ev = {};
+ ::epoll_ctl(epoll_fd_, EPOLL_CTL_DEL, fd_, &ev); // Ignore errors.
+ }
+
+ bool do_work(uint32_t events) {
+ {
+ lock_guard g(lock_);
+ if (working_)
+ return true; // Another thread is already working.
+ working_ = true;
+ notified_ = false;
+ }
+ uint32_t new_events = work(events); // Serialized, outside the lock.
+ if (new_events) {
+ lock_guard g(lock_);
+ rearm(notified_ ? EPOLLIN|EPOLLOUT : new_events);
+ }
+ return new_events;
+ }
+
+ // Called from any thread to wake up the connection handler.
+ void notify() {
+ lock_guard g(lock_);
+ if (!notified_) {
+ notified_ = true;
+ if (!working_) // No worker thread, rearm now.
+ rearm(EPOLLIN|EPOLLOUT);
+ }
+ }
+
+ protected:
+
+ // Subclass implements work.
+ // Returns epoll events to re-enable or 0 if finished.
+ virtual uint32_t work(uint32_t events) = 0;
+
+ const unique_fd fd_;
+ const int epoll_fd_;
+
+ private:
+
+ void rearm(uint32_t events) {
+ epoll_event ev;
+ ev.data.ptr = this;
+ ev.events = EPOLLONESHOT | events;
+ check(::epoll_ctl(epoll_fd_, EPOLL_CTL_MOD, fd_, &ev), "re-arm epoll");
+ working_ = false;
+ }
+
+ std::mutex lock_;
+ bool notified_;
+ bool working_;
+};
+
+class epoll_event_loop : public proton::event_loop {
+ public:
+ typedef std::vector<std::function<void()> > jobs;
+
+ epoll_event_loop(pollable& p) : pollable_(p), closed_(false) {}
+
+ bool inject(std::function<void()> f) PN_CPP_OVERRIDE {
+ // Note this is an unbounded work queue.
+ // A resource-safe implementation should be bounded.
+ lock_guard g(lock_);
+ if (closed_)
+ return false;
+ jobs_.push_back(f);
+ pollable_.notify();
+ return true;
+ }
+
+ bool inject(proton::inject_handler& h) PN_CPP_OVERRIDE {
+ return inject(std::bind(&proton::inject_handler::on_inject, &h));
+ }
+
+ jobs pop_all() {
+ lock_guard g(lock_);
+ return std::move(jobs_);
+ }
+
+ void close() {
+ lock_guard g(lock_);
+ closed_ = true;
+ }
+
+ private:
+ std::mutex lock_;
+ pollable& pollable_;
+ jobs jobs_;
+ bool closed_;
+};
+
+// Handle epoll wakeups for a connection_engine.
+class pollable_engine : public pollable {
+ public:
+ pollable_engine(epoll_container& c, int fd, int epoll_fd) :
+ pollable(fd, epoll_fd),
+ loop_(new epoll_event_loop(*this)),
+ engine_(c, c.link_namer, loop_) {}
+
+ ~pollable_engine() {
+ loop_->close(); // No calls to notify() after this.
+ engine_.dispatch(); // Run any final events.
+ try { write(); } catch(...) {} // Write connection close if we can.
+ for (auto f : loop_->pop_all()) {// Run final queued work for side-effects.
+ try { f(); } catch(...) {}
+ }
+ }
+
+ uint32_t work(uint32_t events) {
+ try {
+ bool can_read = events & EPOLLIN, can_write = events && EPOLLOUT;
+ do {
+ can_write = can_write && write();
+ can_read = can_read && read();
+ for (auto f : loop_->pop_all()) // Run queued work
+ f();
+ engine_.dispatch();
+ } while (can_read || can_write);
+ return (engine_.read_buffer().size ? EPOLLIN:0) |
+ (engine_.write_buffer().size ? EPOLLOUT:0);
+ } catch (const std::exception& e) {
+ engine_.disconnected(proton::error_condition("exception", e.what()));
+ }
+ return 0; // Ending
+ }
+
+ proton::io::connection_engine& engine() { return engine_; }
+
+ private:
+
+ bool write() {
+ if (engine_.write_buffer().size) {
+ ssize_t n = ::write(fd_, engine_.write_buffer().data, engine_.write_buffer().size);
+ while (n == EINTR)
+ n = ::write(fd_, engine_.write_buffer().data, engine_.write_buffer().size);
+ if (n > 0) {
+ engine_.write_done(n);
+ return true;
+ } else if (errno != EAGAIN && errno != EWOULDBLOCK)
+ check(n, "write");
+ }
+ return false;
+ }
+
+ bool read() {
+ if (engine_.read_buffer().size) {
+ ssize_t n = ::read(fd_, engine_.read_buffer().data, engine_.read_buffer().size);
+ while (n == EINTR)
+ n = ::read(fd_, engine_.read_buffer().data, engine_.read_buffer().size);
+ if (n > 0) {
+ engine_.read_done(n);
+ return true;
+ }
+ else if (n == 0)
+ engine_.read_close();
+ else if (errno != EAGAIN && errno != EWOULDBLOCK)
+ check(n, "read");
+ }
+ return false;
+ }
+
+ // Lifecycle note: loop_ belongs to the proton::connection, which can live
+ // longer than the engine if the application holds a reference to it, we
+ // disconnect ourselves with loop_->close() in ~connection_engine()
+ epoll_event_loop* loop_;
+ proton::io::connection_engine engine_;
+};
+
+// A pollable listener fd that creates pollable_engine for incoming connections.
+class pollable_listener : public pollable {
+ public:
+ pollable_listener(
+ const std::string& addr,
+ proton::listen_handler& l,
+ int epoll_fd,
+ epoll_container& c
+ ) :
+ pollable(socket_listen(addr), epoll_fd),
+ addr_(addr),
+ container_(c),
+ listener_(l)
+ {}
+
+ uint32_t work(uint32_t events) {
+ if (events & EPOLLRDHUP) {
+ try { listener_.on_close(); } catch (...) {}
+ return 0;
+ }
+ try {
+ int accepted = check(::accept(fd_, NULL, 0), "accept");
+ container_.add_engine(listener_.on_accept(), accepted, true);
+ return EPOLLIN;
+ } catch (const std::exception& e) {
+ listener_.on_error(e.what());
+ return 0;
+ }
+ }
+
+ std::string addr() { return addr_; }
+
+ private:
+
+ static int socket_listen(const std::string& addr) {
+ std::string msg = "listen on "+addr;
+ unique_addrinfo ainfo(addr);
+ unique_fd fd(check(::socket(ainfo->ai_family, SOCK_STREAM, 0), msg));
+ int yes = 1;
+ check(::setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(yes)), msg);
+ check(::bind(fd, ainfo->ai_addr, ainfo->ai_addrlen), msg);
+ check(::listen(fd, 32), msg);
+ return fd.release();
+ }
+
+ std::string addr_;
+ std::function<proton::connection_options(const std::string&)> factory_;
+ epoll_container& container_;
+ proton::connection_options opts_;
+ proton::listen_handler& listener_;
+};
+
+
+epoll_container::epoll_container(const std::string& id)
+ : id_(id), epoll_fd_(check(epoll_create(1), "epoll_create")),
+ interrupt_fd_(check(eventfd(1, 0), "eventfd")),
+ stopping_(false), threads_(0)
+{}
+
+epoll_container::~epoll_container() {
+ try {
+ stop(proton::error_condition("exception", "container shut-down"));
+ wait();
+ } catch (...) {}
+}
+
+proton::connection epoll_container::add_engine(proton::connection_options opts, int fd, bool server)
+{
+ lock_guard g(lock_);
+ if (stopping_)
+ throw proton::error("container is stopping");
+ std::unique_ptr<pollable_engine> eng(new pollable_engine(*this, fd, epoll_fd_));
+ if (server)
+ eng->engine().accept(opts);
+ else
+ eng->engine().connect(opts);
+ proton::connection c = eng->engine().connection();
+ eng->notify();
+ engines_[eng.get()] = std::move(eng);
+ return c;
+}
+
+void epoll_container::erase(pollable* e) {
+ lock_guard g(lock_);
+ if (!engines_.erase(e)) {
+ pollable_listener* l = dynamic_cast<pollable_listener*>(e);
+ if (l)
+ listeners_.erase(l->addr());
+ }
+ idle_check(g);
+}
+
+void epoll_container::idle_check(const lock_guard&) {
+ if (stopping_ && engines_.empty() && listeners_.empty())
+ interrupt();
+}
+
+proton::returned<proton::connection> epoll_container::connect(
+ const std::string& addr, const proton::connection_options& opts)
+{
+ std::string msg = "connect to "+addr;
+ unique_addrinfo ainfo(addr);
+ unique_fd fd(check(::socket(ainfo->ai_family, SOCK_STREAM, 0), msg));
+ check(::connect(fd, ainfo->ai_addr, ainfo->ai_addrlen), msg);
+ return make_thread_safe(add_engine(opts, fd.release(), false));
+}
+
+proton::listener epoll_container::listen(const std::string& addr, proton::listen_handler& lh) {
+ lock_guard g(lock_);
+ if (stopping_)
+ throw proton::error("container is stopping");
+ auto& l = listeners_[addr];
+ try {
+ l.reset(new pollable_listener(addr, lh, epoll_fd_, *this));
+ l->notify();
+ return proton::listener(*this, addr);
+ } catch (const std::exception& e) {
+ lh.on_error(e.what());
+ lh.on_close();
+ throw;
+ }
+}
+
+void epoll_container::stop_listening(const std::string& addr) {
+ lock_guard g(lock_);
+ listeners_.erase(addr);
+ idle_check(g);
+}
+
+void epoll_container::run() {
+ ++threads_;
+ try {
+ epoll_event e;
+ while(true) {
+ check(::epoll_wait(epoll_fd_, &e, 1, -1), "epoll_wait");
+ pollable* p = reinterpret_cast<pollable*>(e.data.ptr);
+ if (!p)
+ break; // Interrupted
+ if (!p->do_work(e.events))
+ erase(p);
+ }
+ } catch (const std::exception& e) {
+ stop(proton::error_condition("exception", e.what()));
+ }
+ if (--threads_ == 0)
+ stopped_.notify_all();
+}
+
+void epoll_container::auto_stop(bool set) {
+ lock_guard g(lock_);
+ stopping_ = set;
+}
+
+void epoll_container::stop(const proton::error_condition& err) {
+ lock_guard g(lock_);
+ stop_err_ = err;
+ interrupt();
+}
+
+void epoll_container::wait() {
+ std::unique_lock<std::mutex> l(lock_);
+ stopped_.wait(l, [this]() { return this->threads_ == 0; } );
+ for (auto& eng : engines_)
+ eng.second->engine().disconnected(stop_err_);
+ listeners_.clear();
+ engines_.clear();
+}
+
+void epoll_container::interrupt() {
+ // Add an always-readable fd with 0 data and no ONESHOT to interrupt all threads.
+ epoll_event ev = {};
+ ev.events = EPOLLIN;
+ check(epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, interrupt_fd_, &ev), "interrupt");
+}
+
+}
+
+// This is the only public function.
+std::unique_ptr<proton::container> make_mt_container(const std::string& id) {
+ return std::unique_ptr<proton::container>(new epoll_container(id));
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[3/4] qpid-proton git commit: PROTON-1184: C++ merge APIs for single
and multi-threaded use.
Posted by ac...@apache.org.
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/examples/cpp/mt/epoll_controller.cpp
----------------------------------------------------------------------
diff --git a/examples/cpp/mt/epoll_controller.cpp b/examples/cpp/mt/epoll_controller.cpp
deleted file mode 100644
index 80aba0c..0000000
--- a/examples/cpp/mt/epoll_controller.cpp
+++ /dev/null
@@ -1,517 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-#include <proton/controller.hpp>
-#include <proton/work_queue.hpp>
-#include <proton/url.hpp>
-
-#include <proton/io/connection_engine.hpp>
-#include <proton/io/default_controller.hpp>
-
-#include <atomic>
-#include <memory>
-#include <mutex>
-#include <condition_variable>
-#include <thread>
-#include <set>
-#include <sstream>
-#include <system_error>
-
-// Linux native IO
-#include <assert.h>
-#include <errno.h>
-#include <fcntl.h>
-#include <netdb.h>
-#include <sys/epoll.h>
-#include <sys/eventfd.h>
-#include <unistd.h>
-
-// This is the only public function, the implementation is private.
-std::unique_ptr<proton::controller> make_epoll_controller();
-
-// Private implementation
-namespace {
-
-
-using lock_guard = std::lock_guard<std::mutex>;
-
-// Get string from errno
-std::string errno_str(const std::string& msg) {
- return std::system_error(errno, std::system_category(), msg).what();
-}
-
-// Throw proton::error(errno_str(msg)) if result < 0
-int check(int result, const std::string& msg) {
- if (result < 0)
- throw proton::error(errno_str(msg));
- return result;
-}
-
-// Wrapper for getaddrinfo() that cleans up in destructor.
-class unique_addrinfo {
- public:
- unique_addrinfo(const std::string& addr) : addrinfo_(0) {
- proton::url u(addr);
- int result = ::getaddrinfo(char_p(u.host()), char_p(u.port()), 0, &addrinfo_);
- if (result)
- throw proton::error(std::string("bad address: ") + gai_strerror(result));
- }
- ~unique_addrinfo() { if (addrinfo_) ::freeaddrinfo(addrinfo_); }
-
- ::addrinfo* operator->() const { return addrinfo_; }
-
- private:
- static const char* char_p(const std::string& s) { return s.empty() ? 0 : s.c_str(); }
- ::addrinfo *addrinfo_;
-};
-
-// File descriptor wrapper that calls ::close in destructor.
-class unique_fd {
- public:
- unique_fd(int fd) : fd_(fd) {}
- ~unique_fd() { if (fd_ >= 0) ::close(fd_); }
- operator int() const { return fd_; }
- int release() { int ret = fd_; fd_ = -1; return ret; }
-
- protected:
- int fd_;
-};
-
-class pollable;
-class pollable_engine;
-class pollable_listener;
-
-class epoll_controller : public proton::controller {
- public:
- epoll_controller();
- ~epoll_controller();
-
- // Implemenet the proton::controller interface
- void connect(const std::string& addr,
- proton::handler& h,
- const proton::connection_options& opts) override;
-
- void listen(const std::string& addr,
- std::function<proton::handler*(const std::string&)> factory,
- const proton::connection_options& opts) override;
-
- void stop_listening(const std::string& addr) override;
-
- void options(const proton::connection_options& opts) override;
- proton::connection_options options() override;
-
- void run() override;
-
- void stop_on_idle() override;
- void stop(const proton::error_condition& err) override;
- void wait() override;
-
- // Functions used internally.
-
- void add_engine(proton::handler* h, proton::connection_options opts, int fd);
- void erase(pollable*);
-
- private:
- void idle_check(const lock_guard&);
- void interrupt();
-
- const unique_fd epoll_fd_;
- const unique_fd interrupt_fd_;
-
- mutable std::mutex lock_;
-
- proton::connection_options options_;
- std::map<std::string, std::unique_ptr<pollable_listener> > listeners_;
- std::map<pollable*, std::unique_ptr<pollable_engine> > engines_;
-
- std::condition_variable stopped_;
- bool stopping_;
- proton::error_condition stop_err_;
- std::atomic<size_t> threads_;
-};
-
-// Base class for pollable file-descriptors. Manages epoll interaction,
-// subclasses implement virtual work() to do their serialized work.
-class pollable {
- public:
- pollable(int fd, int epoll_fd) : fd_(fd), epoll_fd_(epoll_fd), notified_(false), working_(false)
- {
- int flags = check(::fcntl(fd, F_GETFL, 0), "non-blocking");
- check(::fcntl(fd, F_SETFL, flags | O_NONBLOCK), "non-blocking");
- ::epoll_event ev = {};
- ev.data.ptr = this;
- ::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, fd_, &ev);
- }
-
- virtual ~pollable() {
- ::epoll_event ev = {};
- ::epoll_ctl(epoll_fd_, EPOLL_CTL_DEL, fd_, &ev); // Ignore errors.
- }
-
- bool do_work(uint32_t events) {
- {
- lock_guard g(lock_);
- if (working_)
- return true; // Another thread is already working.
- working_ = true;
- notified_ = false;
- }
- uint32_t new_events = work(events); // Serialized, outside the lock.
- if (new_events) {
- lock_guard g(lock_);
- rearm(notified_ ? EPOLLIN|EPOLLOUT : new_events);
- }
- return new_events;
- }
-
- // Called by work_queue to notify that there are jobs.
- void notify() {
- lock_guard g(lock_);
- if (!notified_) {
- notified_ = true;
- if (!working_) // No worker thread, rearm now.
- rearm(EPOLLIN|EPOLLOUT);
- }
- }
-
- protected:
-
- // Subclass implements work.
- // Returns epoll events to re-enable or 0 if finished.
- virtual uint32_t work(uint32_t events) = 0;
-
- const unique_fd fd_;
- const int epoll_fd_;
-
- private:
-
- void rearm(uint32_t events) {
- epoll_event ev;
- ev.data.ptr = this;
- ev.events = EPOLLONESHOT | events;
- check(::epoll_ctl(epoll_fd_, EPOLL_CTL_MOD, fd_, &ev), "re-arm epoll");
- working_ = false;
- }
-
- std::mutex lock_;
- bool notified_;
- bool working_;
-};
-
-class work_queue : public proton::work_queue {
- public:
- typedef std::vector<std::function<void()> > jobs;
-
- work_queue(pollable& p, proton::controller& c) :
- pollable_(p), closed_(false), controller_(c) {}
-
- bool push(std::function<void()> f) override {
- // Note this is an unbounded work queue.
- // A resource-safe implementation should be bounded.
- lock_guard g(lock_);
- if (closed_)
- return false;
- jobs_.push_back(f);
- pollable_.notify();
- return true;
- }
-
- jobs pop_all() {
- lock_guard g(lock_);
- return std::move(jobs_);
- }
-
- void close() {
- lock_guard g(lock_);
- closed_ = true;
- }
-
- proton::controller& controller() const override { return controller_; }
-
- private:
- std::mutex lock_;
- pollable& pollable_;
- jobs jobs_;
- bool closed_;
- proton::controller& controller_;
-};
-
-// Handle epoll wakeups for a connection_engine.
-class pollable_engine : public pollable {
- public:
-
- pollable_engine(
- proton::handler* h, proton::connection_options opts, epoll_controller& c,
- int fd, int epoll_fd
- ) : pollable(fd, epoll_fd),
- engine_(*h, opts),
- queue_(new work_queue(*this, c))
- {
- engine_.work_queue(queue_.get());
- }
-
- ~pollable_engine() {
- queue_->close(); // No calls to notify() after this.
- engine_.dispatch(); // Run any final events.
- try { write(); } catch(...) {} // Write connection close if we can.
- for (auto f : queue_->pop_all()) {// Run final queued work for side-effects.
- try { f(); } catch(...) {}
- }
- }
-
- uint32_t work(uint32_t events) {
- try {
- bool can_read = events & EPOLLIN, can_write = events && EPOLLOUT;
- do {
- can_write = can_write && write();
- can_read = can_read && read();
- for (auto f : queue_->pop_all()) // Run queued work
- f();
- engine_.dispatch();
- } while (can_read || can_write);
- return (engine_.read_buffer().size ? EPOLLIN:0) |
- (engine_.write_buffer().size ? EPOLLOUT:0);
- } catch (const std::exception& e) {
- close(proton::error_condition("exception", e.what()));
- }
- return 0; // Ending
- }
-
- void close(const proton::error_condition& err) {
- engine_.connection().close(err);
- }
-
- private:
-
- bool write() {
- if (engine_.write_buffer().size) {
- ssize_t n = ::write(fd_, engine_.write_buffer().data, engine_.write_buffer().size);
- while (n == EINTR)
- n = ::write(fd_, engine_.write_buffer().data, engine_.write_buffer().size);
- if (n > 0) {
- engine_.write_done(n);
- return true;
- } else if (errno != EAGAIN && errno != EWOULDBLOCK)
- check(n, "write");
- }
- return false;
- }
-
- bool read() {
- if (engine_.read_buffer().size) {
- ssize_t n = ::read(fd_, engine_.read_buffer().data, engine_.read_buffer().size);
- while (n == EINTR)
- n = ::read(fd_, engine_.read_buffer().data, engine_.read_buffer().size);
- if (n > 0) {
- engine_.read_done(n);
- return true;
- }
- else if (n == 0)
- engine_.read_close();
- else if (errno != EAGAIN && errno != EWOULDBLOCK)
- check(n, "read");
- }
- return false;
- }
-
- proton::io::connection_engine engine_;
- std::shared_ptr<work_queue> queue_;
-};
-
-// A pollable listener fd that creates pollable_engine for incoming connections.
-class pollable_listener : public pollable {
- public:
- pollable_listener(
- const std::string& addr,
- std::function<proton::handler*(const std::string&)> factory,
- int epoll_fd,
- epoll_controller& c,
- const proton::connection_options& opts
- ) :
- pollable(listen(addr), epoll_fd),
- addr_(addr),
- factory_(factory),
- controller_(c),
- opts_(opts)
- {}
-
- uint32_t work(uint32_t events) {
- if (events & EPOLLRDHUP)
- return 0;
- int accepted = check(::accept(fd_, NULL, 0), "accept");
- controller_.add_engine(factory_(addr_), opts_, accepted);
- return EPOLLIN;
- }
-
- std::string addr() { return addr_; }
-
- private:
-
- int listen(const std::string& addr) {
- std::string msg = "listen on "+addr;
- unique_addrinfo ainfo(addr);
- unique_fd fd(check(::socket(ainfo->ai_family, SOCK_STREAM, 0), msg));
- int yes = 1;
- check(::setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(yes)), msg);
- check(::bind(fd, ainfo->ai_addr, ainfo->ai_addrlen), msg);
- check(::listen(fd, 32), msg);
- return fd.release();
- }
-
- std::string addr_;
- std::function<proton::handler*(const std::string&)> factory_;
- epoll_controller& controller_;
- proton::connection_options opts_;
-};
-
-
-epoll_controller::epoll_controller()
- : epoll_fd_(check(epoll_create(1), "epoll_create")),
- interrupt_fd_(check(eventfd(1, 0), "eventfd")),
- stopping_(false), threads_(0)
-{}
-
-epoll_controller::~epoll_controller() {
- try {
- stop(proton::error_condition("exception", "controller shut-down"));
- wait();
- } catch (...) {}
-}
-
-void epoll_controller::add_engine(proton::handler* h, proton::connection_options opts, int fd) {
- lock_guard g(lock_);
- if (stopping_)
- throw proton::error("controller is stopping");
- std::unique_ptr<pollable_engine> e(new pollable_engine(h, opts, *this, fd, epoll_fd_));
- e->notify();
- engines_[e.get()] = std::move(e);
-}
-
-void epoll_controller::erase(pollable* e) {
- lock_guard g(lock_);
- if (!engines_.erase(e)) {
- pollable_listener* l = dynamic_cast<pollable_listener*>(e);
- if (l)
- listeners_.erase(l->addr());
- }
- idle_check(g);
-}
-
-void epoll_controller::idle_check(const lock_guard&) {
- if (stopping_ && engines_.empty() && listeners_.empty())
- interrupt();
-}
-
-void epoll_controller::connect(const std::string& addr,
- proton::handler& h,
- const proton::connection_options& opts)
-{
- std::string msg = "connect to "+addr;
- unique_addrinfo ainfo(addr);
- unique_fd fd(check(::socket(ainfo->ai_family, SOCK_STREAM, 0), msg));
- check(::connect(fd, ainfo->ai_addr, ainfo->ai_addrlen), msg);
- add_engine(&h, options().update(opts), fd.release());
-}
-
-void epoll_controller::listen(const std::string& addr,
- std::function<proton::handler*(const std::string&)> factory,
- const proton::connection_options& opts)
-{
- lock_guard g(lock_);
- if (!factory)
- throw proton::error("null function to listen on "+addr);
- if (stopping_)
- throw proton::error("controller is stopping");
- auto& l = listeners_[addr];
- l.reset(new pollable_listener(addr, factory, epoll_fd_, *this, options_.update(opts)));
- l->notify();
-}
-
-void epoll_controller::stop_listening(const std::string& addr) {
- lock_guard g(lock_);
- listeners_.erase(addr);
- idle_check(g);
-}
-
-void epoll_controller::options(const proton::connection_options& opts) {
- lock_guard g(lock_);
- options_ = opts;
-}
-
-proton::connection_options epoll_controller::options() {
- lock_guard g(lock_);
- return options_;
-}
-
-void epoll_controller::run() {
- ++threads_;
- try {
- epoll_event e;
- while(true) {
- check(::epoll_wait(epoll_fd_, &e, 1, -1), "epoll_wait");
- pollable* p = reinterpret_cast<pollable*>(e.data.ptr);
- if (!p)
- break; // Interrupted
- if (!p->do_work(e.events))
- erase(p);
- }
- } catch (const std::exception& e) {
- stop(proton::error_condition("exception", e.what()));
- }
- if (--threads_ == 0)
- stopped_.notify_all();
-}
-
-void epoll_controller::stop_on_idle() {
- lock_guard g(lock_);
- stopping_ = true;
- idle_check(g);
-}
-
-void epoll_controller::stop(const proton::error_condition& err) {
- lock_guard g(lock_);
- stop_err_ = err;
- interrupt();
-}
-
-void epoll_controller::wait() {
- std::unique_lock<std::mutex> l(lock_);
- stopped_.wait(l, [this]() { return this->threads_ == 0; } );
- for (auto& eng : engines_)
- eng.second->close(stop_err_);
- listeners_.clear();
- engines_.clear();
-}
-
-void epoll_controller::interrupt() {
- // Add an always-readable fd with 0 data and no ONESHOT to interrupt all threads.
- epoll_event ev = {};
- ev.events = EPOLLIN;
- check(epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, interrupt_fd_, &ev), "interrupt");
-}
-
-// Register make_epoll_controller() as proton::controller::create().
-proton::io::default_controller instance(make_epoll_controller);
-
-}
-
-// This is the only public function.
-std::unique_ptr<proton::controller> make_epoll_controller() {
- return std::unique_ptr<proton::controller>(new epoll_controller());
-}
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/examples/cpp/mt/mt_container.hpp
----------------------------------------------------------------------
diff --git a/examples/cpp/mt/mt_container.hpp b/examples/cpp/mt/mt_container.hpp
new file mode 100644
index 0000000..164fe72
--- /dev/null
+++ b/examples/cpp/mt/mt_container.hpp
@@ -0,0 +1,29 @@
+#ifndef MT_MT_CONTROLLER_HPP
+#define MT_MT_CONTROLLER_HPP
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include <proton/default_container.hpp>
+#include <memory>
+
+// Defined in whichever MT container implementation we are linked with.
+std::unique_ptr<proton::container> make_mt_container(const std::string& id);
+
+#endif // MT_MT_DEFAULT_CONTAINER.HPP
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/examples/cpp/queue_browser.cpp
----------------------------------------------------------------------
diff --git a/examples/cpp/queue_browser.cpp b/examples/cpp/queue_browser.cpp
index e4fc29e..9f52a9d 100644
--- a/examples/cpp/queue_browser.cpp
+++ b/examples/cpp/queue_browser.cpp
@@ -20,7 +20,7 @@
*/
#include "proton/connection.hpp"
-#include "proton/container.hpp"
+#include "proton/default_container.hpp"
#include "proton/delivery.hpp"
#include "proton/handler.hpp"
#include "proton/receiver_options.hpp"
@@ -29,7 +29,7 @@
#include <iostream>
-#include "fake_cpp11.hpp"
+#include <proton/config.hpp>
using proton::source_options;
@@ -40,13 +40,13 @@ class browser : public proton::handler {
public:
browser(const std::string& u) : url(u) {}
- void on_container_start(proton::container &c) override {
+ void on_container_start(proton::container &c) PN_CPP_OVERRIDE {
proton::connection conn = c.connect(url);
source_options browsing = source_options().distribution_mode(proton::source::COPY);
conn.open_receiver(url.path(), proton::receiver_options().source(browsing));
}
- void on_message(proton::delivery &, proton::message &m) override {
+ void on_message(proton::delivery &, proton::message &m) PN_CPP_OVERRIDE {
std::cout << m.body() << std::endl;
}
};
@@ -56,7 +56,7 @@ int main(int argc, char **argv) {
std::string url = argc > 1 ? argv[1] : "127.0.0.1:5672/examples";
browser b(url);
- proton::container(b).run();
+ proton::default_container(b).run();
return 0;
} catch (const std::exception& e) {
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/examples/cpp/selected_recv.cpp
----------------------------------------------------------------------
diff --git a/examples/cpp/selected_recv.cpp b/examples/cpp/selected_recv.cpp
index 5bb65ff..3c57905 100644
--- a/examples/cpp/selected_recv.cpp
+++ b/examples/cpp/selected_recv.cpp
@@ -20,7 +20,7 @@
*/
#include "proton/connection.hpp"
-#include "proton/container.hpp"
+#include "proton/default_container.hpp"
#include "proton/handler.hpp"
#include "proton/receiver_options.hpp"
#include "proton/source_options.hpp"
@@ -28,7 +28,7 @@
#include <iostream>
-#include "fake_cpp11.hpp"
+#include <proton/config.hpp>
namespace {
@@ -60,14 +60,14 @@ class selected_recv : public proton::handler {
public:
selected_recv(const std::string& u) : url(u) {}
- void on_container_start(proton::container &c) override {
+ void on_container_start(proton::container &c) PN_CPP_OVERRIDE {
proton::source_options opts;
set_filter(opts, "colour = 'green'");
proton::connection conn = c.connect(url);
conn.open_receiver(url.path(), proton::receiver_options().source(opts));
}
- void on_message(proton::delivery &, proton::message &m) override {
+ void on_message(proton::delivery &, proton::message &m) PN_CPP_OVERRIDE {
std::cout << m.body() << std::endl;
}
};
@@ -77,7 +77,7 @@ int main(int argc, char **argv) {
std::string url = argc > 1 ? argv[1] : "127.0.0.1:5672/examples";
selected_recv recv(url);
- proton::container(recv).run();
+ proton::default_container(recv).run();
return 0;
} catch (const std::exception& e) {
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/examples/cpp/server.cpp
----------------------------------------------------------------------
diff --git a/examples/cpp/server.cpp b/examples/cpp/server.cpp
index 6f2f150..332fb1b 100644
--- a/examples/cpp/server.cpp
+++ b/examples/cpp/server.cpp
@@ -22,7 +22,7 @@
#include "options.hpp"
#include "proton/connection.hpp"
-#include "proton/container.hpp"
+#include "proton/default_container.hpp"
#include "proton/handler.hpp"
#include "proton/tracker.hpp"
#include "proton/url.hpp"
@@ -32,7 +32,7 @@
#include <string>
#include <cctype>
-#include "fake_cpp11.hpp"
+#include <proton/config.hpp>
class server : public proton::handler {
private:
@@ -44,7 +44,7 @@ class server : public proton::handler {
public:
server(const std::string &u) : url(u) {}
- void on_container_start(proton::container &c) override {
+ void on_container_start(proton::container &c) PN_CPP_OVERRIDE {
connection = c.connect(url);
connection.open_receiver(url.path());
@@ -59,7 +59,7 @@ class server : public proton::handler {
return uc;
}
- void on_message(proton::delivery &, proton::message &m) override {
+ void on_message(proton::delivery &, proton::message &m) PN_CPP_OVERRIDE {
std::cout << "Received " << m.body() << std::endl;
std::string reply_to = m.reply_to();
@@ -87,7 +87,7 @@ int main(int argc, char **argv) {
opts.parse();
server srv(address);
- proton::container(srv).run();
+ proton::default_container(srv).run();
return 0;
} catch (const example::bad_option& e) {
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/examples/cpp/server_direct.cpp
----------------------------------------------------------------------
diff --git a/examples/cpp/server_direct.cpp b/examples/cpp/server_direct.cpp
index fc7aa67..74dcfd9 100644
--- a/examples/cpp/server_direct.cpp
+++ b/examples/cpp/server_direct.cpp
@@ -21,8 +21,7 @@
#include "options.hpp"
-#include "proton/acceptor.hpp"
-#include "proton/container.hpp"
+#include "proton/default_container.hpp"
#include "proton/handler.hpp"
#include "proton/sender.hpp"
#include "proton/source_options.hpp"
@@ -34,7 +33,7 @@
#include <sstream>
#include <cctype>
-#include "fake_cpp11.hpp"
+#include <proton/config.hpp>
class server : public proton::handler {
private:
@@ -46,7 +45,7 @@ class server : public proton::handler {
public:
server(const std::string &u) : url(u), address_counter(0) {}
- void on_container_start(proton::container &c) override {
+ void on_container_start(proton::container &c) PN_CPP_OVERRIDE {
c.listen(url);
std::cout << "server listening on " << url << std::endl;
}
@@ -68,7 +67,7 @@ class server : public proton::handler {
return addr.str();
}
- void on_sender_open(proton::sender &sender) override {
+ void on_sender_open(proton::sender &sender) PN_CPP_OVERRIDE {
if (sender.source().dynamic()) {
std::string addr = generate_address();
sender.open(proton::sender_options().source(proton::source_options().address(addr)));
@@ -76,7 +75,7 @@ class server : public proton::handler {
}
}
- void on_message(proton::delivery &, proton::message &m) override {
+ void on_message(proton::delivery &, proton::message &m) PN_CPP_OVERRIDE {
std::cout << "Received " << m.body() << std::endl;
std::string reply_to = m.reply_to();
@@ -107,7 +106,7 @@ int main(int argc, char **argv) {
opts.parse();
server srv(address);
- proton::container(srv).run();
+ proton::default_container(srv).run();
return 0;
} catch (const example::bad_option& e) {
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/examples/cpp/simple_recv.cpp
----------------------------------------------------------------------
diff --git a/examples/cpp/simple_recv.cpp b/examples/cpp/simple_recv.cpp
index 3eadf32..b1578ec 100644
--- a/examples/cpp/simple_recv.cpp
+++ b/examples/cpp/simple_recv.cpp
@@ -22,7 +22,7 @@
#include "options.hpp"
#include "proton/connection.hpp"
-#include "proton/container.hpp"
+#include "proton/default_container.hpp"
#include "proton/delivery.hpp"
#include "proton/handler.hpp"
#include "proton/link.hpp"
@@ -32,7 +32,7 @@
#include <iostream>
#include <map>
-#include "fake_cpp11.hpp"
+#include <proton/config.hpp>
class simple_recv : public proton::handler {
private:
@@ -44,12 +44,12 @@ class simple_recv : public proton::handler {
public:
simple_recv(const std::string &s, int c) : url(s), expected(c), received(0) {}
- void on_container_start(proton::container &c) override {
+ void on_container_start(proton::container &c) PN_CPP_OVERRIDE {
receiver = c.open_receiver(url);
std::cout << "simple_recv listening on " << url << std::endl;
}
- void on_message(proton::delivery &d, proton::message &msg) override {
+ void on_message(proton::delivery &d, proton::message &msg) PN_CPP_OVERRIDE {
if (msg.id().get<uint64_t>() < received) {
return; // Ignore duplicate
}
@@ -79,7 +79,7 @@ int main(int argc, char **argv) {
opts.parse();
simple_recv recv(address, message_count);
- proton::container(recv).run();
+ proton::default_container(recv).run();
return 0;
} catch (const example::bad_option& e) {
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/examples/cpp/simple_send.cpp
----------------------------------------------------------------------
diff --git a/examples/cpp/simple_send.cpp b/examples/cpp/simple_send.cpp
index 594b7d6..921bb95 100644
--- a/examples/cpp/simple_send.cpp
+++ b/examples/cpp/simple_send.cpp
@@ -22,7 +22,7 @@
#include "options.hpp"
#include "proton/connection.hpp"
-#include "proton/container.hpp"
+#include "proton/default_container.hpp"
#include "proton/handler.hpp"
#include "proton/tracker.hpp"
#include "proton/value.hpp"
@@ -30,7 +30,7 @@
#include <iostream>
#include <map>
-#include "fake_cpp11.hpp"
+#include <proton/config.hpp>
class simple_send : public proton::handler {
private:
@@ -43,11 +43,11 @@ class simple_send : public proton::handler {
public:
simple_send(const std::string &s, int c) : url(s), sent(0), confirmed(0), total(c) {}
- void on_container_start(proton::container &c) override {
+ void on_container_start(proton::container &c) PN_CPP_OVERRIDE {
sender = c.open_sender(url);
}
- void on_sendable(proton::sender &s) override {
+ void on_sendable(proton::sender &s) PN_CPP_OVERRIDE {
while (s.credit() && sent < total) {
proton::message msg;
std::map<std::string, int> m;
@@ -61,7 +61,7 @@ class simple_send : public proton::handler {
}
}
- void on_tracker_accept(proton::tracker &t) override {
+ void on_tracker_accept(proton::tracker &t) PN_CPP_OVERRIDE {
confirmed++;
if (confirmed == total) {
@@ -70,7 +70,7 @@ class simple_send : public proton::handler {
}
}
- void on_transport_close(proton::transport &) override {
+ void on_transport_close(proton::transport &) PN_CPP_OVERRIDE {
sent = confirmed;
}
};
@@ -87,7 +87,7 @@ int main(int argc, char **argv) {
opts.parse();
simple_send send(address, message_count);
- proton::container(send).run();
+ proton::default_container(send).run();
return 0;
} catch (const example::bad_option& e) {
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/examples/cpp/ssl.cpp
----------------------------------------------------------------------
diff --git a/examples/cpp/ssl.cpp b/examples/cpp/ssl.cpp
index f4e724d..379c95b 100644
--- a/examples/cpp/ssl.cpp
+++ b/examples/cpp/ssl.cpp
@@ -19,10 +19,9 @@
*
*/
-#include "proton/acceptor.hpp"
#include "proton/connection_options.hpp"
#include "proton/connection.hpp"
-#include "proton/container.hpp"
+#include "proton/default_container.hpp"
#include "proton/handler.hpp"
#include "proton/ssl.hpp"
#include "proton/tracker.hpp"
@@ -30,7 +29,7 @@
#include <iostream>
-#include "fake_cpp11.hpp"
+#include <proton/config.hpp>
using proton::connection_options;
using proton::ssl_client_options;
@@ -46,15 +45,15 @@ std::string find_CN(const std::string &);
struct server_handler : public proton::handler {
- proton::acceptor acceptor;
+ std::string url;
- void on_connection_open(proton::connection &c) override {
+ void on_connection_open(proton::connection &c) PN_CPP_OVERRIDE {
std::cout << "Inbound server connection connected via SSL. Protocol: " <<
c.transport().ssl().protocol() << std::endl;
- acceptor.close();
+ c.container().stop_listening(url);
}
- void on_message(proton::delivery &, proton::message &m) override {
+ void on_message(proton::delivery &, proton::message &m) PN_CPP_OVERRIDE {
std::cout << m.body() << std::endl;
}
};
@@ -68,12 +67,12 @@ class hello_world_direct : public proton::handler {
public:
hello_world_direct(const std::string& u) : url(u) {}
- void on_container_start(proton::container &c) override {
+ void on_container_start(proton::container &c) PN_CPP_OVERRIDE {
// Configure listener. Details vary by platform.
ssl_certificate server_cert = platform_certificate("tserver", "tserverpw");
ssl_server_options ssl_srv(server_cert);
connection_options server_opts;
- server_opts.ssl_server_options(ssl_srv).handler(&s_handler);
+ server_opts.ssl_server_options(ssl_srv).handler(s_handler);
c.server_connection_options(server_opts);
// Configure client with a Certificate Authority database populated with the server's self signed certificate.
@@ -84,24 +83,25 @@ class hello_world_direct : public proton::handler {
client_opts.ssl_client_options(ssl_cli);
c.client_connection_options(client_opts);
- s_handler.acceptor = c.listen(url);
+ s_handler.url = url;
+ c.listen(url);
c.open_sender(url);
}
- void on_connection_open(proton::connection &c) override {
+ void on_connection_open(proton::connection &c) PN_CPP_OVERRIDE {
std::string subject = c.transport().ssl().remote_subject();
std::cout << "Outgoing client connection connected via SSL. Server certificate identity " <<
find_CN(subject) << std::endl;
}
- void on_sendable(proton::sender &s) override {
+ void on_sendable(proton::sender &s) PN_CPP_OVERRIDE {
proton::message m;
m.body("Hello World!");
s.send(m);
s.close();
}
- void on_tracker_accept(proton::tracker &t) override {
+ void on_tracker_accept(proton::tracker &t) PN_CPP_OVERRIDE {
// All done.
t.connection().close();
}
@@ -122,7 +122,7 @@ int main(int argc, char **argv) {
else cert_directory = "ssl_certs/";
hello_world_direct hwd(url);
- proton::container(hwd).run();
+ proton::default_container(hwd).run();
return 0;
} catch (const std::exception& e) {
std::cerr << e.what() << std::endl;
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/examples/cpp/ssl_client_cert.cpp
----------------------------------------------------------------------
diff --git a/examples/cpp/ssl_client_cert.cpp b/examples/cpp/ssl_client_cert.cpp
index 0708d2c..b16aad4 100644
--- a/examples/cpp/ssl_client_cert.cpp
+++ b/examples/cpp/ssl_client_cert.cpp
@@ -19,10 +19,9 @@
*
*/
-#include "proton/acceptor.hpp"
#include "proton/connection.hpp"
#include "proton/connection_options.hpp"
-#include "proton/container.hpp"
+#include "proton/default_container.hpp"
#include "proton/handler.hpp"
#include "proton/sasl.hpp"
#include "proton/ssl.hpp"
@@ -31,7 +30,7 @@
#include <iostream>
-#include "fake_cpp11.hpp"
+#include <proton/config.hpp>
using proton::connection_options;
using proton::ssl_client_options;
@@ -48,9 +47,9 @@ static std::string find_CN(const std::string &);
struct server_handler : public proton::handler {
- proton::acceptor inbound_listener;
+ proton::listener listener;
- void on_connection_open(proton::connection &c) override {
+ void on_connection_open(proton::connection &c) PN_CPP_OVERRIDE {
std::cout << "Inbound server connection connected via SSL. Protocol: " <<
c.transport().ssl().protocol() << std::endl;
if (c.transport().sasl().outcome() == sasl::OK) {
@@ -61,10 +60,10 @@ struct server_handler : public proton::handler {
std::cout << "Inbound client authentication failed" <<std::endl;
c.close();
}
- inbound_listener.close();
+ listener.stop();
}
- void on_message(proton::delivery &, proton::message &m) override {
+ void on_message(proton::delivery &, proton::message &m) PN_CPP_OVERRIDE {
std::cout << m.body() << std::endl;
}
};
@@ -78,14 +77,14 @@ class hello_world_direct : public proton::handler {
public:
hello_world_direct(const std::string& u) : url(u) {}
- void on_container_start(proton::container &c) override {
+ void on_container_start(proton::container &c) PN_CPP_OVERRIDE {
// Configure listener. Details vary by platform.
ssl_certificate server_cert = platform_certificate("tserver", "tserverpw");
std::string client_CA = platform_CA("tclient");
// Specify an SSL domain with CA's for client certificate verification.
ssl_server_options srv_ssl(server_cert, client_CA);
connection_options server_opts;
- server_opts.ssl_server_options(srv_ssl).handler(&s_handler);
+ server_opts.ssl_server_options(srv_ssl).handler(s_handler);
server_opts.sasl_allowed_mechs("EXTERNAL");
c.server_connection_options(server_opts);
@@ -99,24 +98,24 @@ class hello_world_direct : public proton::handler {
client_opts.ssl_client_options(ssl_cli).sasl_allowed_mechs("EXTERNAL");
c.client_connection_options(client_opts);
- s_handler.inbound_listener = c.listen(url);
+ s_handler.listener = c.listen(url);
c.open_sender(url);
}
- void on_connection_open(proton::connection &c) override {
+ void on_connection_open(proton::connection &c) PN_CPP_OVERRIDE {
std::string subject = c.transport().ssl().remote_subject();
std::cout << "Outgoing client connection connected via SSL. Server certificate identity " <<
find_CN(subject) << std::endl;
}
- void on_sendable(proton::sender &s) override {
+ void on_sendable(proton::sender &s) PN_CPP_OVERRIDE {
proton::message m;
m.body("Hello World!");
s.send(m);
s.close();
}
- void on_tracker_accept(proton::tracker &t) override {
+ void on_tracker_accept(proton::tracker &t) PN_CPP_OVERRIDE {
// All done.
t.connection().close();
}
@@ -137,7 +136,7 @@ int main(int argc, char **argv) {
else cert_directory = "ssl_certs/";
hello_world_direct hwd(url);
- proton::container(hwd).run();
+ proton::default_container(hwd).run();
return 0;
} catch (const std::exception& e) {
std::cerr << e.what() << std::endl;
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/examples/cpp/tutorial.dox
----------------------------------------------------------------------
diff --git a/examples/cpp/tutorial.dox b/examples/cpp/tutorial.dox
index b8383f7..ab46a1d 100644
--- a/examples/cpp/tutorial.dox
+++ b/examples/cpp/tutorial.dox
@@ -268,7 +268,7 @@ implicitly a connection), we listen for incoming connections.
\until }
When we have received all the expected messages, we then stop listening for
-incoming connections by closing the acceptor object.
+incoming connections by calling `container::stop_listening()`
\skip on_message
\until }
@@ -292,8 +292,8 @@ link, we listen for incoming connections.
\skip on_start
\until }
-When we have received confirmation of all the messages we sent, we can
-close the acceptor in order to exit.
+When we have received confirmation of all the messages we sent, we call
+`container::stop_listening()` to exit.
\skip on_delivery_accept
\until }
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/CMakeLists.txt b/proton-c/bindings/cpp/CMakeLists.txt
index 26b8dd4..e18d4b9 100644
--- a/proton-c/bindings/cpp/CMakeLists.txt
+++ b/proton-c/bindings/cpp/CMakeLists.txt
@@ -27,7 +27,6 @@ include_directories(
add_definitions(${CXX_WARNING_FLAGS})
set(qpid-proton-cpp-source
- ${qpid-proton-mt-source}
src/acceptor.cpp
src/binary.cpp
src/byte_array.cpp
@@ -46,10 +45,11 @@ set(qpid-proton-cpp-source
src/endpoint.cpp
src/error.cpp
src/error_condition.cpp
+ src/event_loop.cpp
src/handler.cpp
- src/id_generator.cpp
src/io/connection_engine.cpp
src/link.cpp
+ src/listener.cpp
src/message.cpp
src/messaging_adapter.cpp
src/node_options.cpp
@@ -75,20 +75,14 @@ set(qpid-proton-cpp-source
src/terminus.cpp
src/timestamp.cpp
src/tracker.cpp
- src/transport.cpp
src/transfer.cpp
+ src/transport.cpp
src/type_id.cpp
src/url.cpp
src/uuid.cpp
src/value.cpp
)
-if (HAS_CPP11)
- message(STATUS "Enable C++11 extensions")
- list(APPEND qpid-proton-cpp-source src/controller.cpp)
-endif()
-
-
set_source_files_properties (
${qpid-proton-cpp-source}
PROPERTIES
@@ -181,6 +175,7 @@ endmacro(add_cpp_test)
add_cpp_test(codec_test)
add_cpp_test(engine_test)
+add_cpp_test(thread_safe_test)
add_cpp_test(interop_test ${CMAKE_SOURCE_DIR}/tests)
add_cpp_test(message_test)
add_cpp_test(scalar_test)
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/docs/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/docs/CMakeLists.txt b/proton-c/bindings/cpp/docs/CMakeLists.txt
index c5ae4e5..d512d15 100644
--- a/proton-c/bindings/cpp/docs/CMakeLists.txt
+++ b/proton-c/bindings/cpp/docs/CMakeLists.txt
@@ -23,15 +23,19 @@ if (DOXYGEN_FOUND)
configure_file (
${CMAKE_CURRENT_SOURCE_DIR}/user.doxygen.in
${CMAKE_CURRENT_BINARY_DIR}/user.doxygen)
+
+ file(GLOB_RECURSE headers ../include/proton/*.hpp)
add_custom_target (docs-cpp
COMMAND ${CMAKE_COMMAND} -E remove_directory html # get rid of old files
- COMMAND ${DOXYGEN_EXECUTABLE} user.doxygen)
+ COMMAND ${DOXYGEN_EXECUTABLE} user.doxygen
+ DEPENDS ${headers}
+ )
add_dependencies (docs docs-cpp)
# HTML files are generated to ./html - put those in the install.
install (DIRECTORY "${CMAKE_CURRENT_BINARY_DIR}/html/"
- DESTINATION "${PROTON_SHARE}/docs/api-cpp"
- COMPONENT documentation
- OPTIONAL)
+ DESTINATION "${PROTON_SHARE}/docs/api-cpp"
+ COMPONENT documentation
+ OPTIONAL)
endif (DOXYGEN_FOUND)
set_directory_properties(PROPERTIES ADDITIONAL_MAKE_CLEAN_FILES html)
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/docs/mainpage.md
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/docs/mainpage.md b/proton-c/bindings/cpp/docs/mainpage.md
index dffb3ed..2ba2eaf 100644
--- a/proton-c/bindings/cpp/docs/mainpage.md
+++ b/proton-c/bindings/cpp/docs/mainpage.md
@@ -106,27 +106,15 @@ can be re-used if you change your approach.
### %proton::container - easy single-threaded applications
-`proton::container` is an easy way to get started with single threaded
-applications. It manages the low-level socket connection and polling for you so
-you can write portable applications using just the @ref proton API. You an use
+`proton::container` is the top level object in a proton application. Use
proton::connection::connect() and proton::container::listen() to create
-connections. The container polls multiple connections and calls protocol eventsa
+connections. The container polls multiple connections and calls protocol events
on your `proton::handler` sub-classes.
-The container has two limitations:
-- it is not thread-safe, all container work must happen in a single thread.
-- it provides portable IO handling for you but cannot be used for different types of IO.
+The default container implementation is created by `proton::new_default_container()`.
-### %proton::mt::controller - multi-threaded applications
-
-The proton::controller is similar to the proton::container but it can process
-connections in multiple threads concurrently. It uses the `proton::handler` like
-the container. If you follow the recommended model you can re-use statefull,
-per-connection, handlers with the controller. See @ref mt_page and the examples
-@ref mt/broker.cpp and @ref mt/epoll\_controller.cpp
-
-Default controller IO implementations are provided but you can also implement
-your own controller using the proton::io::connection_engine, read on...
+You can implement your own container to integrate proton with arbitrary your own
+container using the proton::io::connection_engine.
### %proton::io::connection_engine - integrating with foreign IO
@@ -146,5 +134,5 @@ framework or library, memory-based streams or any other source/sink for byte
stream data.
You can also use the engine to build a custom implementation of
-proton::mt::controller and proton::mt::work\_queue so portable proton
-applications using the controller can run without modification on your platform.
+proton::container and so portable proton applications can run without
+modification on your platform.
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/docs/mt_page.md
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/docs/mt_page.md b/proton-c/bindings/cpp/docs/mt_page.md
index bed53e9..a1ac849 100644
--- a/proton-c/bindings/cpp/docs/mt_page.md
+++ b/proton-c/bindings/cpp/docs/mt_page.md
@@ -1,5 +1,7 @@
# Multi-threaded proton {#mt_page}
+<!-- FIXME aconway 2016-05-04: doc -->
+
Most classes in namespace @ref proton are not thread-safe. Objects associated
with a single connection *must not* be used concurrently. However objects
associated with *different* connections *can* be used concurrently in separate
@@ -8,7 +10,7 @@ threads.
The recommended way to use proton multi-threaded is to *serialize* the work for
each connection but allow different connections to be processed concurrently.
-proton::controller allows you to manage connections in a multi-threaded way. You
+proton::container allows you to manage connections in a multi-threaded way. You
supply a proton::handler for each connection. Proton will ensure that the
`proton::handler::on_*()` functions are never called concurrently so
per-connection handlers do not need a lock even if they have state.
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/include/proton/acceptor.hpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/include/proton/acceptor.hpp b/proton-c/bindings/cpp/include/proton/acceptor.hpp
deleted file mode 100644
index b0e11ed..0000000
--- a/proton-c/bindings/cpp/include/proton/acceptor.hpp
+++ /dev/null
@@ -1,61 +0,0 @@
-#ifndef PROTON_CPP_ACCEPTOR_H
-#define PROTON_CPP_ACCEPTOR_H
-
-/*
- *
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- *
- */
-
-#include <proton/reactor.h>
-#include <proton/export.hpp>
-#include <proton/object.hpp>
-
-struct pn_connection_t;
-
-namespace proton {
-
-/// A context for accepting inbound connections.
-///
-/// @see container::listen
-class acceptor : public internal::object<pn_acceptor_t> {
- /// @cond INTERNAL
- acceptor(pn_acceptor_t* a) : internal::object<pn_acceptor_t>(a) {}
- /// @endcond
-
- public:
- acceptor() : internal::object<pn_acceptor_t>(0) {}
-
- /// Close the acceptor.
- PN_CPP_EXTERN void close();
-
- /// Return the current set of connection options applied to
- /// inbound connectons by the acceptor.
- ///
- /// Note that changes made to the connection options only affect
- /// connections accepted after this call returns.
- PN_CPP_EXTERN class connection_options &connection_options();
-
- /// @cond INTERNAL
- friend class internal::factory<acceptor>;
- /// @endcond
-};
-
-}
-
-#endif // PROTON_CPP_ACCEPTOR_H
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/include/proton/config.hpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/include/proton/config.hpp b/proton-c/bindings/cpp/include/proton/config.hpp
index e6cb8ae..f6b10d6 100644
--- a/proton-c/bindings/cpp/include/proton/config.hpp
+++ b/proton-c/bindings/cpp/include/proton/config.hpp
@@ -59,6 +59,12 @@
#define PN_CPP_HAS_OVERRIDE PN_CPP_HAS_CPP11
#endif
+#if PN_CPP_HAS_OVERRIDE
+#define PN_CPP_OVERRIDE override
+#else
+#define PN_CPP_OVERRIDE
+#endif
+
#ifndef PN_CPP_HAS_EXPLICIT_CONVERSIONS
#define PN_CPP_HAS_EXPLICIT_CONVERSIONS PN_CPP_HAS_CPP11
#endif
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/include/proton/connection.hpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/include/proton/connection.hpp b/proton-c/bindings/cpp/include/proton/connection.hpp
index fd7bffa..1851625 100644
--- a/proton-c/bindings/cpp/include/proton/connection.hpp
+++ b/proton-c/bindings/cpp/include/proton/connection.hpp
@@ -27,6 +27,7 @@
#include "proton/object.hpp"
#include "proton/session.hpp"
#include "proton/types.h"
+
#include <string>
struct pn_connection_t;
@@ -39,16 +40,14 @@ class sender;
class sender_options;
class receiver;
class receiver_options;
-
-namespace io {
-class connection_engine;
-}
+class container;
+template <class T> class thread_safe;
/// A connection to a remote AMQP peer.
class
PN_CPP_CLASS_EXTERN connection : public internal::object<pn_connection_t>, public endpoint {
/// @cond INTERNAL
- connection(pn_connection_t* c) : internal::object<pn_connection_t>(c) {}
+ PN_CPP_EXTERN connection(pn_connection_t* c) : internal::object<pn_connection_t>(c) {}
/// @endcond
public:
@@ -119,13 +118,13 @@ PN_CPP_CLASS_EXTERN connection : public internal::object<pn_connection_t>, publi
PN_CPP_EXTERN uint16_t max_sessions() const;
PN_CPP_EXTERN uint32_t idle_timeout() const;
- /// @cond INTERNAL
private:
void user(const std::string &);
void password(const std::string &);
friend class internal::factory<connection>;
friend class connector;
+ friend class proton::thread_safe<connection>;
/// @endcond
};
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/include/proton/connection_options.hpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/include/proton/connection_options.hpp b/proton-c/bindings/cpp/include/proton/connection_options.hpp
index c6a7c04..01826e7 100644
--- a/proton-c/bindings/cpp/include/proton/connection_options.hpp
+++ b/proton-c/bindings/cpp/include/proton/connection_options.hpp
@@ -70,6 +70,9 @@ class connection_options {
/// Create an empty set of options.
PN_CPP_EXTERN connection_options();
+ /// Shorthand for connection_options().handler(h)
+ PN_CPP_EXTERN connection_options(class handler& h);
+
/// Copy options.
PN_CPP_EXTERN connection_options(const connection_options&);
@@ -80,8 +83,10 @@ class connection_options {
// XXX add C++11 move operations
- /// Set a handler for the connection.
- PN_CPP_EXTERN connection_options& handler(class handler *);
+ /// Set a connection handler.
+ ///
+ /// The handler must not be deleted until handler::on_transport_close() is called.
+ PN_CPP_EXTERN connection_options& handler(class handler&);
/// Set the maximum frame size.
PN_CPP_EXTERN connection_options& max_frame_size(uint32_t max);
@@ -136,9 +141,6 @@ class connection_options {
/// Update option values from values set in other.
PN_CPP_EXTERN connection_options& update(const connection_options& other);
- /// Copy and update option values from values set in other.
- PN_CPP_EXTERN connection_options update(const connection_options& other) const;
-
private:
void apply(connection&) const;
proton_handler* handler() const;
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/include/proton/container.hpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/include/proton/container.hpp b/proton-c/bindings/cpp/include/proton/container.hpp
index 540f4cf..6d613e2 100644
--- a/proton-c/bindings/cpp/include/proton/container.hpp
+++ b/proton-c/bindings/cpp/include/proton/container.hpp
@@ -22,9 +22,13 @@
*
*/
-#include "proton/duration.hpp"
-#include "proton/export.hpp"
-#include "proton/pn_unique_ptr.hpp"
+// FIXME aconway 2016-05-04: doc
+
+#include <proton/connection_options.hpp>
+#include <proton/error_condition.hpp>
+#include <proton/listener.hpp>
+#include <proton/pn_unique_ptr.hpp>
+#include <proton/thread_safe.hpp>
#include <string>
@@ -32,131 +36,152 @@ namespace proton {
class connection;
class connection_options;
-class acceptor;
-class handler;
+class container_impl;
class handler;
+class listen_handler;
class receiver;
class receiver_options;
class sender;
class sender_options;
class task;
-class container_impl;
-namespace internal {
-class link;
-}
+class container;
/// A top-level container of connections, sessions, senders and receivers.
///
/// A container gives a unique identity to each communicating peer. It
/// is often a process-level object.
-
+///
/// It serves as an entry point to the API, allowing connections, senders
/// and receivers to be established. It can be supplied with an event handler
/// in order to intercept important messaging events, such as newly
/// received messages or newly issued credit for sending
/// messages.
-class container {
- public:
- /// Create a container.
+class PN_CPP_CLASS_EXTERN container {
+ public:
+ PN_CPP_EXTERN virtual ~container();
+
+ /// Connect to url, send an `open` request to the remote peer.
///
- /// Container ID should be unique within your system. By default a
- /// random ID is generated.
+ /// Options are applied to the connection as follows, values in later
+ /// options override earlier ones:
///
- /// This container will not be very useful unless event handlers are supplied
- /// as options when creating a connection/listener/sender or receiver.
- PN_CPP_EXTERN container();
- PN_CPP_EXTERN container(const std::string& id);
-
- /// Create a container with an event handler.
+ /// 1. client_connection_options()
+ /// 2. options passed to connect()
+ ///
+ /// The handler in the composed options is used to call
+ /// proton::handler::on_connection_open() when the remote peer's open response
+ /// is received.
+ ///@{
+ virtual returned<connection> connect(const std::string& url, const connection_options &) = 0;
+ PN_CPP_EXTERN returned<connection> connect(const std::string& url);
+ ///@}
+
+ ///@cond INTERNAL
+ /// Stop listening on url, must match the url string given to listen().
+ /// You can also use the proton::listener object returned by listen()
+ virtual void stop_listening(const std::string& url) = 0;
+ ///@endcond
+
+ // FIXME aconway 2016-05-13: doc options
+
+ /// Start listening on url.
///
- /// Container ID should be unique within your system. By default a
- /// random ID is generated.
- PN_CPP_EXTERN container(handler& mhandler);
- PN_CPP_EXTERN container(handler& mhandler, const std::string& id);
+ /// Calls to the @ref listen_handler are serialized for this listener,
+ /// but handlers attached to separate listeners may be called concurrently.
+ ///
+ /// @param url identifies a listening url.
+ /// @param lh handles listening events
+ /// @return listener lets you stop listening
+ virtual listener listen(const std::string& url, listen_handler& lh) = 0;
+
+ /// Listen with a fixed set of options for all accepted connections.
+ /// See listen(const std::string&, listen_handler&)
+ PN_CPP_EXTERN virtual listener listen(const std::string& url, const connection_options&);
- PN_CPP_EXTERN ~container();
+ /// Start listening on URL.
+ /// New connections will use the handler from server_connection_options()
+ PN_CPP_EXTERN virtual listener listen(const std::string& url);
- /// Open a connection to `url`.
- PN_CPP_EXTERN connection connect(const std::string& url);
- PN_CPP_EXTERN connection connect(const std::string& url,
- const connection_options &opts);
+ /// Run the container in this thread.
+ /// Returns when the container stops: see auto_stop() and stop().
+ ///
+ /// With a multithreaded container, call run() in multiple threads to create a thread pool.
+ virtual void run() = 0;
- /// Listen on `url` for incoming connections.
- PN_CPP_EXTERN acceptor listen(const std::string &url);
- PN_CPP_EXTERN acceptor listen(const std::string &url,
- const connection_options &opts);
+ /// If true, the container will stop (i.e. run() will return) when all
+ /// active connections and listeners are closed. If false the container
+ /// will keep running till stop() is called.
+ ///
+ /// auto_stop is set by default when a new container is created.
+ // FIXME aconway 2016-05-06: doc
+ virtual void auto_stop(bool) = 0;
- /// Start processing events. It returns when all connections and
- /// acceptors are closed.
- PN_CPP_EXTERN void run();
+ ///@name Stop the container with an optional error_condition err.
+ /// - abort all open connections and listeners.
+ /// - process final handler events and injected functions
+ /// - if !err.empty(), handlers will receive on_transport_error(err)
+ /// - run() will return in all threads.
+ virtual void stop(const error_condition& err = error_condition()) = 0;
/// Open a connection to `url` and open a sender for `url.path()`.
/// Any supplied sender or connection options will override the
/// container's template options.
- PN_CPP_EXTERN sender open_sender(const std::string &url);
- PN_CPP_EXTERN sender open_sender(const std::string &url,
- const proton::sender_options &o);
- PN_CPP_EXTERN sender open_sender(const std::string &url,
- const proton::sender_options &o,
- const connection_options &c);
+ /// @{
+ PN_CPP_EXTERN virtual returned<sender> open_sender(const std::string &url);
+ PN_CPP_EXTERN virtual returned<sender> open_sender(const std::string &url,
+ const proton::sender_options &o);
+ virtual returned<sender> open_sender(const std::string &url,
+ const proton::sender_options &o,
+ const connection_options &c) = 0;
+ //@}
/// Open a connection to `url` and open a receiver for
/// `url.path()`. Any supplied receiver or connection options will
/// override the container's template options.
- PN_CPP_EXTERN receiver open_receiver(const std::string&url);
- PN_CPP_EXTERN receiver open_receiver(const std::string&url,
- const proton::receiver_options &o);
- PN_CPP_EXTERN receiver open_receiver(const std::string&url,
- const proton::receiver_options &o,
- const connection_options &c);
+ /// @{
+ PN_CPP_EXTERN virtual returned<receiver> open_receiver(const std::string&url);
+ PN_CPP_EXTERN virtual returned<receiver> open_receiver(const std::string&url,
+ const proton::receiver_options &o);
+ virtual returned<receiver> open_receiver(const std::string&url,
+ const proton::receiver_options &o,
+ const connection_options &c) = 0;
+ ///@}
/// A unique identifier for the container.
- PN_CPP_EXTERN std::string id() const;
-
- /// @cond INTERNAL
- /// XXX settle some API questions
- /// Schedule a timer task event in delay milliseconds.
- PN_CPP_EXTERN task schedule(int delay);
- PN_CPP_EXTERN task schedule(int delay, handler *h);
-
- /// @endcond
-
- /// Copy the connection options to a template which will be
- /// applied to subsequent outgoing connections. These are applied
- /// first and overriden by additional connection options provided
- /// in other methods.
- PN_CPP_EXTERN void client_connection_options(const connection_options &);
-
- /// Copy the connection options to a template which will be
- /// applied to incoming connections. These are applied before the
- /// first open event on the connection.
- PN_CPP_EXTERN void server_connection_options(const connection_options &);
-
- /// Copy the sender options to a template applied to new senders
- /// created and opened by this container. They are applied before
- /// the open event on the sender and may be overriden by sender
- /// options in other methods.
- PN_CPP_EXTERN void sender_options(const sender_options &);
-
- /// Copy the receiver options to a template applied to new receivers
- /// created and opened by this container. They are applied before
- /// the open event on the receiver and may be overriden by receiver
- /// options in other methods.
- PN_CPP_EXTERN void receiver_options(const receiver_options &);
-
- /// @cond INTERNAL
- private:
- internal::pn_unique_ptr<container_impl> impl_;
-
- friend class connector;
- friend class messaging_adapter;
- friend class receiver_options;
- friend class sender_options;
- friend class session_options;
- /// @endcond
+ virtual std::string id() const = 0;
+
+ // FIXME aconway 2016-05-04: need timed injection to replace schedule()
+
+ /// Connection options that will be to outgoing connections. These are
+ /// applied first and overriden by options provided in connect() and
+ /// handler::on_connection_open()
+ /// @{
+ virtual void client_connection_options(const connection_options &) = 0;
+ virtual connection_options client_connection_options() const = 0;
+ ///@}
+
+ /// Connection options that will be applied to incoming connections. These
+ /// are applied first and overridden by options provided in listen(),
+ /// listen_handler::on_accept() and handler::on_connection_open()
+ /// @{
+ virtual void server_connection_options(const connection_options &) = 0;
+ virtual connection_options server_connection_options() const = 0;
+ ///@}
+
+ /// Sender options applied to senders created by this container. They are
+ /// applied before handler::on_sender_open() and can be over-ridden. @{
+ /// @{
+ virtual void sender_options(const sender_options &) = 0;
+ virtual class sender_options sender_options() const = 0;
+ ///@}
+
+ /// Receiver options applied to receivers created by this container. They
+ /// are applied before handler::on_receiver_open() and can be over-ridden.
+ /// @{
+ virtual void receiver_options(const receiver_options &) = 0;
+ virtual class receiver_options receiver_options() const = 0;
+ /// @}
};
-
}
-
#endif // PROTON_CPP_CONTAINER_H
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/include/proton/controller.hpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/include/proton/controller.hpp b/proton-c/bindings/cpp/include/proton/controller.hpp
deleted file mode 100644
index 6b0784c..0000000
--- a/proton-c/bindings/cpp/include/proton/controller.hpp
+++ /dev/null
@@ -1,118 +0,0 @@
-#ifndef PROTON_MT_HPP
-#define PROTON_MT_HPP
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-
-#include <proton/handler.hpp>
-#include <proton/connection_options.hpp>
-#include <proton/error_condition.hpp>
-
-#include <functional>
-#include <memory>
-
-namespace proton {
-
-class connection;
-
-/// The controller lets the application initiate and listen for connections, and
-/// start/stop overall IO activity. A single controller can manage many
-/// connections.
-///
-/// A controller associates a proton::handler with a proton::connection (the
-/// AMQP protocol connection) and the corresponding proton::transport
-/// (representing the underlying IO connection)
-///
-/// The first call to a proton::handler is always handler::on_transport_open(),
-/// the last is handler::on_transport_close(). Handlers can be deleted after
-/// handler::on_transport_close().
-///
-class controller {
- public:
- /// Create an instance of the default controller implementation.
- /// @param container_id set on connections for this controller.
- /// If empty, generate a random QUID default.
- PN_CPP_EXTERN static std::unique_ptr<controller> create();
-
- /// Get the controller associated with a connection.
- /// @throw proton::error if this is not a controller-managed connection.
- PN_CPP_EXTERN static controller& get(const proton::connection&);
-
- controller(const controller&) = delete;
-
- virtual ~controller() {}
-
- /// Start listening on address.
- ///
- /// @param address identifies a listening address.
- ///
- /// @param make_handler returns a handler for each accepted connection.
- /// handler::on_connection_open() is called with the incoming
- /// proton::connection. The handler can accept by calling
- /// connection::open()) or reject by calling connection::close()).
- ///
- /// Calls to the factory for this address are serialized. Calls for separate
- /// addresses in separate calls to listen() may be concurrent.
- ///
- virtual void listen(
- const std::string& address,
- std::function<proton::handler*(const std::string&)> make_handler,
- const connection_options& = connection_options()) = 0;
-
- /// Stop listening on address, must match exactly with address string given to listen().
- virtual void stop_listening(const std::string& address) = 0;
-
- /// Connect to address.
- ///
- /// The handler will get a proton::handler::on_connection_open() call the
- /// connection is completed by the other end.
- virtual void connect(
- const std::string& address, proton::handler&,
- const connection_options& = connection_options()) = 0;
-
- /// Default options for all connections, e.g. container_id.
- virtual void options(const connection_options&) = 0;
-
- /// Default options for all connections, e.g. container_id.
- virtual connection_options options() = 0;
-
- /// Run the controller in this thread. Returns when the controller is
- /// stopped. Multiple threads can call run()
- virtual void run() = 0;
-
- /// Stop the controller: abort open connections, run() will return in all threads.
- /// Handlers will receive on_transport_error() with the error_condition.
- virtual void stop(const error_condition& = error_condition()) = 0;
-
- /// The controller will stop (run() will return) when there are no open
- /// connections or listeners left.
- virtual void stop_on_idle() = 0;
-
- /// Wait till the controller is stopped.
- virtual void wait() = 0;
-
- protected:
- controller() {}
-};
-
-}
-
-
-#endif // PROTON_MT_HPP
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/include/proton/default_container.hpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/include/proton/default_container.hpp b/proton-c/bindings/cpp/include/proton/default_container.hpp
new file mode 100644
index 0000000..d3a7608
--- /dev/null
+++ b/proton-c/bindings/cpp/include/proton/default_container.hpp
@@ -0,0 +1,92 @@
+#ifndef PROTON_DEFAULT_CONTAINER_HPP
+#define PROTON_DEFAULT_CONTAINER_HPP
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+// FIXME aconway 2016-05-04: doc
+
+#include <proton/container.hpp>
+
+namespace proton {
+
+// FIXME aconway 2016-05-04: doc
+class PN_CPP_CLASS_EXTERN default_container : public container {
+ public:
+ /// Create a default, single-threaded container with a handler.
+ /// The handler will be called for all events on all connections in the container.
+ ///
+ /// Container ID should be unique within your system. If empty a random UUID is generated.
+ PN_CPP_EXTERN explicit default_container(proton::handler& h, const std::string& id = "");
+
+ /// Create a default, single-threaded container without a handler.
+ ///
+ /// Connections get their handlesr via proton::connection_options.
+ /// Container-wide defaults are set with client_connection_options() and
+ /// server_connection_options(). Per-connection options are set in connect()
+ /// and proton_listen_handler::on_accept for the proton::listen_handler
+ /// passed to listen()
+ ///
+ /// Container ID should be unique within your system. If empty a random UUID is generated.
+ PN_CPP_EXTERN explicit default_container(const std::string& id = "");
+
+ /// Wrap an existing container implementation as a default_container.
+ /// Takes ownership of c.
+ PN_CPP_EXTERN explicit default_container(container* c) : impl_(c) {}
+
+ // FIXME aconway 2016-05-13: @copydoc all.
+
+ PN_CPP_EXTERN returned<connection> connect(const std::string& url, const connection_options &) PN_CPP_OVERRIDE;
+ PN_CPP_EXTERN listener listen(const std::string& url, listen_handler& l) PN_CPP_OVERRIDE;
+
+ PN_CPP_EXTERN void stop_listening(const std::string& url) PN_CPP_OVERRIDE;
+
+ PN_CPP_EXTERN void run() PN_CPP_OVERRIDE;
+ PN_CPP_EXTERN void auto_stop(bool set) PN_CPP_OVERRIDE;
+
+ PN_CPP_EXTERN void stop(const error_condition& err = error_condition()) PN_CPP_OVERRIDE;
+
+ PN_CPP_EXTERN returned<sender> open_sender(
+ const std::string &url,
+ const proton::sender_options &o = proton::sender_options(),
+ const connection_options &c = connection_options()) PN_CPP_OVERRIDE;
+
+ PN_CPP_EXTERN returned<receiver> open_receiver(
+ const std::string&url,
+ const proton::receiver_options &o = proton::receiver_options(),
+ const connection_options &c = connection_options()) PN_CPP_OVERRIDE;
+
+ PN_CPP_EXTERN std::string id() const PN_CPP_OVERRIDE;
+ PN_CPP_EXTERN void client_connection_options(const connection_options &o) PN_CPP_OVERRIDE;
+ PN_CPP_EXTERN connection_options client_connection_options() const PN_CPP_OVERRIDE;
+ PN_CPP_EXTERN void server_connection_options(const connection_options &o) PN_CPP_OVERRIDE;
+ PN_CPP_EXTERN connection_options server_connection_options() const PN_CPP_OVERRIDE;
+ PN_CPP_EXTERN void sender_options(const class sender_options &o) PN_CPP_OVERRIDE;
+ PN_CPP_EXTERN class sender_options sender_options() const PN_CPP_OVERRIDE;
+ PN_CPP_EXTERN void receiver_options(const class receiver_options & o) PN_CPP_OVERRIDE;
+ PN_CPP_EXTERN class receiver_options receiver_options() const PN_CPP_OVERRIDE;
+
+ private:
+ internal::pn_unique_ptr<container> impl_;
+};
+
+}
+#endif // PROTON_DEFAULT_CONTAINER_HPP
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/include/proton/endpoint.hpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/include/proton/endpoint.hpp b/proton-c/bindings/cpp/include/proton/endpoint.hpp
index 593ff32..88259ca 100644
--- a/proton-c/bindings/cpp/include/proton/endpoint.hpp
+++ b/proton-c/bindings/cpp/include/proton/endpoint.hpp
@@ -21,10 +21,11 @@
* under the License.
*
*/
+
+#include "proton/comparable.hpp"
#include "proton/config.hpp"
-#include "proton/export.hpp"
#include "proton/error_condition.hpp"
-#include "proton/comparable.hpp"
+#include "proton/export.hpp"
namespace proton {
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/include/proton/event_loop.hpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/include/proton/event_loop.hpp b/proton-c/bindings/cpp/include/proton/event_loop.hpp
new file mode 100644
index 0000000..ad358c8
--- /dev/null
+++ b/proton-c/bindings/cpp/include/proton/event_loop.hpp
@@ -0,0 +1,71 @@
+#ifndef PROTON_IO_EVENT_LOOP_HPP
+#define PROTON_IO_EVENT_LOOP_HPP
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include <proton/config.hpp>
+
+#include <functional>
+
+#if PN_CPP_HAS_CPP11
+#include <future>
+#include <type_traits>
+#endif
+
+struct pn_connection_t;
+struct pn_session_t;
+struct pn_link_t;
+
+namespace proton {
+
+// FIXME aconway 2016-05-04: doc
+
+class inject_handler {
+ public:
+ virtual ~inject_handler() {}
+ virtual void on_inject() = 0;
+};
+
+class PN_CPP_CLASS_EXTERN event_loop {
+ public:
+ virtual ~event_loop() {}
+
+ // FIXME aconway 2016-05-05: doc, note bool return not throw because no
+ // atomic way to determine closd status and throw during shutdown is bad.
+ virtual bool inject(inject_handler&) = 0;
+#if PN_CPP_HAS_CPP11
+ virtual bool inject(std::function<void()>) = 0;
+#endif
+
+ protected:
+ event_loop() {}
+
+ private:
+ PN_CPP_EXTERN static event_loop* get(pn_connection_t*);
+ PN_CPP_EXTERN static event_loop* get(pn_session_t*);
+ PN_CPP_EXTERN static event_loop* get(pn_link_t*);
+
+ friend class connection;
+ template <class T> friend class thread_safe;
+};
+
+}
+
+#endif // PROTON_IO_EVENT_LOOP_HPP
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/include/proton/handler.hpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/include/proton/handler.hpp b/proton-c/bindings/cpp/include/proton/handler.hpp
index d72e394..c145fed 100644
--- a/proton-c/bindings/cpp/include/proton/handler.hpp
+++ b/proton-c/bindings/cpp/include/proton/handler.hpp
@@ -75,7 +75,7 @@ PN_CPP_CLASS_EXTERN handler
///
/// @{
- /// The event loop is starting.
+ /// The container event loop is starting.
PN_CPP_EXTERN virtual void on_container_start(container &c);
/// A message is received.
PN_CPP_EXTERN virtual void on_message(delivery &d, message &m);
@@ -151,6 +151,7 @@ PN_CPP_CLASS_EXTERN handler
internal::pn_unique_ptr<messaging_adapter> messaging_adapter_;
friend class container;
+ friend class container_impl;
friend class io::connection_engine;
friend class connection_options;
friend class receiver_options;
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/include/proton/io/connection_engine.hpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/include/proton/io/connection_engine.hpp b/proton-c/bindings/cpp/include/proton/io/connection_engine.hpp
index ded68de..8b8838f 100644
--- a/proton-c/bindings/cpp/include/proton/io/connection_engine.hpp
+++ b/proton-c/bindings/cpp/include/proton/io/connection_engine.hpp
@@ -38,8 +38,10 @@ struct pn_collector_t;
namespace proton {
-class handler;
-class work_queue; // Only used for multi-threaded connection_engines.
+class event_loop;
+class proton_handler;
+
+// FIXME aconway 2016-05-04: doc
/** @page integration
@@ -51,17 +53,15 @@ from any IO source into proton::handler event calls, and generates AMQP
byte-encoded output that can be written to any IO destination.
The integration needs to implement two user-visible interfaces:
- - proton::controller lets the user initiate or listen for connections.
- - proton::work_queue lets the user serialize their own work with a connection.
-
- @see epoll_controller.cpp for an example of an integration.
-
-[TODO controller doesn't belong in the mt namespace, a single-threaded
-integration would need a controller too.]
+ - proton::container lets the user initiate or listen for connections.
+ - proton::event_loop lets the user serialize their own work with a connection.
+ @see epoll_container.cpp for an example of an integration.
*/
namespace io {
+class link_namer;
+
/// Pointer to a mutable memory region with a size.
struct mutable_buffer {
char* data; ///< Beginning of the buffered data.
@@ -111,12 +111,29 @@ struct const_buffer {
class
PN_CPP_CLASS_EXTERN connection_engine {
public:
- /// Create a connection engine that dispatches to handler.
- // TODO aconway 2016-04-06: no options, only via handler.
- PN_CPP_EXTERN connection_engine(handler&, const connection_options& = connection_options());
+ /// Create a connection engine. opts must contain a handler.
+ /// Takes ownership of loop, will be deleted only when the proton::connection is.
+ PN_CPP_EXTERN connection_engine(proton::container&, link_namer&, event_loop* loop = 0);
PN_CPP_EXTERN ~connection_engine();
+ /// Configure a connection by applying exactly the options in opts (including proton::handler)
+ /// Does not apply any default options, to apply container defaults use connect() or accept()
+ /// instead.
+ void configure(const connection_options& opts=connection_options());
+
+ /// Call configure() with client options and call connection::open()
+ /// Options applied: container::id(), container::client_connection_options(), opts.
+ PN_CPP_EXTERN void connect(const connection_options& opts);
+
+ /// Call configure() with server options.
+ /// Options applied: container::id(), container::server_connection_options(), opts.
+ ///
+ /// Note this does not call connection::open(). If there is a handler in the
+ /// composed options it will receive handler::on_connection_open() and can
+ /// respond with connection::open() or connection::close()
+ PN_CPP_EXTERN void accept(const connection_options& opts);
+
/// The engine's read buffer. Read data into this buffer then call read_done() when complete.
/// Returns mutable_buffer(0, 0) if the engine cannot currently read data.
/// Calling dispatch() may open up more buffer space.
@@ -144,10 +161,19 @@ PN_CPP_CLASS_EXTERN connection_engine {
/// Note that there may still be events to dispatch() or data to read.
PN_CPP_EXTERN void write_close();
- /// Close the engine with an error that will be passed to handler::on_transport_error().
- /// Calls read_close() and write_close().
- /// Note: You still need to call dispatch() to process final close-down events.
- PN_CPP_EXTERN void close(const error_condition&);
+ /// Inform the engine that the transport been disconnected unexpectedly,
+ /// without completing the AMQP connection close sequence.
+ ///
+ /// This calls read_close(), write_close(), sets the transport().error() and
+ /// queues an `on_transport_error` event. You must call dispatch() one more
+ /// time to dispatch the handler::on_transport_error() call and other final
+ /// events.
+ ///
+ /// Note this does not close the connection() so that a proton::handler can
+ /// distinguish between a connection close error sent by the remote peer and
+ /// a transport failure.
+ ///
+ PN_CPP_EXTERN void disconnected(const error_condition& = error_condition());
/// Dispatch all available events and call the corresponding \ref handler methods.
///
@@ -168,20 +194,19 @@ PN_CPP_CLASS_EXTERN connection_engine {
/// Get the transport associated with this connection_engine.
PN_CPP_EXTERN proton::transport transport() const;
- /// For controller connections, set the connection's work_queue. Set
- /// via plain pointer, not std::shared_ptr so the connection_engine can be
- /// compiled with C++03. The work_queue must outlive the engine. The
- /// std::shared_ptr<work_queue> will be available via work_queue::get(this->connection())
- PN_CPP_EXTERN void work_queue(class work_queue*);
+ /// Get the container associated with this connection_engine.
+ PN_CPP_EXTERN proton::container& container() const;
- private:
+ private:
connection_engine(const connection_engine&);
connection_engine& operator=(const connection_engine&);
- proton::handler& handler_;
+ // FIXME aconway 2016-05-06: reduce binary compat footprint, move stuff to connection context.
+ proton::proton_handler* handler_;
proton::connection connection_;
proton::transport transport_;
proton::internal::pn_ptr<pn_collector_t> collector_;
+ proton::container& container_;
};
}}
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/include/proton/io/container_impl_base.hpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/include/proton/io/container_impl_base.hpp b/proton-c/bindings/cpp/include/proton/io/container_impl_base.hpp
new file mode 100644
index 0000000..d3fd74a
--- /dev/null
+++ b/proton-c/bindings/cpp/include/proton/io/container_impl_base.hpp
@@ -0,0 +1,120 @@
+#ifndef PROTON_IO_CONTAINER_IMPL_BASE_HPP
+#define PROTON_IO_CONTAINER_IMPL_BASE_HPP
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include <proton/io/link_namer.hpp>
+#include <proton/container.hpp>
+
+#include <mutex>
+#include <sstream>
+
+namespace proton {
+namespace io {
+
+/// Thread-safe partial implementation of proton::container interface to reduce
+/// boilerplate code in container implementations. Requires C++11.
+///
+/// You can ignore this class if you want to implement the functions in a
+/// different way.
+///
+class container_impl_base : public container {
+ public:
+
+ void client_connection_options(const connection_options & opts) {
+ store(client_copts_, opts);
+ }
+ connection_options client_connection_options() const {
+ return load(client_copts_);
+ }
+ void server_connection_options(const connection_options & opts) {
+ store(server_copts_, opts);
+ }
+ connection_options server_connection_options() const {
+ return load(server_copts_);
+ }
+ void sender_options(const class sender_options & opts) {
+ store(sender_opts_, opts);
+ }
+ class sender_options sender_options() const {
+ return load(sender_opts_);
+ }
+ void receiver_options(const class receiver_options & opts) {
+ store(receiver_opts_, opts);
+ }
+ class receiver_options receiver_options() const {
+ return load(receiver_opts_);
+ }
+
+ returned<sender> open_sender(
+ const std::string &url, const class sender_options &opts, const connection_options &copts)
+ {
+ return open_link<sender, class sender_options>(url, opts, copts, &connection::open_sender);
+ }
+
+ returned<receiver> open_receiver(
+ const std::string &url, const class receiver_options &opts, const connection_options &copts)
+ {
+ return open_link<receiver>(url, opts, copts, &connection::open_receiver);
+ }
+
+
+ private:
+ template<class T, class Opts>
+ returned<T> open_link(
+ const std::string &url_str, const Opts& opts, const connection_options& copts,
+ T (connection::*open_fn)(const std::string&, const Opts&))
+ {
+ std::string addr = url(url_str).path();
+ std::shared_ptr<thread_safe<connection> > ts_connection = connect(url_str, copts);
+ std::promise<returned<T> > result_promise;
+ auto do_open = [ts_connection, addr, opts, open_fn, &result_promise]() {
+ try {
+ connection c = ts_connection->unsafe();
+ returned<T> s = make_thread_safe((c.*open_fn)(addr, opts));
+ result_promise.set_value(s);
+ } catch (...) {
+ result_promise.set_exception(std::current_exception());
+ }
+ };
+ ts_connection->event_loop()->inject(do_open);
+ std::future<returned<T> > result_future = result_promise.get_future();
+ if (!result_future.valid())
+ throw error(url_str+": connection closed");
+ return result_future.get();
+ }
+
+ mutable std::mutex lock_;
+ template <class T> T load(const T& v) const {
+ std::lock_guard<std::mutex> g(lock_);
+ return v;
+ }
+ template <class T> void store(T& v, const T& x) const {
+ std::lock_guard<std::mutex> g(lock_);
+ v = x;
+ }
+ connection_options client_copts_, server_copts_;
+ class receiver_options receiver_opts_;
+ class sender_options sender_opts_;
+};
+
+}}
+
+#endif // PROTON_IO_CONTAINER_IMPL_BASE_HPP
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/include/proton/io/default_controller.hpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/include/proton/io/default_controller.hpp b/proton-c/bindings/cpp/include/proton/io/default_controller.hpp
deleted file mode 100644
index f876d5f..0000000
--- a/proton-c/bindings/cpp/include/proton/io/default_controller.hpp
+++ /dev/null
@@ -1,47 +0,0 @@
-#ifndef PROTON_IO_DRIVER_HPP
-#define PROTON_IO_DRIVER_HPP
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-#include <functional>
-#include <memory>
-
-namespace proton {
-
-class controller;
-
-namespace io {
-
-/// A proton::controller implementation can create a static instance of default_controller
-/// to register as the default implementation.
-/// If more than one implementation is linked, which one becomes the default
-/// is undefined.
-struct default_controller {
-
- /// A controller make-function takes a string container-id and returns a unique_ptr<controller>
- typedef std::function<std::unique_ptr<controller>()> make_fn;
-
- /// Construct a static instance of default_controller to register your controller factory.
- PN_CPP_EXTERN default_controller(make_fn);
-};
-
-}}
-
-#endif // PROTON_IO_CONTROLLER_HPP
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[2/4] qpid-proton git commit: PROTON-1184: C++ merge APIs for single
and multi-threaded use.
Posted by ac...@apache.org.
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/include/proton/io/link_namer.hpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/include/proton/io/link_namer.hpp b/proton-c/bindings/cpp/include/proton/io/link_namer.hpp
new file mode 100644
index 0000000..8add9a3
--- /dev/null
+++ b/proton-c/bindings/cpp/include/proton/io/link_namer.hpp
@@ -0,0 +1,37 @@
+#ifndef PROTON_IO_LINK_NAMER
+#define PROTON_IO_LINK_NAMER
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include <string>
+
+namespace proton {
+namespace io {
+
+/// Generate default link names that are unique within a container.
+/// base_container provides a default implementation.
+class link_namer {
+ public:
+ virtual ~link_namer() {}
+ virtual std::string link_name() = 0;
+};
+
+}}
+
+#endif // PROTON_IO_LINK_NAMER
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/include/proton/listen_handler.hpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/include/proton/listen_handler.hpp b/proton-c/bindings/cpp/include/proton/listen_handler.hpp
new file mode 100644
index 0000000..f836513
--- /dev/null
+++ b/proton-c/bindings/cpp/include/proton/listen_handler.hpp
@@ -0,0 +1,50 @@
+#ifndef PROTON_LISTEN_HANDLER_HPP
+#define PROTON_LISTEN_HANDLER_HPP
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+
+namespace proton {
+
+/// Implement this interface and pass to proton::container::listen() to be
+/// notified of new connections.
+class listen_handler {
+ public:
+ virtual ~listen_handler() {}
+
+ /// Called for each accepted connection.
+ ///
+ /// Returns connection_options to apply, including a proton::handler for
+ /// the connection. handler::on_connection_open() will be called with
+ /// the proton::connection, it can call connection::open() to accept or
+ /// connection::close() to reject the connection.
+ virtual connection_options on_accept()= 0;
+
+ /// Called if there is a listening error, with an error message.
+ /// close() will also be called.
+ virtual void on_error(const std::string&) {}
+
+ /// Called when this listen_handler is no longer needed, and can be deleted.
+ virtual void on_close() {}
+};
+}
+
+
+#endif // PROTON_LISTEN_HANDLER_HPP
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/include/proton/listener.hpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/include/proton/listener.hpp b/proton-c/bindings/cpp/include/proton/listener.hpp
new file mode 100644
index 0000000..2441e2b
--- /dev/null
+++ b/proton-c/bindings/cpp/include/proton/listener.hpp
@@ -0,0 +1,51 @@
+#ifndef PROTON_LISTENER_HPP
+#define PROTON_LISTENER_HPP
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include <proton/export.hpp>
+
+#include <string>
+
+namespace proton {
+
+class container;
+
+/// Returned by container::listen to allow you to stop listening on that address.
+class PN_CPP_CLASS_EXTERN listener {
+ public:
+ PN_CPP_EXTERN listener();
+ ///@cond internal
+ PN_CPP_EXTERN listener(container&, const std::string&);
+ ///@endcond internal
+
+ /// Stop listening on the address provided to the call to container::listen that
+ /// returned this listener.
+ PN_CPP_EXTERN void stop();
+
+ private:
+ std::string url_;
+ container* container_;
+};
+
+
+}
+
+#endif // PROTON_LISTENER_HPP
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/include/proton/object.hpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/include/proton/object.hpp b/proton-c/bindings/cpp/include/proton/object.hpp
index db52975..e94e4e6 100644
--- a/proton-c/bindings/cpp/include/proton/object.hpp
+++ b/proton-c/bindings/cpp/include/proton/object.hpp
@@ -28,6 +28,9 @@
#include <memory>
namespace proton {
+
+template <class T> class thread_safe;
+
namespace internal {
class pn_ptr_base {
@@ -83,6 +86,7 @@ template <class T> class object : private comparable<object<T> > {
#endif
protected:
+ typedef T pn_type;
object(pn_ptr<T> o) : object_(o) {}
T* pn_object() const { return object_.get(); }
@@ -91,6 +95,7 @@ template <class T> class object : private comparable<object<T> > {
friend bool operator==(const object& a, const object& b) { return a.object_ == b.object_; }
friend bool operator<(const object& a, const object& b) { return a.object_ < b.object_; }
+ template <class U> friend class proton::thread_safe;
};
/// Factory class used internally to make wrappers and extract proton objects
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/include/proton/receiver.hpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/include/proton/receiver.hpp b/proton-c/bindings/cpp/include/proton/receiver.hpp
index 13f615c..f0fd2c0 100644
--- a/proton-c/bindings/cpp/include/proton/receiver.hpp
+++ b/proton-c/bindings/cpp/include/proton/receiver.hpp
@@ -31,12 +31,13 @@
struct pn_connection_t;
namespace proton {
+template <class T> class thread_safe;
/// A link for receiving messages.
class
PN_CPP_CLASS_EXTERN receiver : public link {
/// @cond INTERNAL
- receiver(pn_link_t* r);
+ PN_CPP_EXTERN receiver(pn_link_t* r);
/// @endcond
public:
@@ -71,6 +72,7 @@ PN_CPP_CLASS_EXTERN receiver : public link {
/// @cond INTERNAL
friend class internal::factory<receiver>;
friend class receiver_iterator;
+ friend class thread_safe<receiver>;
/// @endcond
};
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/include/proton/receiver_options.hpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/include/proton/receiver_options.hpp b/proton-c/bindings/cpp/include/proton/receiver_options.hpp
index 5e2d62a..f40066f 100644
--- a/proton-c/bindings/cpp/include/proton/receiver_options.hpp
+++ b/proton-c/bindings/cpp/include/proton/receiver_options.hpp
@@ -75,9 +75,9 @@ class receiver_options {
/// Merge with another option set
PN_CPP_EXTERN void update(const receiver_options& other);
- /// Set a handler for events scoped to the receiver. If NULL,
- /// receiver-scoped events are discarded.
- PN_CPP_EXTERN receiver_options& handler(class handler *);
+ /// Set a handler for receiver events only.
+ /// The handler is no longer in use when handler::on_receiver_close() is called.
+ PN_CPP_EXTERN receiver_options& handler(class handler&);
/// Set the delivery mode on the receiver.
PN_CPP_EXTERN receiver_options& delivery_mode(delivery_mode);
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/include/proton/ret_ptr.hpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/include/proton/ret_ptr.hpp b/proton-c/bindings/cpp/include/proton/ret_ptr.hpp
new file mode 100644
index 0000000..95d63a0
--- /dev/null
+++ b/proton-c/bindings/cpp/include/proton/ret_ptr.hpp
@@ -0,0 +1,51 @@
+#ifndef PROTON_RET_PTR_HPP
+#define PROTON_RET_PTR_HPP
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "proton/config.hpp"
+#include <memory>
+
+namespace proton {
+
+/// A simple unique ownership pointer, used only as a return value from
+/// functions that transfer ownership to the caller.
+///
+/// If a ret_ptr return value is ignored, it will delete the return value
+/// automatically. Otherwise implicitly converts to a plain pointer that must be
+/// deleted by the caller using std::unique_ptr, std::shared_ptr, std::auto_ptr.
+/// or operator delete
+///
+template <class T> class ret_ptr {
+ public:
+ ret_ptr(const ret_ptr& x) : ptr_(x) {}
+ ~ret_ptr() { if (ptr_) delete(ptr_); }
+ operator T*() const { T* p = ptr_; ptr_ = 0; return p; }
+
+ private:
+ void operator=(const ret_ptr&);
+ ret_ptr(T* p=0) : ptr_(p) {}
+ T* ptr_;
+};
+
+}
+
+/// @endcond
+
+#endif // PROTON_RET_PTR_HPP
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/include/proton/sender.hpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/include/proton/sender.hpp b/proton-c/bindings/cpp/include/proton/sender.hpp
index 678a603..114e8bb 100644
--- a/proton-c/bindings/cpp/include/proton/sender.hpp
+++ b/proton-c/bindings/cpp/include/proton/sender.hpp
@@ -33,13 +33,14 @@
struct pn_connection_t;
namespace proton {
+template <class T> class thread_safe;
/// A link for sending messages.
class
PN_CPP_CLASS_EXTERN sender : public link
{
/// @cond INTERNAL
- sender(pn_link_t* s);
+ PN_CPP_EXTERN sender(pn_link_t* s);
/// @endcond
public:
@@ -68,6 +69,7 @@ PN_CPP_CLASS_EXTERN sender : public link
/// @cond INTERNAL
friend class internal::factory<sender>;
friend class sender_iterator;
+ friend class thread_safe<sender>;
/// @endcond
};
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/include/proton/sender_options.hpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/include/proton/sender_options.hpp b/proton-c/bindings/cpp/include/proton/sender_options.hpp
index 3569171..d74f1f8 100644
--- a/proton-c/bindings/cpp/include/proton/sender_options.hpp
+++ b/proton-c/bindings/cpp/include/proton/sender_options.hpp
@@ -76,9 +76,11 @@ class sender_options {
/// Merge with another option set
PN_CPP_EXTERN void update(const sender_options& other);
- /// Set a handler for events scoped to the sender. If NULL,
- /// sender-scoped events are discarded.
- PN_CPP_EXTERN sender_options& handler(class handler *);
+ /// Set a handler for sender events only.
+ /// The handler is no longer in use when handler::on_sender_close() is called.
+ /// handler::on_sender_close() may not be called if a connection is aborted,
+ /// in that case it should be cleaned up in its connection's handler::on_transport_close()
+ PN_CPP_EXTERN sender_options& handler(class handler&);
/// Set the delivery mode on the sender.
PN_CPP_EXTERN sender_options& delivery_mode(delivery_mode);
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/include/proton/session.hpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/include/proton/session.hpp b/proton-c/bindings/cpp/include/proton/session.hpp
index 30429b7..540f1bd 100644
--- a/proton-c/bindings/cpp/include/proton/session.hpp
+++ b/proton-c/bindings/cpp/include/proton/session.hpp
@@ -39,13 +39,14 @@ namespace proton {
class container;
class handler;
+template <class T> class thread_safe;
/// A container of senders and receivers.
class
PN_CPP_CLASS_EXTERN session : public internal::object<pn_session_t>, public endpoint
{
/// @cond INTERNAL
- session(pn_session_t* s) : internal::object<pn_session_t>(s) {}
+ PN_CPP_EXTERN session(pn_session_t* s) : internal::object<pn_session_t>(s) {}
/// @endcond
public:
@@ -102,6 +103,7 @@ PN_CPP_CLASS_EXTERN session : public internal::object<pn_session_t>, public endp
friend class internal::factory<session>;
friend class session_iterator;
+ friend class thread_safe<session>;
};
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/include/proton/source.hpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/include/proton/source.hpp b/proton-c/bindings/cpp/include/proton/source.hpp
index efe5889..a30a866 100644
--- a/proton-c/bindings/cpp/include/proton/source.hpp
+++ b/proton-c/bindings/cpp/include/proton/source.hpp
@@ -68,7 +68,7 @@ class source : public terminus {
source(pn_terminus_t* t);
source(const sender&);
source(const receiver&);
- friend class internal::factory<source>;
+ friend class proton::internal::factory<source>;
friend class sender;
friend class receiver;
/// @endcond
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/include/proton/thread_safe.hpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/include/proton/thread_safe.hpp b/proton-c/bindings/cpp/include/proton/thread_safe.hpp
new file mode 100644
index 0000000..b721f76
--- /dev/null
+++ b/proton-c/bindings/cpp/include/proton/thread_safe.hpp
@@ -0,0 +1,173 @@
+#ifndef PROTON_THREAD_SAFE_HPP
+#define PROTON_THREAD_SAFE_HPP
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ pp * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include <proton/config.hpp>
+#include <proton/connection.hpp>
+#include <proton/event_loop.hpp>
+#include <proton/object.hpp>
+#include <proton/type_traits.hpp>
+
+#include <functional>
+
+// FIXME aconway 2016-05-03: doc
+
+namespace proton {
+class connection;
+class session;
+class link;
+class sender;
+class receiver;
+
+namespace internal {
+template <class T> struct endpoint_traits;
+template<> struct endpoint_traits<connection> {};
+template<> struct endpoint_traits<session> {};
+template<> struct endpoint_traits<link> {};
+template<> struct endpoint_traits<sender> {};
+template<> struct endpoint_traits<receiver> {};
+}
+
+template <class T> class returned;
+
+// FIXME aconway 2016-05-09: doc
+/// Events for each proton::connection are processed sequentially in an
+/// event-loop. proton::handler functions for a single connection are never
+/// called concurrently. inject() lets you add user-defined function calls to
+/// be processed in the event loop sequence.
+///
+/// thread_safe is useful with multi-threaded programs, where different
+/// connection's event-loops can run concurrently. Proton objects associated
+/// with a connection (proton::connection, proton:sender etc.) are not thread
+/// safe, so they can only be used in the context of the connections thread_safe.
+/// inject() allows any thread (application threads or thread_safe threads for
+/// different connections) to communicate safely.
+///
+template <class T>
+class thread_safe : private internal::pn_ptr_base, private internal::endpoint_traits<T>
+{
+ typedef typename T::pn_type pn_type;
+
+ struct inject_decref : public inject_handler {
+ pn_type* ptr_;
+ inject_decref(pn_type* p) : ptr_(p) {}
+ void on_inject() { decref(ptr_); delete this; }
+ };
+
+ public:
+ static void operator delete(void*) {}
+
+ ~thread_safe() {
+ if (ptr()) {
+ if (event_loop()) {
+#if PN_CPP_HAS_CPP11
+ event_loop()->inject(std::bind(&decref, ptr()));
+#else
+ event_loop()->inject(*new inject_decref(ptr()));
+#endif
+ } else {
+ decref(ptr());
+ }
+ }
+ }
+
+ class event_loop* event_loop() { return event_loop::get(ptr()); }
+
+ // FIXME aconway 2016-05-04: doc
+ T unsafe() { return T(ptr()); }
+
+ // Caller must delete
+ static thread_safe* create(const T& obj) { return new (obj.pn_object()) thread_safe(); }
+
+ private:
+ static void* operator new(size_t, pn_type* p) { return p; }
+ static void operator delete(void*, pn_type*) {}
+ thread_safe() { incref(ptr()); }
+ pn_type* ptr() { return reinterpret_cast<pn_type*>(this); }
+
+ // Non-copyable.
+ thread_safe(const thread_safe&);
+ thread_safe& operator=(const thread_safe&);
+
+ friend class returned<T>;
+};
+
+// FIXME aconway 2016-05-04: doc.
+// Temporary return value only, not a real smart_ptr.
+// Release or convert to some other pointer type immediately.
+template <class T>
+class returned : private internal::endpoint_traits<T>
+{
+ public:
+ /// Take ownership
+ explicit returned(thread_safe<T>* p) : ptr_(p) {}
+ /// Transfer ownership.
+ /// Use the same "cheat" as std::auto_ptr, calls x.release() even though x is const.
+ returned(const returned& x) : ptr_(const_cast<returned&>(x).release()) {}
+ /// Delete if still owned.
+ ~returned() { if (ptr_) delete ptr_; }
+
+ /// Release ownership.
+ thread_safe<T>* release() const { thread_safe<T>* p = ptr_; ptr_ = 0; return p; }
+
+ /// Implicit conversion to target, usable only in a safe context.
+ operator T() { return ptr_->unsafe(); }
+
+#if PN_CPP_HAS_CPP11
+ /// Release to a std::shared_ptr
+ operator std::shared_ptr<thread_safe<T> >() {
+ return std::shared_ptr<thread_safe<T> >(release());
+ }
+
+ /// Release to a std::unique_ptr
+ operator std::unique_ptr<thread_safe<T> >() {
+ return std::unique_ptr<thread_safe<T> >(release());
+ }
+#endif
+
+ private:
+ void operator=(const returned&);
+ mutable thread_safe<T>* ptr_;
+};
+
+template <class T> returned<T> make_thread_safe(const T& obj) {
+ return returned<T>(thread_safe<T>::create(obj));
+}
+
+template <class T> T make_thread_unsafe(T* p) { return p->unsafe(); }
+
+
+#if PN_CPP_HAS_CPP11
+template <class T> std::shared_ptr<thread_safe<T> > make_shared_thread_safe(const T& obj) {
+ return std::shared_ptr<thread_safe<T> >(thread_safe<T>::create(obj));
+}
+template <class T> std::unique_ptr<thread_safe<T> > make_unique_thread_safe(const T& obj) {
+ return std::unique_ptr<thread_safe<T> >(thread_safe<T>::create(obj));
+}
+
+template <class T> T make_thread_unsafe(const std::shared_ptr<T>& p) { return p->unsafe(); }
+template <class T> T make_thread_unsafe(const std::unique_ptr<T>& p) { return p->unsafe(); }
+#endif
+
+
+}
+
+#endif // PROTON_THREAD_SAFE_HPP
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/include/proton/work_queue.hpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/include/proton/work_queue.hpp b/proton-c/bindings/cpp/include/proton/work_queue.hpp
deleted file mode 100644
index 1fb84ce..0000000
--- a/proton-c/bindings/cpp/include/proton/work_queue.hpp
+++ /dev/null
@@ -1,75 +0,0 @@
-#ifndef PROTON_WORK_QUEUE_HPP
-#define PROTON_WORK_QUEUE_HPP
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-pp * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-#include <proton/handler.hpp>
-#include <proton/connection_options.hpp>
-
-#include <functional>
-#include <memory>
-
-namespace proton {
-
-class connection;
-
-/// A work_queue takes work (in the form of function objects) that will be be
-/// serialized with other activity on a connection. Typically the work is a call
-/// to user-defined member functions on the handler(s) associated with a
-/// connection, which will be called serialized with respect to
-/// proton::handler::on_* event functions.
-///
-class work_queue : public std::enable_shared_from_this<work_queue> {
- public:
- work_queue(const work_queue&) = delete;
- virtual ~work_queue() {}
-
- /// Get the work_queue associated with a connection.
- /// @throw proton::error if this is not a controller-managed connection.
- PN_CPP_EXTERN static std::shared_ptr<work_queue> get(const proton::connection&);
-
- /// push a function object on the queue to be invoked in a safely serialized
- /// away.
- ///
- /// @return true if `f()` was pushed and will be called. False if the
- /// work_queue is already closed and f() will never be called.
- ///
- /// Note 1: On returning true, the application can rely on f() being called
- /// eventually. However f() should check the state when it executes as
- /// links, sessions or even the connection may have closed by the time f()
- /// is executed.
- ///
- /// Note 2: You must not push() in a handler or work_queue function on the
- /// *same connection* as the work_queue you are pushing to. That could cause
- /// a deadlock.
- ///
- virtual bool push(std::function<void()>) = 0;
-
- /// Get the controller associated with this work_queue.
- virtual class controller& controller() const = 0;
-
- protected:
- work_queue() {}
-};
-
-}
-
-
-#endif // PROTON_WORK_QUEUE_HPP
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/src/acceptor.cpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/src/acceptor.cpp b/proton-c/bindings/cpp/src/acceptor.cpp
index 8f4c722..1b5a0bd 100644
--- a/proton-c/bindings/cpp/src/acceptor.cpp
+++ b/proton-c/bindings/cpp/src/acceptor.cpp
@@ -18,20 +18,10 @@
* under the License.
*
*/
-
-#include "proton/acceptor.hpp"
-#include "proton/error.hpp"
-#include "proton/connection_options.hpp"
-#include "msg.hpp"
-#include "contexts.hpp"
+#include "acceptor.hpp"
namespace proton {
void acceptor::close() { pn_acceptor_close(pn_object()); }
-class connection_options& acceptor::connection_options() {
- listener_context& lc(listener_context::get(pn_object()));
- return lc.connection_options;
-}
-
}
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/src/acceptor.hpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/src/acceptor.hpp b/proton-c/bindings/cpp/src/acceptor.hpp
new file mode 100644
index 0000000..51d1094
--- /dev/null
+++ b/proton-c/bindings/cpp/src/acceptor.hpp
@@ -0,0 +1,61 @@
+#ifndef PROTON_CPP_ACCEPTOR_H
+#define PROTON_CPP_ACCEPTOR_H
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include <proton/reactor.h>
+#include <proton/export.hpp>
+#include <proton/object.hpp>
+
+struct pn_acceptor_t;
+
+namespace proton {
+
+/// A context for accepting inbound connections.
+///
+/// @see container::listen
+class acceptor : public internal::object<pn_acceptor_t> {
+ /// @cond INTERNAL
+ acceptor(pn_acceptor_t* a) : internal::object<pn_acceptor_t>(a) {}
+ /// @endcond
+
+ public:
+ acceptor() : internal::object<pn_acceptor_t>(0) {}
+
+ /// Close the acceptor.
+ PN_CPP_EXTERN void close();
+
+ /// Return the current set of connection options applied to
+ /// inbound connectons by the acceptor.
+ ///
+ /// Note that changes made to the connection options only affect
+ /// connections accepted after this call returns.
+ PN_CPP_EXTERN class connection_options &connection_options();
+
+ /// @cond INTERNAL
+ friend class internal::factory<acceptor>;
+ /// @endcond
+};
+
+}
+
+#endif // PROTON_CPP_ACCEPTOR_H
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/src/connection.cpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/src/connection.cpp b/proton-c/bindings/cpp/src/connection.cpp
index 175d495..6bab1a1 100644
--- a/proton-c/bindings/cpp/src/connection.cpp
+++ b/proton-c/bindings/cpp/src/connection.cpp
@@ -22,13 +22,13 @@
#include "proton_bits.hpp"
#include "proton/connection.hpp"
-
#include "proton/container.hpp"
-#include "proton/transport.hpp"
-#include "proton/session.hpp"
#include "proton/error.hpp"
-#include "connector.hpp"
+#include "proton/event_loop.hpp"
+#include "proton/session.hpp"
+#include "proton/transport.hpp"
+#include "connector.hpp"
#include "container_impl.hpp"
#include "contexts.hpp"
#include "msg.hpp"
@@ -72,10 +72,15 @@ std::string connection::container_id() const {
}
container& connection::container() const {
- pn_reactor_t *r = pn_object_reactor(pn_object());
- if (!r)
+ class container* c = connection_context::get(pn_object()).container;
+ if (!c) {
+ pn_reactor_t *r = pn_object_reactor(pn_object());
+ if (r)
+ c = &container_context::get(r);
+ }
+ if (!c)
throw proton::error("connection does not have a container");
- return container_context::get(r);
+ return *c;
}
session_range connection::sessions() const {
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/src/connection_options.cpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/src/connection_options.cpp b/proton-c/bindings/cpp/src/connection_options.cpp
index ab45e5b..cbcc5f8 100644
--- a/proton-c/bindings/cpp/src/connection_options.cpp
+++ b/proton-c/bindings/cpp/src/connection_options.cpp
@@ -25,6 +25,7 @@
#include "proton/ssl.hpp"
#include "proton/sasl.hpp"
+#include "acceptor.hpp"
#include "contexts.hpp"
#include "connector.hpp"
#include "messaging_adapter.hpp"
@@ -80,6 +81,7 @@ class connection_options::impl {
if (pn_ssl_init(ssl, ssl_client_options.value.pn_domain(), NULL))
throw error(MSG("client SSL/TLS initialization error"));
} else if (!outbound) {
+ // TODO aconway 2016-05-13: reactor only
pn_acceptor_t *pnp = pn_connection_acceptor(pnc);
if (pnp) {
listener_context &lc(listener_context::get(pnp));
@@ -144,6 +146,8 @@ class connection_options::impl {
connection_options::connection_options() : impl_(new impl()) {}
+connection_options::connection_options(class handler& h) : impl_(new impl()) { handler(h); }
+
connection_options::connection_options(const connection_options& x) : impl_(new impl()) {
*this = x;
}
@@ -160,13 +164,7 @@ connection_options& connection_options::update(const connection_options& x) {
return *this;
}
-connection_options connection_options::update(const connection_options& x) const {
- connection_options copy(*this);
- copy.update(x);
- return copy;
-}
-
-connection_options& connection_options::handler(class handler *h) { impl_->handler = h->messaging_adapter_.get(); return *this; }
+connection_options& connection_options::handler(class handler &h) { impl_->handler = h.messaging_adapter_.get(); return *this; }
connection_options& connection_options::max_frame_size(uint32_t n) { impl_->max_frame_size = n; return *this; }
connection_options& connection_options::max_sessions(uint16_t n) { impl_->max_sessions = n; return *this; }
connection_options& connection_options::idle_timeout(duration t) { impl_->idle_timeout = t; return *this; }
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/src/connector.cpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/src/connector.cpp b/proton-c/bindings/cpp/src/connector.cpp
index 8a70b77..7b46f7e 100644
--- a/proton-c/bindings/cpp/src/connector.cpp
+++ b/proton-c/bindings/cpp/src/connector.cpp
@@ -20,6 +20,7 @@
*/
#include "connector.hpp"
+#include "container_impl.hpp"
#include "proton/connection.hpp"
#include "proton/transport.hpp"
@@ -105,7 +106,7 @@ void connector::on_transport_closed(proton_event &) {
}
else {
// log "Disconnected, reconnecting in " << delay << " milliseconds"
- connection_.container().impl_.get()->schedule(delay, this);
+ static_cast<container_impl&>(connection_.container()).schedule(delay, this);
return;
}
}
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/src/container.cpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/src/container.cpp b/proton-c/bindings/cpp/src/container.cpp
index a147851..2da2353 100644
--- a/proton-c/bindings/cpp/src/container.cpp
+++ b/proton-c/bindings/cpp/src/container.cpp
@@ -18,13 +18,14 @@
* under the License.
*
*/
-#include "proton/container.hpp"
+#include "container_impl.hpp"
+
+#include "proton/container.hpp"
#include "proton/connection.hpp"
#include "proton/sender_options.hpp"
#include "proton/receiver_options.hpp"
#include "proton/session.hpp"
-#include "proton/acceptor.hpp"
#include "proton/error.hpp"
#include "proton/receiver.hpp"
#include "proton/receiver_options.hpp"
@@ -43,79 +44,47 @@
namespace proton {
-//// Public container class.
-
-container::container() {
- impl_.reset(new container_impl(*this, 0, std::string()));
-}
-
-container::container(const std::string& id) {
- impl_.reset(new container_impl(*this, 0, id));
-}
-
-container::container(handler &mhandler) {
- impl_.reset(new container_impl(*this, mhandler.messaging_adapter_.get(), std::string()));
-}
-
-container::container(handler &mhandler, const std::string& id) {
- impl_.reset(new container_impl(*this, mhandler.messaging_adapter_.get(), id));
-}
-
container::~container() {}
-connection container::connect(const std::string &url) {
- return impl_->connect(url, connection_options());
-}
-
-connection container::connect(const std::string &url, const connection_options &opts) {
- return impl_->connect(url, opts);
-}
-
-std::string container::id() const { return impl_->id_; }
-
-void container::run() { impl_->reactor_.run(); }
+/// Functions defined here are convenience overrides that can be triviall
+/// defined in terms of other pure virtual functions on container. Don't make
+/// container implementers wade thru all this boiler-plate.
-sender container::open_sender(const std::string &url) {
- return impl_->open_sender(url, proton::sender_options(), connection_options());
+returned<connection> container::connect(const std::string &url) {
+ return connect(url, connection_options());
}
-sender container::open_sender(const std::string &url, const proton::sender_options &lo) {
- return impl_->open_sender(url, lo, connection_options());
+returned<sender> container::open_sender(const std::string &url) {
+ return open_sender(url, proton::sender_options(), connection_options());
}
-sender container::open_sender(const std::string &url, const proton::sender_options &lo, const connection_options &co) {
- return impl_->open_sender(url, lo, co);
+returned<sender> container::open_sender(const std::string &url, const proton::sender_options &lo) {
+ return open_sender(url, lo, connection_options());
}
-receiver container::open_receiver(const std::string &url) {
- return impl_->open_receiver(url, proton::receiver_options(), connection_options());
+returned<receiver> container::open_receiver(const std::string &url) {
+ return open_receiver(url, proton::receiver_options(), connection_options());
}
-receiver container::open_receiver(const std::string &url, const proton::receiver_options &lo) {
- return impl_->open_receiver(url, lo, connection_options());
+returned<receiver> container::open_receiver(const std::string &url, const proton::receiver_options &lo) {
+ return open_receiver(url, lo, connection_options());
}
-receiver container::open_receiver(const std::string &url, const proton::receiver_options &lo, const connection_options &co) {
- return impl_->open_receiver(url, lo, co);
+namespace{
+ struct listen_opts : public listen_handler {
+ connection_options opts;
+ listen_opts(const connection_options& o) : opts(o) {}
+ connection_options on_accept() { return opts; }
+ void on_close() { delete this; }
+ };
}
-acceptor container::listen(const std::string &url) {
- return impl_->listen(url, connection_options());
+listener container::listen(const std::string& url, const connection_options& opts) {
+ return listen(url, *new listen_opts(opts));
}
-acceptor container::listen(const std::string &url, const connection_options &opts) {
- return impl_->listen(url, opts);
+listener container::listen(const std::string &url) {
+ return listen(url, connection_options());
}
-task container::schedule(int delay) { return impl_->schedule(delay, 0); }
-task container::schedule(int delay, handler *h) { return impl_->schedule(delay, h ? h->messaging_adapter_.get() : 0); }
-
-void container::client_connection_options(const connection_options &o) { impl_->client_connection_options(o); }
-
-void container::server_connection_options(const connection_options &o) { impl_->server_connection_options(o); }
-
-void container::sender_options(const class sender_options &o) { impl_->sender_options(o); }
-
-void container::receiver_options(const class receiver_options &o) { impl_->receiver_options(o); }
-
} // namespace proton
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/src/container_impl.cpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/src/container_impl.cpp b/proton-c/bindings/cpp/src/container_impl.cpp
index 19c3c94..e448fe3 100644
--- a/proton-c/bindings/cpp/src/container_impl.cpp
+++ b/proton-c/bindings/cpp/src/container_impl.cpp
@@ -18,11 +18,11 @@
* under the License.
*
*/
-#include "proton/container.hpp"
+
+#include "proton/default_container.hpp"
#include "proton/connection_options.hpp"
#include "proton/connection.hpp"
#include "proton/session.hpp"
-#include "proton/acceptor.hpp"
#include "proton/error.hpp"
#include "proton/sender.hpp"
#include "proton/receiver.hpp"
@@ -33,6 +33,7 @@
#include "proton/url.hpp"
#include "proton/uuid.hpp"
+#include "acceptor.hpp"
#include "connector.hpp"
#include "container_impl.hpp"
#include "contexts.hpp"
@@ -48,9 +49,8 @@
namespace proton {
-namespace {
-
-struct handler_context {
+class handler_context {
+ public:
static handler_context& get(pn_handler_t* h) {
return *reinterpret_cast<handler_context*>(pn_handler_mem(h));
}
@@ -67,7 +67,7 @@ struct handler_context {
static void dispatch(pn_handler_t *c_handler, pn_event_t *c_event, pn_event_type_t)
{
handler_context& hc(handler_context::get(c_handler));
- proton_event pevent(c_event, hc.container_);
+ proton_event pevent(c_event, *hc.container_);
pevent.dispatch(*hc.handler_);
return;
}
@@ -76,8 +76,6 @@ struct handler_context {
proton_handler *handler_;
};
-} // namespace
-
// Used to sniff for connector events before the reactor's global handler sees them.
class override_handler : public proton_handler
{
@@ -94,12 +92,12 @@ class override_handler : public proton_handler
pn_event_t *cevent = pe.pn_event();
pn_connection_t *conn = pn_event_connection(cevent);
if (conn) {
- proton_handler *override = connection_context::get(conn).handler.get();
- if (override && type != proton_event::CONNECTION_INIT) {
+ proton_handler *oh = connection_context::get(conn).handler.get();
+ if (oh && type != proton_event::CONNECTION_INIT) {
// Send event to connector
- pe.dispatch(*override);
+ pe.dispatch(*oh);
}
- else if (!override && type == proton_event::CONNECTION_INIT) {
+ else if (!oh && type == proton_event::CONNECTION_INIT) {
// Newly accepted connection from lister socket
connection c(make_wrapper(conn));
container_impl_.configure_server_connection(c);
@@ -111,21 +109,22 @@ class override_handler : public proton_handler
internal::pn_ptr<pn_handler_t> container_impl::cpp_handler(proton_handler *h) {
pn_handler_t *handler = h ? pn_handler_new(&handler_context::dispatch,
- sizeof(struct handler_context),
+ sizeof(class handler_context),
&handler_context::cleanup) : 0;
if (handler) {
handler_context &hc = handler_context::get(handler);
- hc.container_ = &container_;
+ hc.container_ = this;
hc.handler_ = h;
}
return internal::take_ownership(handler);
}
-container_impl::container_impl(container& c, messaging_adapter *h, const std::string& id) :
- container_(c), reactor_(reactor::create()), handler_(h),
- id_(id.empty() ? uuid::random().str() : id), id_gen_()
+container_impl::container_impl(const std::string& id, handler *h) :
+ reactor_(reactor::create()), handler_(h ? h->messaging_adapter_.get() : 0),
+ id_(id.empty() ? uuid::random().str() : id), id_gen_(),
+ auto_stop_(true)
{
- container_context::set(reactor_, container_);
+ container_context::set(reactor_, *this);
// Set our own global handler that "subclasses" the existing one
pn_handler_t *global_handler = reactor_.pn_global_handler();
@@ -141,51 +140,68 @@ container_impl::container_impl(container& c, messaging_adapter *h, const std::st
// the reactor's default globalhandler (pn_iohandler)
}
-container_impl::~container_impl() {}
+namespace {
+void close_acceptor(acceptor a) {
+ listen_handler*& lh = listener_context::get(unwrap(a)).listen_handler_;
+ if (lh) {
+ lh->on_close();
+ lh = 0;
+ }
+ a.close();
+}
+}
+
+container_impl::~container_impl() {
+ for (acceptors::iterator i = acceptors_.begin(); i != acceptors_.end(); ++i)
+ close_acceptor(i->second);
+}
-connection container_impl::connect(const proton::url &url, const connection_options &user_opts) {
+returned<connection> container_impl::connect(const std::string &urlstr, const connection_options &user_opts) {
connection_options opts = client_connection_options(); // Defaults
opts.update(user_opts);
proton_handler *h = opts.handler();
+ proton::url url(urlstr);
internal::pn_ptr<pn_handler_t> chandler = h ? cpp_handler(h) : internal::pn_ptr<pn_handler_t>();
connection conn(reactor_.connection_to_host(url.host(), url.port(), chandler.get()));
internal::pn_unique_ptr<connector> ctor(new connector(conn, url, opts));
connection_context& cc(connection_context::get(conn));
cc.handler.reset(ctor.release());
- cc.link_gen.prefix(id_gen_.next() + "/");
pn_connection_set_container(unwrap(conn), id_.c_str());
conn.open(opts);
- return conn;
+ return make_thread_safe(conn);
}
-sender container_impl::open_sender(const proton::url &url, const proton::sender_options &o1, const connection_options &o2) {
+returned<sender> container_impl::open_sender(const std::string &url, const proton::sender_options &o1, const connection_options &o2) {
proton::sender_options lopts(sender_options_);
lopts.update(o1);
connection_options copts(client_connection_options_);
copts.update(o2);
connection conn = connect(url, copts);
- std::string path = url.path();
- return conn.default_session().open_sender(path, lopts);
+ return make_thread_safe(conn.default_session().open_sender(proton::url(url).path(), lopts));
}
-receiver container_impl::open_receiver(const proton::url &url, const proton::receiver_options &o1, const connection_options &o2) {
+returned<receiver> container_impl::open_receiver(const std::string &url, const proton::receiver_options &o1, const connection_options &o2) {
proton::receiver_options lopts(receiver_options_);
lopts.update(o1);
connection_options copts(client_connection_options_);
copts.update(o2);
connection conn = connect(url, copts);
- std::string path = url.path();
- return conn.default_session().open_receiver(path, lopts);
+ return make_thread_safe(
+ conn.default_session().open_receiver(proton::url(url).path(), lopts));
}
-acceptor container_impl::listen(const proton::url& url, const connection_options &user_opts) {
+listener container_impl::listen(const std::string& url, listen_handler& lh) {
+ if (acceptors_.find(url) != acceptors_.end())
+ throw error("already listening on " + url);
connection_options opts = server_connection_options(); // Defaults
- opts.update(user_opts);
proton_handler *h = opts.handler();
+ // FIXME aconway 2016-05-12: chandler and acceptor??
internal::pn_ptr<pn_handler_t> chandler = h ? cpp_handler(h) : internal::pn_ptr<pn_handler_t>();
- pn_acceptor_t *acptr = pn_reactor_acceptor(reactor_.pn_object(), url.host().c_str(), url.port().c_str(), chandler.get());
+ proton::url u(url);
+ pn_acceptor_t *acptr = pn_reactor_acceptor(
+ reactor_.pn_object(), u.host().c_str(), u.port().c_str(), chandler.get());
if (!acptr)
throw error(MSG("accept fail: " <<
pn_error_text(pn_io_error(reactor_.pn_io())))
@@ -193,9 +209,17 @@ acceptor container_impl::listen(const proton::url& url, const connection_options
// Do not use pn_acceptor_set_ssl_domain(). Manage the incoming connections ourselves for
// more flexibility (i.e. ability to change the server cert for a long running listener).
listener_context& lc(listener_context::get(acptr));
- lc.connection_options = opts;
- lc.ssl = url.scheme() == url::AMQPS;
- return make_wrapper(acptr);
+ lc.listen_handler_ = &lh;
+ lc.ssl = u.scheme() == url::AMQPS;
+ listener_context::get(acptr).listen_handler_ = &lh;
+ acceptors_[url] = make_wrapper(acptr);
+ return listener(*this, url);
+}
+
+void container_impl::stop_listening(const std::string& url) {
+ acceptors::iterator i = acceptors_.find(url);
+ if (i != acceptors_.end())
+ close_acceptor(i->second);
}
task container_impl::schedule(int delay, proton_handler *h) {
@@ -224,9 +248,50 @@ void container_impl::receiver_options(const proton::receiver_options &opts) {
void container_impl::configure_server_connection(connection &c) {
pn_acceptor_t *pnp = pn_connection_acceptor(unwrap(c));
listener_context &lc(listener_context::get(pnp));
- connection_context::get(c).link_gen.prefix(id_gen_.next() + "/");
pn_connection_set_container(unwrap(c), id_.c_str());
- lc.connection_options.apply(c);
+ connection_options opts = server_connection_options_;
+ opts.update(lc.get_options());
+ opts.apply(c);
}
+void container_impl::run() {
+ do {
+ reactor_.run();
+ } while (!auto_stop_);
+}
+
+void container_impl::stop(const error_condition&) {
+ reactor_.stop();
+}
+
+void container_impl::auto_stop(bool set) {
+ auto_stop_ = set;
+}
+
+
+default_container::default_container(handler& h, const std::string& id) : impl_(new container_impl(id, &h)) {}
+default_container::default_container(const std::string& id) : impl_(new container_impl(id)) {}
+
+returned<connection> default_container::connect(const std::string& url, const connection_options &o) { return impl_->connect(url, o); }
+listener default_container::listen(const std::string& url, listen_handler& l) { return impl_->listen(url, l); }
+void default_container::stop_listening(const std::string& url) { impl_->stop_listening(url); }
+
+void default_container::run() { impl_->run(); }
+void default_container::auto_stop(bool set) { impl_->auto_stop(set); }
+void default_container::stop(const error_condition& err) { impl_->stop(err); }
+
+returned<sender> default_container::open_sender(const std::string &u, const proton::sender_options &o, const connection_options &c) { return impl_->open_sender(u, o, c); }
+returned<receiver> default_container::open_receiver(const std::string &u, const proton::receiver_options &o, const connection_options &c) { return impl_->open_receiver(u, o, c); }
+
+std::string default_container::id() const { return impl_->id(); }
+void default_container::client_connection_options(const connection_options &o) { impl_->client_connection_options(o); }
+connection_options default_container::client_connection_options() const { return impl_->client_connection_options(); }
+void default_container::server_connection_options(const connection_options &o) { impl_->server_connection_options(o); }
+connection_options default_container::server_connection_options() const { return impl_->server_connection_options(); }
+void default_container::sender_options(const class sender_options &o) { impl_->sender_options(o); }
+class sender_options default_container::sender_options() const { return impl_->sender_options(); }
+void default_container::receiver_options(const class receiver_options & o) { impl_->receiver_options(o); }
+class receiver_options default_container::receiver_options() const { return impl_->receiver_options(); }
+
+
}
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/src/container_impl.hpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/src/container_impl.hpp b/proton-c/bindings/cpp/src/container_impl.hpp
index 8d57233..dbd1c79 100644
--- a/proton-c/bindings/cpp/src/container_impl.hpp
+++ b/proton-c/bindings/cpp/src/container_impl.hpp
@@ -22,8 +22,9 @@
*
*/
-#include "id_generator.hpp"
+#include "proton/io/link_namer.hpp"
+#include "proton/container.hpp"
#include "proton/connection.hpp"
#include "proton/connection_options.hpp"
#include "proton/duration.hpp"
@@ -36,6 +37,7 @@
#include "proton_handler.hpp"
#include <string>
+#include <sstream>
namespace proton {
@@ -46,48 +48,65 @@ class acceptor;
class container;
class url;
class task;
+class listen_handler;
-class container_impl
-{
+class container_impl : public container {
public:
- container_impl(container&, messaging_adapter*, const std::string& id);
+ container_impl(const std::string& id, handler* = 0);
~container_impl();
- connection connect(const url&, const connection_options&);
- sender open_sender(const url&, const proton::sender_options &, const connection_options &);
- receiver open_receiver(const url&, const proton::receiver_options &, const connection_options &);
- class acceptor listen(const url&, const connection_options &);
- duration timeout();
- void timeout(duration timeout);
- void client_connection_options(const connection_options &);
- const connection_options& client_connection_options() { return client_connection_options_; }
- void server_connection_options(const connection_options &);
- const connection_options& server_connection_options() { return server_connection_options_; }
- void sender_options(const proton::sender_options&);
- const proton::sender_options& sender_options() { return sender_options_; }
- void receiver_options(const proton::receiver_options&);
- const proton::receiver_options& receiver_options() { return receiver_options_; }
+ std::string id() const PN_CPP_OVERRIDE { return id_; }
+ returned<connection> connect(const std::string&, const connection_options&) PN_CPP_OVERRIDE;
+ returned<sender> open_sender(
+ const std::string&, const proton::sender_options &, const connection_options &) PN_CPP_OVERRIDE;
+ returned<receiver> open_receiver(
+ const std::string&, const proton::receiver_options &, const connection_options &) PN_CPP_OVERRIDE;
+ listener listen(const std::string&, listen_handler& lh) PN_CPP_OVERRIDE;
+ void stop_listening(const std::string&) PN_CPP_OVERRIDE;
+ void client_connection_options(const connection_options &) PN_CPP_OVERRIDE;
+ connection_options client_connection_options() const PN_CPP_OVERRIDE { return client_connection_options_; }
+ void server_connection_options(const connection_options &) PN_CPP_OVERRIDE;
+ connection_options server_connection_options() const PN_CPP_OVERRIDE { return server_connection_options_; }
+ void sender_options(const proton::sender_options&) PN_CPP_OVERRIDE;
+ class sender_options sender_options() const PN_CPP_OVERRIDE { return sender_options_; }
+ void receiver_options(const proton::receiver_options&) PN_CPP_OVERRIDE;
+ class receiver_options receiver_options() const PN_CPP_OVERRIDE { return receiver_options_; }
+ void run() PN_CPP_OVERRIDE;
+ void stop(const error_condition& err) PN_CPP_OVERRIDE;
+ void auto_stop(bool set) PN_CPP_OVERRIDE;
+ // non-interface functions
void configure_server_connection(connection &c);
task schedule(int delay, proton_handler *h);
internal::pn_ptr<pn_handler_t> cpp_handler(proton_handler *h);
-
std::string next_link_name();
private:
+ typedef std::map<std::string, acceptor> acceptors;
+
+ struct count_link_namer : public io::link_namer {
+ count_link_namer() : count_(0) {}
+ std::string link_name() {
+ // TODO aconway 2016-01-19: more efficient conversion, fixed buffer.
+ std::ostringstream o;
+ o << "PN" << std::hex << ++count_;
+ return o.str();
+ }
+ uint64_t count_;
+ };
- container& container_;
reactor reactor_;
proton_handler *handler_;
internal::pn_unique_ptr<proton_handler> override_handler_;
internal::pn_unique_ptr<proton_handler> flow_controller_;
std::string id_;
- id_generator id_gen_;
+ count_link_namer id_gen_;
connection_options client_connection_options_;
connection_options server_connection_options_;
proton::sender_options sender_options_;
proton::receiver_options receiver_options_;
+ bool auto_stop_;
+ acceptors acceptors_;
- friend class container;
friend class messaging_adapter;
};
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/src/container_test.cpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/src/container_test.cpp b/proton-c/bindings/cpp/src/container_test.cpp
index 53ea172..9725a34 100644
--- a/proton-c/bindings/cpp/src/container_test.cpp
+++ b/proton-c/bindings/cpp/src/container_test.cpp
@@ -21,9 +21,9 @@
#include "test_bits.hpp"
#include "proton/connection.hpp"
#include "proton/connection_options.hpp"
-#include "proton/container.hpp"
+#include "proton/default_container.hpp"
#include "proton/handler.hpp"
-#include "proton/acceptor.hpp"
+#include "proton/listener.hpp"
#include <cstdlib>
#include <ctime>
@@ -31,12 +31,11 @@
#include <cstdio>
#include <sstream>
-#if __cplusplus < 201103L
-#define override
-#endif
+namespace {
using namespace test;
+
static std::string int2string(int n) {
std::ostringstream strm;
strm << n;
@@ -51,13 +50,13 @@ class test_handler : public proton::handler {
bool done;
std::string peer_vhost;
- proton::acceptor acptr;
+ proton::listener listener;
test_handler(const std::string h, const proton::connection_options& c_opts)
: host(h), opts(c_opts), closing(false), done(false)
{}
- void on_container_start(proton::container &c) override {
+ void on_container_start(proton::container &c) PN_CPP_OVERRIDE {
int port;
// I'm going to hell for this:
@@ -65,7 +64,7 @@ class test_handler : public proton::handler {
while (true) {
port = 20000 + (rand() % 30000);
try {
- acptr = c.listen("0.0.0.0:" + int2string(port));
+ listener = c.listen("0.0.0.0:" + int2string(port));
break;
} catch (...) {
// keep trying
@@ -74,15 +73,15 @@ class test_handler : public proton::handler {
proton::connection conn = c.connect(host + ":" + int2string(port), opts);
}
- void on_connection_open(proton::connection &c) override {
+ void on_connection_open(proton::connection &c) PN_CPP_OVERRIDE {
if (peer_vhost.empty() && !c.virtual_host().empty())
peer_vhost = c.virtual_host();
if (!closing) c.close();
closing = true;
}
- void on_connection_close(proton::connection &c) override {
- if (!done) acptr.close();
+ void on_connection_close(proton::connection &) PN_CPP_OVERRIDE {
+ if (!done) listener.stop();
done = true;
}
};
@@ -91,7 +90,7 @@ int test_container_vhost() {
proton::connection_options opts;
opts.virtual_host(std::string("a.b.c"));
test_handler th(std::string("127.0.0.1"), opts);
- proton::container(th).run();
+ proton::default_container(th).run();
ASSERT_EQUAL(th.peer_vhost, std::string("a.b.c"));
return 0;
}
@@ -99,7 +98,7 @@ int test_container_vhost() {
int test_container_default_vhost() {
proton::connection_options opts;
test_handler th(std::string("127.0.0.1"), opts);
- proton::container(th).run();
+ proton::default_container(th).run();
ASSERT_EQUAL(th.peer_vhost, std::string("127.0.0.1"));
return 0;
}
@@ -112,11 +111,13 @@ int test_container_no_vhost() {
proton::connection_options opts;
opts.virtual_host(std::string(""));
test_handler th(std::string("127.0.0.1"), opts);
- proton::container(th).run();
+ proton::default_container(th).run();
ASSERT_EQUAL(th.peer_vhost, std::string(""));
return 0;
}
+}
+
int main(int, char**) {
int failed = 0;
RUN_TEST(failed, test_container_vhost());
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/src/contexts.cpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/src/contexts.cpp b/proton-c/bindings/cpp/src/contexts.cpp
index 173764e..c70db01 100644
--- a/proton-c/bindings/cpp/src/contexts.cpp
+++ b/proton-c/bindings/cpp/src/contexts.cpp
@@ -92,6 +92,7 @@ container &container_context::get(pn_reactor_t *pn_reactor) {
}
listener_context& listener_context::get(pn_acceptor_t* a) {
+ // TODO aconway 2016-05-13: reactor only
// A Proton C pn_acceptor_t is really just a selectable
pn_selectable_t *sel = reinterpret_cast<pn_selectable_t*>(a);
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/src/contexts.hpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/src/contexts.hpp b/proton-c/bindings/cpp/src/contexts.hpp
index 9a4a9fe..05a4fa7 100644
--- a/proton-c/bindings/cpp/src/contexts.hpp
+++ b/proton-c/bindings/cpp/src/contexts.hpp
@@ -22,13 +22,16 @@
*
*/
-#include "proton/pn_unique_ptr.hpp"
-#include "proton/message.hpp"
#include "proton/connection.hpp"
#include "proton/container.hpp"
#include "proton/io/connection_engine.hpp"
+#include "proton/event_loop.hpp"
+#include "proton/listen_handler.hpp"
+#include "proton/message.hpp"
+#include "proton/pn_unique_ptr.hpp"
+
+#include "proton/io/link_namer.hpp"
-#include "id_generator.hpp"
#include "proton_handler.hpp"
struct pn_session_t;
@@ -41,7 +44,6 @@ namespace proton {
class proton_handler;
class reactor;
-class work_queue;
// Base class for C++ classes that are used as proton contexts.
// Contexts are pn_objects managed by pn reference counts, the C++ value is allocated in-place.
@@ -83,16 +85,16 @@ class context {
// Connection context used by all connections.
class connection_context : public context {
public:
- connection_context() : default_session(0), work_queue(0), collector(0) {}
+ connection_context() : container(0), default_session(0), link_gen(0), collector(0) {}
- // Used by all connections
+ class container* container;
pn_session_t *default_session; // Owned by connection.
message event_message; // re-used by messaging_adapter for performance.
- id_generator link_gen; // Link name generator.
- class work_queue* work_queue; // Work queue if this is proton::controller connection.
+ io::link_namer* link_gen; // Link name generator.
pn_collector_t* collector;
internal::pn_unique_ptr<proton_handler> handler;
+ internal::pn_unique_ptr<class event_loop> event_loop;
static connection_context& get(pn_connection_t *c) { return ref<connection_context>(id(c)); }
static connection_context& get(const connection& c) { return ref<connection_context>(id(c)); }
@@ -113,8 +115,9 @@ class container_context {
class listener_context : public context {
public:
static listener_context& get(pn_acceptor_t* c);
- listener_context() : ssl(false) {}
- class connection_options connection_options;
+ listener_context() : listen_handler_(0), ssl(false) {}
+ connection_options get_options() { return listen_handler_->on_accept(); }
+ class listen_handler* listen_handler_;
bool ssl;
};
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/src/controller.cpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/src/controller.cpp b/proton-c/bindings/cpp/src/controller.cpp
deleted file mode 100644
index 73403c2..0000000
--- a/proton-c/bindings/cpp/src/controller.cpp
+++ /dev/null
@@ -1,59 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-#include "contexts.hpp"
-
-#include <proton/error.hpp>
-#include <proton/controller.hpp>
-#include <proton/work_queue.hpp>
-
-#include <proton/io/default_controller.hpp>
-
-#include <utility>
-#include <memory>
-
-static proton::io::default_controller::make_fn make_default_controller;
-
-namespace proton {
-
-std::unique_ptr<controller> controller::create() {
- if (!make_default_controller)
- throw error("no default controller");
- return make_default_controller();
-}
-
-controller& controller::get(const proton::connection& c) {
- return work_queue::get(c)->controller();
-}
-
-std::shared_ptr<work_queue> work_queue::get(const proton::connection& c) {
- work_queue* wq = connection_context::get(c).work_queue;
- if (!wq)
- throw proton::error("connection has no controller");
- return wq->shared_from_this();
-}
-
-namespace io {
-// Register a default controller factory.
-default_controller::default_controller(default_controller::make_fn f) {
- make_default_controller = f;
-}
-} // namespace io
-
-} // namespace proton
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/src/engine_test.cpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/src/engine_test.cpp b/proton-c/bindings/cpp/src/engine_test.cpp
index 7ae05b1..1f5dbad 100644
--- a/proton-c/bindings/cpp/src/engine_test.cpp
+++ b/proton-c/bindings/cpp/src/engine_test.cpp
@@ -19,18 +19,19 @@
#include "test_bits.hpp"
+#include "test_dummy_container.hpp"
+#include "proton_bits.hpp"
+
+#include <proton/container.hpp>
#include <proton/uuid.hpp>
#include <proton/io/connection_engine.hpp>
+#include <proton/io/link_namer.hpp>
#include <proton/handler.hpp>
#include <proton/types_fwd.hpp>
#include <proton/link.hpp>
#include <deque>
#include <algorithm>
-#if __cplusplus < 201103L
-#define override
-#endif
-
namespace {
using namespace proton::io;
@@ -40,15 +41,22 @@ using namespace std;
typedef std::deque<char> byte_stream;
+struct dummy_link_namer : link_namer {
+ char name;
+ std::string link_name() { return std::string(1, name++); }
+};
+
+static dummy_link_namer namer;
+
/// In memory connection_engine that reads and writes from byte_streams
struct in_memory_engine : public connection_engine {
byte_stream& reads;
byte_stream& writes;
- in_memory_engine(byte_stream& rd, byte_stream& wr, handler &h,
- const connection_options &opts = connection_options()) :
- connection_engine(h, opts), reads(rd), writes(wr) {}
+ // Cheat on link_namer.
+ in_memory_engine(byte_stream& rd, byte_stream& wr, class container& cont) :
+ connection_engine(cont, namer), reads(rd), writes(wr) {}
void do_read() {
mutable_buffer rbuf = read_buffer();
@@ -70,18 +78,29 @@ struct in_memory_engine : public connection_engine {
}
}
- void process() { do_read(); do_write(); dispatch(); }
+ void process() {
+ if (!dispatch())
+ throw std::runtime_error("unexpected close: "+connection().error().what());
+ do_read();
+ do_write();
+ dispatch();
+ }
};
-/// A pair of engines that talk to each other in-memory.
+/// A pair of engines that talk to each other in-memory, simulating a connection.
struct engine_pair {
+ dummy_container conta, contb;
byte_stream ab, ba;
in_memory_engine a, b;
- engine_pair(handler& ha, handler& hb,
- const connection_options& ca = connection_options(),
- const connection_options& cb = connection_options()) :
- a(ba, ab, ha, ca), b(ab, ba, hb, cb) {}
+ engine_pair(const connection_options& oa, const connection_options& ob,
+ const std::string& name=""
+ ) :
+ conta(name+"a"), contb(name+"b"), a(ba, ab, conta), b(ab, ba, contb)
+ {
+ a.connect(oa);
+ b.accept(ob);
+ }
void process() { a.process(); b.process(); }
};
@@ -100,47 +119,64 @@ struct record_handler : public handler {
std::deque<proton::session> sessions;
std::deque<std::string> unhandled_errors, transport_errors, connection_errors;
- void on_receiver_open(receiver &l) override {
+ void on_receiver_open(receiver &l) PN_CPP_OVERRIDE {
receivers.push_back(l);
}
- void on_sender_open(sender &l) override {
+ void on_sender_open(sender &l) PN_CPP_OVERRIDE {
senders.push_back(l);
}
- void on_session_open(session &s) override {
+ void on_session_open(session &s) PN_CPP_OVERRIDE {
sessions.push_back(s);
}
- void on_transport_error(transport& t) override {
+ void on_transport_error(transport& t) PN_CPP_OVERRIDE {
transport_errors.push_back(t.error().what());
}
- void on_connection_error(connection& c) override {
+ void on_connection_error(connection& c) PN_CPP_OVERRIDE {
connection_errors.push_back(c.error().what());
}
- void on_error(const proton::error_condition& c) override {
+ void on_error(const proton::error_condition& c) PN_CPP_OVERRIDE {
unhandled_errors.push_back(c.what());
}
};
-void test_engine_container_id() {
- // Set container ID and prefix explicitly
+void test_engine_container_link_id() {
record_handler ha, hb;
- engine_pair e(ha, hb,
- connection_options().container_id("a"),
- connection_options().container_id("b"));
- e.a.connection().open();
- ASSERT_EQUAL("a", e.a.connection().container_id());
+ engine_pair e(ha, hb, "ids-");
+ e.a.connect(ha);
+ e.b.accept(hb);
+ ASSERT_EQUAL("ids-a", e.a.connection().container_id());
e.b.connection().open();
- ASSERT_EQUAL("b", e.b.connection().container_id());
+ ASSERT_EQUAL("ids-b", e.b.connection().container_id());
+
+ // Seed the global link namer
+ namer.name = 'x';
+
+ e.a.connection().open_sender("foo");
+ while (ha.senders.empty() || hb.receivers.empty()) e.process();
+ sender s = quick_pop(ha.senders);
+ ASSERT_EQUAL("x", s.name());
+
+ ASSERT_EQUAL("x", quick_pop(hb.receivers).name());
+
+ e.a.connection().open_receiver("bar");
+ while (ha.receivers.empty() || hb.senders.empty()) e.process();
+ ASSERT_EQUAL("y", quick_pop(ha.receivers).name());
+ ASSERT_EQUAL("y", quick_pop(hb.senders).name());
+
+ e.b.connection().open_receiver("");
+ while (ha.senders.empty() || hb.receivers.empty()) e.process();
+ ASSERT_EQUAL("z", quick_pop(ha.senders).name());
+ ASSERT_EQUAL("z", quick_pop(hb.receivers).name());
}
void test_endpoint_close() {
record_handler ha, hb;
engine_pair e(ha, hb);
- e.a.connection().open();
e.a.connection().open_sender("x");
e.a.connection().open_receiver("y");
while (ha.senders.size()+ha.receivers.size() < 2 ||
@@ -170,28 +206,44 @@ void test_endpoint_close() {
ASSERT_EQUAL("conn: bad connection", hb.connection_errors.front());
}
-void test_transport_close() {
- // Make sure an engine close calls the local on_transport_error() and aborts the remote.
+void test_engine_disconnected() {
+ // engine.disconnected() aborts the connection and calls the local on_transport_error()
record_handler ha, hb;
- engine_pair e(ha, hb);
- e.a.connection().open();
- while (!e.b.connection().active()) e.process();
- e.a.close(proton::error_condition("oops", "engine failure"));
- ASSERT(!e.a.dispatch()); // Final dispatch on a.
- ASSERT_EQUAL(1u, ha.transport_errors.size());
- ASSERT_EQUAL("oops: engine failure", ha.transport_errors.front());
- ASSERT_EQUAL(proton::error_condition("oops", "engine failure"),e.a.transport().error());
- // But connectoin was never protocol closed.
+ engine_pair e(ha, hb, "disconnected");
+ e.a.connect(ha);
+ e.b.accept(hb);
+ while (!e.a.connection().active() || !e.b.connection().active())
+ e.process();
+
+ // Close a with an error condition. The AMQP connection is still open.
+ e.a.disconnected(proton::error_condition("oops", "engine failure"));
+ ASSERT(!e.a.dispatch());
ASSERT(!e.a.connection().closed());
+ ASSERT(e.a.connection().error().empty());
ASSERT_EQUAL(0u, ha.connection_errors.size());
+ ASSERT_EQUAL("oops: engine failure", e.a.transport().error().what());
+ ASSERT_EQUAL(1u, ha.transport_errors.size());
+ ASSERT_EQUAL("oops: engine failure", ha.transport_errors.front());
+
+ // In a real app the IO code would detect the abort and do this:
+ e.b.disconnected(proton::error_condition("broken", "it broke"));
+ ASSERT(!e.b.dispatch());
+ ASSERT(!e.b.connection().closed());
+ ASSERT(e.b.connection().error().empty());
+ ASSERT_EQUAL(0u, hb.connection_errors.size());
+ // Proton-C adds (connection aborted) if transport closes too early,
+ // and provides a default message if there is no user message.
+ ASSERT_EQUAL("broken: it broke (connection aborted)", e.b.transport().error().what());
+ ASSERT_EQUAL(1u, hb.transport_errors.size());
+ ASSERT_EQUAL("broken: it broke (connection aborted)", hb.transport_errors.front());
}
}
int main(int, char**) {
int failed = 0;
- RUN_TEST(failed, test_engine_container_id());
+ RUN_TEST(failed, test_engine_container_link_id());
RUN_TEST(failed, test_endpoint_close());
- RUN_TEST(failed, test_transport_close());
+ RUN_TEST(failed, test_engine_disconnected());
return failed;
}
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/src/event_loop.cpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/src/event_loop.cpp b/proton-c/bindings/cpp/src/event_loop.cpp
new file mode 100644
index 0000000..ec10a8c
--- /dev/null
+++ b/proton-c/bindings/cpp/src/event_loop.cpp
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "contexts.hpp"
+
+#include <proton/session.h>
+#include <proton/link.h>
+
+#include <proton/event_loop.hpp>
+
+namespace proton {
+
+event_loop* event_loop::get(pn_connection_t* c) {
+ return connection_context::get(c).event_loop.get();
+}
+
+event_loop* event_loop::get(pn_session_t* s) {
+ return get(pn_session_connection(s));
+}
+
+event_loop* event_loop::get(pn_link_t* l) {
+ return get(pn_link_session(l));
+}
+
+}
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/src/id_generator.cpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/src/id_generator.cpp b/proton-c/bindings/cpp/src/id_generator.cpp
index c12d5a4..fde1658 100644
--- a/proton-c/bindings/cpp/src/id_generator.cpp
+++ b/proton-c/bindings/cpp/src/id_generator.cpp
@@ -17,14 +17,14 @@
* under the License.
*/
-#include "id_generator.hpp"
+#include "link_namer.hpp"
#include <sstream>
namespace proton {
-id_generator::id_generator(const std::string& s) : prefix_(s), count_(0) {}
+link_namer::link_namer(const std::string& s) : prefix_(s), count_(0) {}
-std::string id_generator::next() {
+std::string link_namer::next() {
// TODO aconway 2016-01-19: more efficient conversion, fixed buffer.
std::ostringstream o;
o << prefix_ << std::hex << ++count_;
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/src/id_generator.hpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/src/id_generator.hpp b/proton-c/bindings/cpp/src/id_generator.hpp
index f3df1e8..31261db 100644
--- a/proton-c/bindings/cpp/src/id_generator.hpp
+++ b/proton-c/bindings/cpp/src/id_generator.hpp
@@ -20,15 +20,15 @@
///@internal
#include "proton/types_fwd.hpp"
-#ifndef ID_GENERATOR_HPP
-#define ID_GENERATOR_HPP
+#ifndef LINK_NAMER_HPP
+#define LINK_NAMER_HPP
namespace proton {
/// @cond INTERNAL
-class id_generator {
+class link_namer {
public:
- id_generator(const std::string &prefix="");
+ link_namer(const std::string &prefix="");
std::string next();
void prefix(const std::string &p) { prefix_ = p; }
const std::string& prefix() const { return prefix_; }
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/src/io/connection_engine.cpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/src/io/connection_engine.cpp b/proton-c/bindings/cpp/src/io/connection_engine.cpp
index b7bb343..3153833 100644
--- a/proton-c/bindings/cpp/src/io/connection_engine.cpp
+++ b/proton-c/bindings/cpp/src/io/connection_engine.cpp
@@ -18,12 +18,14 @@
*/
#include "proton/io/connection_engine.hpp"
+#include "proton/io/link_namer.hpp"
+
+#include "proton/event_loop.hpp"
#include "proton/error.hpp"
#include "proton/handler.hpp"
#include "proton/uuid.hpp"
#include "contexts.hpp"
-#include "id_generator.hpp"
#include "messaging_adapter.hpp"
#include "msg.hpp"
#include "proton_bits.hpp"
@@ -42,45 +44,70 @@
namespace proton {
namespace io {
-connection_engine::connection_engine(class handler &h, const connection_options& opts):
- handler_(h),
+connection_engine::connection_engine(class container& cont, link_namer& namer, event_loop* loop) :
+ handler_(0),
connection_(make_wrapper(internal::take_ownership(pn_connection()).get())),
transport_(make_wrapper(internal::take_ownership(pn_transport()).get())),
- collector_(internal::take_ownership(pn_collector()).get())
+ collector_(internal::take_ownership(pn_collector()).get()),
+ container_(cont)
{
if (!connection_ || !transport_ || !collector_)
- throw proton::error("engine create");
+ throw proton::error("connection_engine create failed");
pn_transport_bind(unwrap(transport_), unwrap(connection_));
pn_connection_collect(unwrap(connection_), collector_.get());
- opts.apply(connection_);
+ connection_context& ctx = connection_context::get(connection_);
+ ctx.container = &container_;
+ ctx.link_gen = &namer;
+ ctx.event_loop.reset(loop);
+}
- // Provide local random defaults for connection_id and link_prefix if not by opts.
- if (connection_.container_id().empty())
- pn_connection_set_container(unwrap(connection_), uuid::random().str().c_str());
- id_generator &link_gen = connection_context::get(connection_).link_gen;
+void connection_engine::configure(const connection_options& opts) {
+ opts.apply(connection_);
+ handler_ = opts.handler();
+ if (handler_) {
+ collector_ = internal::take_ownership(pn_collector());
+ pn_connection_collect(unwrap(connection_), collector_.get());
+ }
connection_context::get(connection_).collector = collector_.get();
- if (link_gen.prefix().empty())
- link_gen.prefix(uuid::random().str()+"/");
}
connection_engine::~connection_engine() {
pn_transport_unbind(unwrap(transport_));
- pn_collector_free(collector_.release()); // Break cycle with connection_
+ if (collector_.get())
+ pn_collector_free(collector_.release()); // Break cycle with connection_
+}
+
+void connection_engine::connect(const connection_options& opts) {
+ connection_options all;
+ all.container_id(container_.id());
+ all.update(container_.client_connection_options());
+ all.update(opts);
+ configure(all);
+ connection().open();
+}
+
+void connection_engine::accept(const connection_options& opts) {
+ connection_options all;
+ all.container_id(container_.id());
+ all.update(container_.server_connection_options());
+ all.update(opts);
+ configure(all);
}
bool connection_engine::dispatch() {
- proton_handler& h = *handler_.messaging_adapter_;
- for (pn_event_t *e = pn_collector_peek(collector_.get());
- e;
- e = pn_collector_peek(collector_.get()))
- {
- proton_event pe(e, 0);
- try {
- pe.dispatch(h);
- } catch (const std::exception& e) {
- close(error_condition("exception", e.what()));
+ if (collector_.get()) {
+ for (pn_event_t *e = pn_collector_peek(collector_.get());
+ e;
+ e = pn_collector_peek(collector_.get()))
+ {
+ proton_event pe(e, container_);
+ try {
+ pe.dispatch(*handler_);
+ } catch (const std::exception& e) {
+ disconnected(error_condition("exception", e.what()));
+ }
+ pn_collector_pop(collector_.get());
}
- pn_collector_pop(collector_.get());
}
return !(pn_transport_closed(unwrap(transport_)));
}
@@ -119,7 +146,7 @@ void connection_engine::write_close() {
pn_transport_close_head(unwrap(transport_));
}
-void connection_engine::close(const proton::error_condition& err) {
+void connection_engine::disconnected(const proton::error_condition& err) {
set_error_condition(err, pn_transport_condition(unwrap(transport_)));
read_close();
write_close();
@@ -133,8 +160,8 @@ proton::transport connection_engine::transport() const {
return transport_;
}
-void connection_engine::work_queue(class work_queue* wq) {
- connection_context::get(connection()).work_queue = wq;
+proton::container& connection_engine::container() const {
+ return container_;
}
}}
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/src/listener.cpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/src/listener.cpp b/proton-c/bindings/cpp/src/listener.cpp
new file mode 100644
index 0000000..2639f5e
--- /dev/null
+++ b/proton-c/bindings/cpp/src/listener.cpp
@@ -0,0 +1,29 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "proton/listener.hpp"
+#include "proton/container.hpp"
+
+namespace proton {
+
+listener::listener() : container_(0) {}
+listener::listener(container& c, const std::string& u) : url_(u), container_(&c) {}
+void listener::stop() { if (container_) container_->stop_listening(url_); }
+
+}
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/src/messaging_adapter.cpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/src/messaging_adapter.cpp b/proton-c/bindings/cpp/src/messaging_adapter.cpp
index 9d6fa39..7f62082 100644
--- a/proton-c/bindings/cpp/src/messaging_adapter.cpp
+++ b/proton-c/bindings/cpp/src/messaging_adapter.cpp
@@ -60,9 +60,7 @@ messaging_adapter::messaging_adapter(handler &delegate) : delegate_(delegate) {}
messaging_adapter::~messaging_adapter(){}
void messaging_adapter::on_reactor_init(proton_event &pe) {
- // Container specific event
- if (pe.container())
- delegate_.on_container_start(*pe.container());
+ delegate_.on_container_start(pe.container());
}
void messaging_adapter::on_link_flow(proton_event &pe) {
@@ -249,26 +247,18 @@ void messaging_adapter::on_link_local_open(proton_event &pe) {
void messaging_adapter::on_link_remote_open(proton_event &pe) {
pn_link_t *lnk = pn_event_link(pe.pn_event());
- container *c = pe.container();
+ container& c = pe.container();
if (pn_link_is_receiver(lnk)) {
receiver r(make_wrapper<receiver>(lnk));
delegate_.on_receiver_open(r);
if (is_local_unititialised(pn_link_state(lnk))) {
- if (c) {
- r.open(c->impl_->receiver_options_);
- } else {
- pn_link_open(lnk); // No default for engine
- }
+ r.open(c.receiver_options());
}
} else {
sender s(make_wrapper<sender>(lnk));
delegate_.on_sender_open(s);
if (is_local_unititialised(pn_link_state(lnk))) {
- if (c) {
- s.open(c->impl_->sender_options_);
- } else {
- pn_link_open(lnk); // No default for engine
- }
+ s.open(c.sender_options());
}
}
credit_topup(lnk);
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/src/proton_bits.cpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/src/proton_bits.cpp b/proton-c/bindings/cpp/src/proton_bits.cpp
index 3b459e2..5514ebc 100644
--- a/proton-c/bindings/cpp/src/proton_bits.cpp
+++ b/proton-c/bindings/cpp/src/proton_bits.cpp
@@ -71,6 +71,7 @@ void set_error_condition(const error_condition& e, pn_condition_t *c) {
if (!e.description().empty()) {
pn_condition_set_description(c, e.description().c_str());
}
+ // FIXME aconway 2016-05-09: value ref/value factory fix.
// TODO: This is wrong as it copies the value so doesn't change
// The internals of c
//proton::value v(pn_condition_info(c));
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/src/proton_bits.hpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/src/proton_bits.hpp b/proton-c/bindings/cpp/src/proton_bits.hpp
index 6b4a295..1c96bdb 100644
--- a/proton-c/bindings/cpp/src/proton_bits.hpp
+++ b/proton-c/bindings/cpp/src/proton_bits.hpp
@@ -96,7 +96,7 @@ template <> struct wrapped<transfer> { typedef pn_delivery_t type; };
template <> struct wrapped<tracker> { typedef pn_delivery_t type; };
template <> struct wrapped<delivery> { typedef pn_delivery_t type; };
template <> struct wrapped<error_condition> { typedef pn_condition_t type; };
-template <> struct wrapped<acceptor> { typedef pn_acceptor_t type; };
+template <> struct wrapped<acceptor> { typedef pn_acceptor_t type; }; // TODO aconway 2016-05-13: reactor only
template <> struct wrapped<terminus> { typedef pn_terminus_t type; };
template <> struct wrapped<source> { typedef pn_terminus_t type; };
template <> struct wrapped<target> { typedef pn_terminus_t type; };
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/src/proton_event.hpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/src/proton_event.hpp b/proton-c/bindings/cpp/src/proton_event.hpp
index 2d6b37f..1d68529 100644
--- a/proton-c/bindings/cpp/src/proton_event.hpp
+++ b/proton-c/bindings/cpp/src/proton_event.hpp
@@ -22,6 +22,8 @@
*
*/
+#include "proton/error.hpp"
+
#include "proton/event.h"
namespace proton {
@@ -264,13 +266,13 @@ class proton_event
};
///@}
- proton_event(pn_event_t *ce, class container *c) :
+ proton_event(pn_event_t *ce, class container& cont) :
pn_event_(ce),
- container_(c)
+ container_(cont)
{}
pn_event_t* pn_event() const { return pn_event_; }
- class container* container() const { return container_; }
+ class container& container() const { return container_; }
/// Get type of event
event_type type() const { return event_type(pn_event_type(pn_event_)); }
@@ -279,7 +281,7 @@ class proton_event
private:
pn_event_t *pn_event_;
- class container *container_;
+ class container& container_;
};
}
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1b8450d6/proton-c/bindings/cpp/src/reactor.cpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/src/reactor.cpp b/proton-c/bindings/cpp/src/reactor.cpp
index e34a4fc..20d331c 100644
--- a/proton-c/bindings/cpp/src/reactor.cpp
+++ b/proton-c/bindings/cpp/src/reactor.cpp
@@ -18,8 +18,8 @@
*/
#include "reactor.hpp"
+#include "acceptor.hpp"
-#include "proton/acceptor.hpp"
#include "proton/connection.hpp"
#include "proton/task.hpp"
#include "proton/url.hpp"
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org