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:31 UTC
[2/4] qpid-proton git commit: PROTON-1184: C++ merge APIs for single
and multi-threaded use.
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