Add coordinator class

DreamPearl · DreamPearl · commit f8eff19f81a4 · 2025-03-17T23:24:10.000+05:30
diff --git a/cpp/examples/tx_recv.cpp b/cpp/examples/tx_recv.cpp
@@ -46,7 +46,7 @@ class tx_recv : public proton::messaging_handler, proton::transaction_handler {
     int committed = 0;
 
     proton::session session;
-    proton::transaction transaction;
+    // proton::transaction transaction;
   public:
     tx_recv(const std::string &s, int c, int b):
         url(s), expected(c), batch_size(b) {}
@@ -62,38 +62,38 @@ class tx_recv : public proton::messaging_handler, proton::transaction_handler {
         std::cout << "    [on_session_open] declare_txn ended..." << std::endl;
     }
 
-    void on_transaction_declare_failed(proton::transaction) {}
-    void on_transaction_commit_failed(proton::transaction t) {
+    void on_transaction_declare_failed(proton::session) {}
+    void on_transaction_commit_failed(proton::session s) {
         std::cout << "Transaction Commit Failed" << std::endl;
-        t.connection().close();
+        s.connection().close();
         exit(-1);
     }
 
-    void on_transaction_declared(proton::transaction t) override {
-        std::cout << "[on_transaction_declared] txn called " << (&t)
+    void on_transaction_declared(proton::session s) override {
+        std::cout << "[on_transaction_declared] txn called " << (&s)
                   << std::endl;
-        std::cout << "[on_transaction_declared] txn is_empty " << (t.is_empty())
-                  << "\t" << transaction.is_empty() << std::endl;
+        // std::cout << "[on_transaction_declared] txn is_empty " << (t.is_empty())
+        //           << "\t" << transaction.is_empty() << std::endl;
         receiver.add_credit(batch_size); 
-        transaction = t;
+        // transaction = t;
     }
 
     void on_message(proton::delivery &d, proton::message &msg) override {
         std::cout<<"# MESSAGE: " << msg.id() <<": "  << msg.body() << std::endl;
-        transaction.accept(d);
+        session.txn_accept(d);
         current_batch += 1;
         if(current_batch == batch_size) {
-            transaction = proton::transaction(); // null
+            // transaction = proton::transaction(); // null
         }
     }
 
-    void on_transaction_committed(proton::transaction t) override {
+    void on_transaction_committed(proton::session s) override {
         committed += current_batch;
         current_batch = 0;
         std::cout<<"    [OnTxnCommitted] Committed:"<< committed<< std::endl;
         if(committed == expected) {
             std::cout << "All messages committed" << std::endl;
-            t.connection().close();
+            s.connection().close();
         }
         else {
             session.declare_transaction(*this);
diff --git a/cpp/examples/tx_send.cpp b/cpp/examples/tx_send.cpp
@@ -89,7 +89,7 @@ class tx_send : public proton::messaging_handler, proton::transaction_handler {
 
     void send(proton::sender &s) {
         static int unique_id = 10000;
-        while (!session.txn_is_empty() && sender.credit() &&
+        while (session.txn_is_declared() && sender.credit() &&
                (committed + current_batch) < total) {
             proton::message msg;
             std::map<std::string, int> m;
diff --git a/cpp/include/proton/fwd.hpp b/cpp/include/proton/fwd.hpp
@@ -52,6 +52,7 @@ class sender_options;
 class session;
 class session_options;
 class source_options;
+class coordinator_options;
 class ssl;
 class target_options;
 class tracker;
diff --git a/cpp/include/proton/sender_options.hpp b/cpp/include/proton/sender_options.hpp
@@ -94,6 +94,9 @@ class sender_options {
     /// Options for the receiver node of the receiver.
     PN_CPP_EXTERN sender_options& target(const target_options&);
 
+    /// Options for the coordinator node of the receiver.
+    PN_CPP_EXTERN sender_options& coordinator(const coordinator_options&);
+
     /// Set the link name. If not set a unique name is generated.
     PN_CPP_EXTERN sender_options& name(const std::string& name);
 
diff --git a/cpp/include/proton/session.hpp b/cpp/include/proton/session.hpp
@@ -117,6 +117,7 @@ PN_CPP_CLASS_EXTERN session : public internal::object<pn_session_t>, public endp
     // PN_CPP_EXTERN transaction();
     // PN_CPP_EXTERN ~transaction();
     PN_CPP_EXTERN bool txn_is_empty();
+    PN_CPP_EXTERN bool txn_is_declared();
     PN_CPP_EXTERN void txn_commit();
     PN_CPP_EXTERN void txn_abort();
     PN_CPP_EXTERN void txn_declare();
@@ -156,6 +157,7 @@ class transaction_impl {
     proton::tracker _declare;
     proton::tracker _discharge;
     bool failed = false;
+    bool is_declared = false;
     std::vector<proton::tracker> pending;
 
     void commit();
diff --git a/cpp/include/proton/target.hpp b/cpp/include/proton/target.hpp
@@ -65,6 +65,29 @@ class target : public terminus {
     /// @endcond
 };
 
+
+/// TODO: A point of coordinator for messages.
+///
+class coordinator : public terminus {
+  public:
+    /// Create an empty coordinator.
+    coordinator() = default;
+
+    /// The address of the coordinator.
+    PN_CPP_EXTERN std::string address() const;
+  private:
+    coordinator(pn_terminus_t* t);
+    coordinator(const sender&);
+    coordinator(const receiver&);
+
+
+    /// @cond INTERNAL
+  friend class proton::internal::factory<coordinator>;
+  friend class sender;
+  friend class receiver;
+    /// @endcond
+};
+
 } // proton
 
 #endif // PROTON_TARGET_HPP
diff --git a/cpp/include/proton/target_options.hpp b/cpp/include/proton/target_options.hpp
@@ -103,6 +103,39 @@ class target_options {
     /// @endcond
 };
 
+class coordinator_options {
+  public:
+    /// Create an empty set of options.
+    PN_CPP_EXTERN coordinator_options();
+
+    /// Copy options.
+    PN_CPP_EXTERN coordinator_options(const coordinator_options&);
+
+    PN_CPP_EXTERN ~coordinator_options();
+
+    /// Copy options.
+    PN_CPP_EXTERN coordinator_options& operator=(const coordinator_options&);
+
+    /// Set the address for the coordinator.  It is unset by default.  The
+    /// address is ignored if dynamic() is true.
+    PN_CPP_EXTERN coordinator_options& address(const std::string& addr);
+
+    /// **Unsettled API** Extension capabilities that are supported/requested
+    PN_CPP_EXTERN coordinator_options& capabilities(const std::vector<symbol>&);
+
+  private:
+    void apply(coordinator&) const;
+
+    class impl;
+    std::unique_ptr<impl> impl_;
+
+    /// @cond INTERNAL
+  friend class coordinator;
+  friend class sender_options;
+  friend class receiver_options;
+    /// @endcond
+};
+
 } // proton
 
 #endif // PROTON_TARGET_OPTIONS_HPP
diff --git a/cpp/include/proton/terminus.hpp b/cpp/include/proton/terminus.hpp
@@ -122,6 +122,7 @@ class terminus {
   friend class internal::factory<terminus>;
   friend class source;
   friend class target;
+  friend class coordinator;
     /// @endcond
 };
 
diff --git a/cpp/src/node_options.cpp b/cpp/src/node_options.cpp
@@ -209,5 +209,36 @@ target_options& target_options::type(int t) { impl_->type = t; return *this;}
 void target_options::apply(target& s) const { impl_->apply(s); }
 
 
+class coordinator_options::impl {
+  public:
+    option<std::string> address;
+    option<std::vector<symbol> > capabilities;
+
+    void apply(coordinator& c) {
+        if(address.set) {
+            pn_terminus_set_address(unwrap(c), address.value.c_str());
+        }
+        if (capabilities.set) {
+            value(pn_terminus_capabilities(unwrap(c))) = capabilities.value;
+        }
+        pn_terminus_set_type(unwrap(c), pn_terminus_type_t(PN_COORDINATOR)); ;
+    }
+};
+
+coordinator_options::coordinator_options() : impl_(new impl()) {}
+coordinator_options::coordinator_options(const coordinator_options& x) : impl_(new impl()) {
+    *this = x;
+}
+coordinator_options::~coordinator_options() = default;
+
+coordinator_options& coordinator_options::operator=(const coordinator_options& x) {
+    *impl_ = *x.impl_;
+    return *this;
+}
+
+coordinator_options& coordinator_options::address(const std::string &addr) { impl_->address = addr; return *this; }
+coordinator_options& coordinator_options::capabilities(const std::vector<symbol>& c) { impl_->capabilities = c; return *this; }
+void coordinator_options::apply(coordinator& c) const { impl_->apply(c); }
+
 
 } // namespace proton
diff --git a/cpp/src/proton_bits.hpp b/cpp/src/proton_bits.hpp
@@ -69,6 +69,7 @@ class acceptor;
 class terminus;
 class source;
 class target;
+class coordinator;
 class reactor;
 class messaging_handler;
 
@@ -109,6 +110,7 @@ template <> struct wrapped<error_condition> { typedef pn_condition_t type; };
 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; };
+template <> struct wrapped<coordinator> { typedef pn_terminus_t type; };
 
 template <class T> struct wrapper {};
 template <> struct wrapper<pn_data_t> { typedef internal::data type; };
diff --git a/cpp/src/sender_options.cpp b/cpp/src/sender_options.cpp
@@ -66,6 +66,7 @@ class sender_options::impl {
     option<bool> auto_settle;
     option<source_options> source;
     option<target_options> target;
+    option<coordinator_options> coordinator;
     option<std::string> name;
     option<std::map<symbol, value> > properties;
 
@@ -82,6 +83,10 @@ class sender_options::impl {
                 proton::target local_t(make_wrapper<proton::target>(pn_link_target(unwrap(s))));
                 target.value.apply(local_t);
             }
+            if (coordinator.set) {
+                proton::coordinator local_t(make_wrapper<proton::coordinator>(pn_link_target(unwrap(s))));
+                coordinator.value.apply(local_t);
+            }
             if (properties.set) {
                 value(pn_link_properties(unwrap(s))) = properties.value;
             }
@@ -118,6 +123,7 @@ sender_options& sender_options::delivery_mode(proton::delivery_mode m) {impl_->d
 sender_options& sender_options::auto_settle(bool b) {impl_->auto_settle = b; return *this; }
 sender_options& sender_options::source(const source_options &s) {impl_->source = s; return *this; }
 sender_options& sender_options::target(const target_options &s) {impl_->target = s; return *this; }
+sender_options& sender_options::coordinator(const coordinator_options &s) {impl_->coordinator = s; return *this; }
 sender_options& sender_options::name(const std::string &s) {impl_->name = s; return *this; }
 sender_options& sender_options::properties(const std::map<symbol, value> &props) { impl_->properties = props; return *this; }
 
diff --git a/cpp/src/session.cpp b/cpp/src/session.cpp
@@ -182,14 +182,20 @@ void session::declare_transaction(proton::transaction_handler &handler, bool set
         }
     };
 
-    proton::target_options t;
+    // proton::target_options t;
+    // std::vector<symbol> cap = {proton::symbol("amqp:local-transactions")};
+    // t.capabilities(cap);
+    // t.type(PN_COORDINATOR);
+
+    proton::coordinator_options co;
     std::vector<symbol> cap = {proton::symbol("amqp:local-transactions")};
-    t.capabilities(cap);
-    t.type(PN_COORDINATOR);
+    co.capabilities(cap);
+
 
     proton::sender_options so;
     so.name("txn-ctrl");
-    so.target(t);
+    so.coordinator(co);
+
     static InternalTransactionHandler internal_handler; // internal_handler going out of scope. Fix it
     so.handler(internal_handler);
     std::cout<<"    [declare_transaction] txn-name sender open with handler: " << &internal_handler << std::endl;
@@ -222,6 +228,7 @@ void session::txn_commit() {session_context::get(pn_object())._txn_impl->commit(
 void session::txn_abort() { session_context::get(pn_object())._txn_impl->abort(); }
 void session::txn_declare() { session_context::get(pn_object())._txn_impl->declare(); }
 bool session::txn_is_empty() { return session_context::get(pn_object())._txn_impl == NULL; }
+bool session::txn_is_declared() { return (!txn_is_empty()) && session_context::get(pn_object())._txn_impl->is_declared; }
 void session::txn_accept(delivery &t) { return session_context::get(pn_object())._txn_impl->accept(t); }
 proton::tracker session::txn_send(proton::sender s, proton::message msg) {
     return session_context::get(pn_object())._txn_impl->send(s, msg);
@@ -244,17 +251,19 @@ void session::txn_handle_outcome(proton::tracker t) {
 transaction_impl::transaction_impl(proton::sender &_txn_ctrl,
                                    proton::transaction_handler &_handler,
                                    bool _settle_before_discharge)
-    : txn_ctrl(_txn_ctrl), handler(&_handler) {
+    : txn_ctrl(_txn_ctrl), handler(&_handler), is_declared(false) {
     // bool settle_before_discharge = _settle_before_discharge;
     declare();
 }
 
 void transaction_impl::commit() {
     discharge(false);
+    is_declared = false;
 }
 
 void transaction_impl::abort() {
     discharge(true);
+    is_declared = false;
 }
 
 void transaction_impl::declare() {
@@ -374,6 +383,7 @@ void transaction_impl::handle_outcome(proton::tracker t) {
             std::cout << "    transaction_impl: handle_outcome.. txn_declared "
                          "got txnid:: "
                       << vd[0] << std::endl;
+            is_declared = true;
             handler->on_transaction_declared(t.session());
         } else if (pn_disposition_is_failed(disposition)) {
             std::cout << "    transaction_impl: handle_outcome.. "
diff --git a/cpp/src/target.cpp b/cpp/src/target.cpp
@@ -39,4 +39,31 @@ std::string target::address() const {
     return str(pn_terminus_get_address(authoritative));
 }
 
+
+// Set parent_ non-null when the local terminus is authoritative and may need to be looked up.
+coordinator::coordinator(pn_terminus_t *t) : terminus(make_wrapper(t))
+{}
+
+coordinator::coordinator(const sender& snd) : terminus(make_wrapper(pn_link_remote_target(unwrap(snd)))) {}
+coordinator::coordinator(const receiver& rcv) : terminus(make_wrapper(pn_link_remote_target(unwrap(rcv)))) { parent_ = unwrap(rcv); }
+// coordinator::coordinator(const sender& snd) :
+//     terminus(make_wrapper(pn_link_remote_source(unwrap(snd)))),
+//     filters_(pn_terminus_filter(object_))
+// {
+//     parent_ = unwrap(snd);
+// }
+
+// coordinator::coordinator(const receiver& rcv) :
+//     terminus(make_wrapper(pn_link_remote_source(unwrap(rcv)))),
+//     filters_(pn_terminus_filter(object_))
+// {}
+
+std::string coordinator::address() const {
+    pn_terminus_t *authoritative = object_;
+    if (parent_ && pn_terminus_is_dynamic(object_))
+        authoritative = pn_link_source(parent_);
+    return str(pn_terminus_get_address(authoritative));
+}
+
+
 }
