PROTON-1481: [C++ binding] Split out general work deferring functions to the work_queue header Add efficient C++11 versions of work factories Reorder defer arguments to be like std::bind Add extra overloads for defer like std::bind (for free functions arbitrary work_queues)
Project: http://git-wip-us.apache.org/repos/asf/qpid-proton/repo Commit: http://git-wip-us.apache.org/repos/asf/qpid-proton/commit/efc899fb Tree: http://git-wip-us.apache.org/repos/asf/qpid-proton/tree/efc899fb Diff: http://git-wip-us.apache.org/repos/asf/qpid-proton/diff/efc899fb Branch: refs/heads/master Commit: efc899fb377611fcf9cf26280a2025c980bb0bf2 Parents: 5dd3f46 Author: Andrew Stitcher <[email protected]> Authored: Mon May 15 01:45:43 2017 -0400 Committer: Andrew Stitcher <[email protected]> Committed: Fri Jul 21 12:50:06 2017 -0400 ---------------------------------------------------------------------- examples/cpp/broker.cpp | 110 ++-------- .../bindings/cpp/include/proton/work_queue.hpp | 210 +++++++++++++++++++ 2 files changed, 222 insertions(+), 98 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/efc899fb/examples/cpp/broker.cpp ---------------------------------------------------------------------- diff --git a/examples/cpp/broker.cpp b/examples/cpp/broker.cpp index 9af60ba..a1ef3a5 100644 --- a/examples/cpp/broker.cpp +++ b/examples/cpp/broker.cpp @@ -70,92 +70,6 @@ // SendMsg(msg) - From a Queue to a Connection (sender) // Unsubscribed() - From a Queue to a Connection (sender) -// Utilities to make injecting member functions palatable in C++03 -template <class T> -struct work0 : public proton::void_function0 { - T& holder_; - void (T::* fn_)(); - - work0(T& h, void (T::* a)()) : - holder_(h), fn_(a) {} - - void operator()() OVERRIDE { - (holder_.*fn_)(); - delete this; - } -}; - -template <class T, class A> -struct work1 : public proton::void_function0 { - T& holder_; - void (T::* fn_)(A); - A a_; - - work1(T& h, void (T::* t)(A), A a) : - holder_(h), fn_(t), a_(a) {} - - void operator()() OVERRIDE { - (holder_.*fn_)(a_); - delete this; - } -}; - -template <class T, class A, class B> -struct work2 : public proton::void_function0 { - T& holder_; - void (T::* fn_)(A, B); - A a_; - B b_; - - work2(T& h, void (T::* t)(A, B), A a, B b) : - holder_(h), fn_(t), a_(a), b_(b) {} - - void operator()() OVERRIDE { - (holder_.*fn_)(a_, b_); - delete this; - } -}; - -template <class T, class A, class B, class C> -struct work3 : public proton::void_function0 { - T& holder_; - void (T::* fn_)(A, B, C); - A a_; - B b_; - C c_; - - work3(T& h, void (T::* t)(A, B, C), A a, B b, C c) : - holder_(h), fn_(t), a_(a), b_(b), c_(c) {} - - void operator()() OVERRIDE { - (holder_.*fn_)(a_, b_, c_); - delete this; - } -}; - -template <class T> -void defer(T* t, void (T::*f)()) { - work0<T>* w = new work0<T>(*t, f); - t->add(*w); -} - -template <class T, class A> -void defer(T* t, void (T::*f)(A), A a) { - work1<T, A>* w = new work1<T, A>(*t, f, a); - t->add(*w); -} - -template <class T, class A, class B> -void defer(T* t, void (T::*f)(A, B), A a, B b) { - work2<T, A, B>* w = new work2<T, A, B>(*t, f, a, b); - t->add(*w); -} - -template <class T, class A, class B, class C> -void defer(T* t, void (T::*f)(A, B, C), A a, B b, C c) { - work3<T, A, B, C>* w = new work3<T, A, B, C>(*t, f, a, b, c); - t->add(*w); -} // Simple debug output bool verbose; @@ -225,7 +139,7 @@ class Queue { DOUT(std::cerr << "(" << current_->second << ") ";); if (current_->second>0) { DOUT(std::cerr << current_->first << " ";); - defer(current_->first, &Sender::sendMsg, messages_.front()); + proton::defer(&Sender::sendMsg, current_->first, messages_.front()); messages_.pop_front(); --current_->second; ++current_; @@ -264,14 +178,14 @@ public: // If we're about to erase the current subscription move on if (current_ != subscriptions_.end() && current_->first==s) ++current_; subscriptions_.erase(s); - defer(s, &Sender::unsubscribed); + proton::defer(&Sender::unsubscribed, s); } }; // We have credit to send a message. void Sender::on_sendable(proton::sender &sender) { if (queue_) { - defer(queue_, &Queue::flow, this, sender.credit()); + proton::defer(&Queue::flow, queue_, this, sender.credit()); } else { pending_credit_ = sender.credit(); } @@ -279,7 +193,7 @@ void Sender::on_sendable(proton::sender &sender) { void Sender::on_sender_close(proton::sender &sender) { if (queue_) { - defer(queue_, &Queue::unsubscribe, this); + proton::defer(&Queue::unsubscribe, queue_, this); } else { // TODO: Is it possible to be closed before we get the queue allocated? // If so, we should have a way to mark the sender deleted, so we can delete @@ -293,12 +207,12 @@ void Sender::boundQueue(Queue* q, std::string qn) { queue_ = q; queue_name_ = qn; - defer(q, &Queue::subscribe, this); + proton::defer(&Queue::subscribe, q, this); sender_.open(proton::sender_options() .source((proton::source_options().address(queue_name_))) .handler(*this)); if (pending_credit_>0) { - defer(queue_, &Queue::flow, this, pending_credit_); + proton::defer(&Queue::flow, queue_, this, pending_credit_); } std::cout << "sending from " << queue_name_ << std::endl; } @@ -323,7 +237,7 @@ class Receiver : public proton::messaging_handler { void queueMsgs() { DOUT(std::cerr << "Receiver: " << this << " queueing " << messages_.size() << " msgs to: " << queue_ << "\n";); while (!messages_.empty()) { - defer(queue_, &Queue::queueMsg, messages_.front()); + proton::defer(&Queue::queueMsg, queue_, messages_.front()); messages_.pop_front(); } } @@ -381,7 +295,7 @@ public: } else { q = i->second; } - defer(&connection, &T::boundQueue, q, qn); + proton::defer(&T::boundQueue, &connection, q, qn); } void findQueueSender(Sender* s, std::string qn) { @@ -411,7 +325,7 @@ public: std::string qn = sender.source().dynamic() ? "" : sender.source().address(); Sender* s = new Sender(sender, senders_); senders_[sender] = s; - defer(&queue_manager_, &QueueManager::findQueueSender, s, qn); + proton::defer(&QueueManager::findQueueSender, &queue_manager_, s, qn); } // A receiver receives messages from a publisher to a queue. @@ -427,7 +341,7 @@ public: DOUT(std::cerr << "ODD - trying to attach to a empty address\n";); } Receiver* r = new Receiver(receiver); - defer(&queue_manager_, &QueueManager::findQueueReceiver, r, qname); + proton::defer(&QueueManager::findQueueReceiver, &queue_manager_, r, qname); } } @@ -438,7 +352,7 @@ public: if (j == senders_.end()) continue; Sender* s = j->second; if (s->queue_) { - defer(s->queue_, &Queue::unsubscribe, s); + proton::defer(&Queue::unsubscribe, s->queue_, s); } senders_.erase(j); } @@ -456,7 +370,7 @@ public: if (j == senders_.end()) continue; Sender* s = j->second; if (s->queue_) { - defer(s->queue_, &Queue::unsubscribe, s); + proton::defer(&Queue::unsubscribe, s->queue_, s); } } delete this; // All done. http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/efc899fb/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 index 7c21710..1937e11 100644 --- a/proton-c/bindings/cpp/include/proton/work_queue.hpp +++ b/proton-c/bindings/cpp/include/proton/work_queue.hpp @@ -117,6 +117,216 @@ class PN_CPP_CLASS_EXTERN work_queue { /// @endcond }; +// Utilities to make injecting functions/member functions palatable in C++03 +// Lots of repetition to handle functions with up to 3 arguments +#if !PN_CPP_HAS_CPP11 +struct work0 : public proton::void_function0 { + void (* fn_)(); + + work0(void (* f)()) : + fn_(f) {} + + void operator()() { + (*fn_)(); + delete this; + } +}; + +template <class A> +struct work1 : public proton::void_function0 { + void (* fn_)(A); + A a_; + + work1(void (* t)(A), A a) : + fn_(t), a_(a) {} + + void operator()() { + (*fn_)(a_); + delete this; + } +}; + +template <class A, class B> +struct work2 : public proton::void_function0 { + void (* fn_)(A, B); + A a_; + B b_; + + work2(void (* t)(A, B), A a, B b) : + fn_(t), a_(a), b_(b) {} + + void operator()() { + (*fn_)(a_, b_); + delete this; + } +}; + +template <class A, class B, class C> +struct work3 : public proton::void_function0 { + void (* fn_)(A, B, C); + A a_; + B b_; + C c_; + + work3(void (* t)(A, B, C), A a, B b, C c) : + fn_(t), a_(a), b_(b), c_(c) {} + + void operator()() { + (*fn_)(a_, b_, c_); + delete this; + } +}; + +template <class T> +struct work_pmf0 : public proton::void_function0 { + T& holder_; + void (T::* fn_)(); + + work_pmf0(void (T::* a)(), T& h) : + holder_(h), fn_(a) {} + + void operator()() { + (holder_.*fn_)(); + delete this; + } +}; + +template <class T, class A> +struct work_pmf1 : public proton::void_function0 { + T& holder_; + void (T::* fn_)(A); + A a_; + + work_pmf1(void (T::* t)(A), T& h, A a) : + holder_(h), fn_(t), a_(a) {} + + void operator()() { + (holder_.*fn_)(a_); + delete this; + } +}; + +template <class T, class A, class B> +struct work_pmf2 : public proton::void_function0 { + T& holder_; + void (T::* fn_)(A, B); + A a_; + B b_; + + work_pmf2(void (T::* t)(A, B), T& h, A a, B b) : + holder_(h), fn_(t), a_(a), b_(b) {} + + void operator()() { + (holder_.*fn_)(a_, b_); + delete this; + } +}; + +template <class T, class A, class B, class C> +struct work_pmf3 : public proton::void_function0 { + T& holder_; + void (T::* fn_)(A, B, C); + A a_; + B b_; + C c_; + + work_pmf3(void (T::* t)(A, B, C), T& h, A a, B b, C c) : + holder_(h), fn_(t), a_(a), b_(b), c_(c) {} + + void operator()() { + (holder_.*fn_)(a_, b_, c_); + delete this; + } +}; + +/// This version of proton::defer defers calling an object's member function to the object's work queue +template <class T> +void defer(void (T::*f)(), T* t) { + work_pmf0<T>* w = new work_pmf0<T>(f, *t); + t->add(*w); +} + +template <class T, class A> +void defer(void (T::*f)(A), T* t, A a) { + work_pmf1<T, A>* w = new work_pmf1<T, A>(f, *t, a); + t->add(*w); +} + +template <class T, class A, class B> +void defer(void (T::*f)(A, B), T* t, A a, B b) { + work_pmf2<T, A, B>* w = new work_pmf2<T, A, B>(f, *t, a, b); + t->add(*w); +} + +template <class T, class A, class B, class C> +void defer(void (T::*f)(A, B, C), T* t, A a, B b, C c) { + work_pmf3<T, A, B, C>* w = new work_pmf3<T, A, B, C>(f, *t, a, b, c); + t->add(*w); +} + +/// This version of proton::defer defers calling a member function to an arbitrary work queue +template <class T, class U> +void defer(U* wq, void (T::*f)(), T* t) { + work_pmf0<T>* w = new work_pmf0<T>(f, *t); + wq->add(*w); +} + +template <class T, class U, class A> +void defer(U* wq, void (T::*f)(A), T* t, A a) { + work_pmf1<T, A>* w = new work_pmf1<T, A>(f, *t, a); + wq->add(*w); +} + +template <class T, class U, class A, class B> +void defer(U* wq, void (T::*f)(A, B), T* t, A a, B b) { + work_pmf2<T, A, B>* w = new work_pmf2<T, A, B>(f, *t, a, b); + wq->add(*w); +} + +template <class T, class U, class A, class B, class C> +void defer(U* wq, void (T::*f)(A, B, C), T* t, A a, B b, C c) { + work_pmf3<T, A, B, C>* w = new work_pmf3<T, A, B, C>(f, *t, a, b, c); + wq->add(*w); +} + +/// This version of proton::defer defers calling a free function to an arbitrary work queue +template <class U> +void defer(U* wq, void (*f)()) { + work0* w = new work0(f); + wq->add(*w); +} + +template <class U, class A> +void defer(U* wq, void (*f)(A), A a) { + work1<A>* w = new work1<A>(f, a); + wq->add(*w); +} + +template <class U, class A, class B> +void defer(U* wq, void (*f)(A, B), A a, B b) { + work2<A, B>* w = new work2<A, B>(f, a, b); + wq->add(*w); +} + +template <class U, class A, class B, class C> +void defer(U* wq, void (*f)(A, B, C), A a, B b, C c) { + work3<A, B, C>* w = new work3<A, B, C>(f, a, b, c); + wq->add(*w); +} +#else +// The C++11 version is *much* simpler and even so more general! +// These 2 definitions encompass everything in the C++03 section +template <class T, class... Rest> +void defer(void(T::*f)(Rest...), T* t, Rest... r) { + t->add(std::bind(f, t, r...)); +} + +template <class U, class... Rest> +void defer(U* wq, Rest&&... r) { + wq->add(std::bind(std::forward<Rest>(r)...)); +} +#endif + } // proton #endif // PROTON_WORK_QUEUE_HPP --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
