Make example compile

DreamPearl · DreamPearl · commit ed624cd0d76c · 2024-11-20T14:05:04.000+05:30
diff --git a/cpp/examples/CMakeLists.txt b/cpp/examples/CMakeLists.txt
@@ -60,7 +60,8 @@ foreach(example
     scheduled_send
     service_bus
     multithreaded_client
-    multithreaded_client_flow_control)
+    multithreaded_client_flow_control
+    tx_send)
   add_executable(${example} ${example}.cpp)
   target_link_libraries(${example} Proton::cpp Threads::Threads)
 endforeach()
diff --git a/cpp/examples/tx_send.cpp b/cpp/examples/tx_send.cpp
@@ -44,8 +44,8 @@ class tx_send : public proton::messaging_handler, proton::transaction_handler {
     int committed = 0;
     int confirmed = 0;
     proton::container *container;
-    proton::transaction_handler transaction_handler;
-    proton::transaction *transaction;
+    // proton::transaction_handler transaction_handler;
+    proton::transaction transaction;
     proton::connection connection;
   public:
     tx_send(const std::string &s, int c, int b):
@@ -55,34 +55,48 @@ class tx_send : public proton::messaging_handler, proton::transaction_handler {
         container = &c; // TODO: Fix error
         sender = c.open_sender(url);
         connection = sender.connection();
-        transaction = NULL;
-        c.declare_transaction(connection, transaction_handler);
+        std::cout << "    [on_container_start] declare_txn started..." << std::endl;
+        transaction = c.declare_transaction(connection, *this);
+        std::cout << "    [on_container_start] completed!! txn: " << &transaction << std::endl;
     }
 
+    void on_transaction_aborted(proton::transaction&) {}
+    void on_transaction_declare_failed(proton::transaction &) {}
+    void on_transaction_commit_failed(proton::transaction&) {}
+
+
     void on_transaction_declared(proton::transaction &t) override {
-        transaction = &t;
-        send();
+        std::cout<<"[on_transaction_declared] txn: "<<(&transaction)<< " new_txn: "<<(&t)<<std::endl;
+        connection.close();
+        // transaction = &t;
+        // ASSUME: THIS FUNCTION DOESN"T WORK
+        // send();
     }
 
     void on_sendable(proton::sender &s) override {
-        send();
+        // send();
+        // std::cout<<"    [OnSendable] transaction: "<< &transaction << std::endl;
+        // send(s);
     }
 
-    void send() {
+    void send(proton::sender &s) {
         // TODO: Add more condition in while loop
-        while (transaction && sender.credit() && (committed + current_batch) < total)
+        // transaction != null
+        while ( sender.credit() && (committed + current_batch) < total)
         {
             proton::message msg;
             std::map<std::string, int> m;
             m["sequence"] = committed + current_batch;
 
             msg.id(committed + current_batch + 1);
             msg.body(m);
+            transaction.send(sender, msg);
             current_batch += 1;
             if(current_batch == batch_size)
             {
-                transaction->commit();
-                transaction = NULL;
+                transaction.commit();
+                // WE DON"T CARE ANY MORE FOR NOW
+                // transaction = NULL;
             }
         }
 
@@ -94,13 +108,14 @@ class tx_send : public proton::messaging_handler, proton::transaction_handler {
 
     void on_transaction_committed(proton::transaction &t) override {
         committed += current_batch;
+        std::cout<<"    [OnTxnCommitted] Committed:"<< committed<< std::endl;
         if(committed == total) {
             std::cout << "All messages committed";
-            connection.close();
+            // connection.close();
         }
         else {
-            current_batch = 0;
-            container->declare_transaction(connection, transaction_handler);
+            // current_batch = 0;
+            // container->declare_transaction(connection, transaction_handler);
         }
     }
 
@@ -112,7 +127,7 @@ class tx_send : public proton::messaging_handler, proton::transaction_handler {
 
 int main(int argc, char **argv) {
     std::string address("127.0.0.1:5672/examples");
-    int message_count = 100;
+    int message_count = 10;
     int batch_size = 10;
     example::options opts(argc, argv);
 
diff --git a/cpp/include/proton/tracker.hpp b/cpp/include/proton/tracker.hpp
@@ -26,7 +26,6 @@
 #include "./internal/export.hpp"
 #include "./transfer.hpp"
 #include "./messaging_handler.hpp"
-#include "./transaction.hpp"
 
 /// @file
 /// @copybrief proton::tracker
diff --git a/cpp/include/proton/transaction.hpp b/cpp/include/proton/transaction.hpp
@@ -44,6 +44,7 @@ PN_CPP_CLASS_EXTERN transaction {
     PN_CPP_EXTERN virtual void commit();
     PN_CPP_EXTERN virtual void abort();
     PN_CPP_EXTERN virtual void declare();
+    PN_CPP_EXTERN virtual void handle_outcome(proton::tracker);
     PN_CPP_EXTERN virtual proton::tracker send(proton::sender s, proton::message msg);
 };
 
diff --git a/cpp/include/proton/transfer.hpp b/cpp/include/proton/transfer.hpp
@@ -78,9 +78,9 @@ class transfer : public internal::object<pn_delivery_t> {
     PN_CPP_EXTERN bool settled() const;
 
    // Set transaction
-    PN_CPP_EXTERN void transaction(transaction& t);
+    PN_CPP_EXTERN void transaction(transaction t);
 
-    PN_CPP_EXTERN class transaction* transaction() const;
+    PN_CPP_EXTERN class transaction transaction() const;
 
     /// Set user data on this transfer.
     PN_CPP_EXTERN void user_data(void* user_data) const;
@@ -93,9 +93,9 @@ class transfer : public internal::object<pn_delivery_t> {
     /// @endcond
 };
 
-/// Human-readalbe name of the transfer::state
+/// Human-readable name of the transfer::state
 PN_CPP_EXTERN std::string to_string(enum transfer::state);
-/// Human-readalbe name of the transfer::state
+/// Human-readable name of the transfer::state
 PN_CPP_EXTERN std::ostream& operator<<(std::ostream&, const enum transfer::state);
 
 } // proton
diff --git a/cpp/src/contexts.hpp b/cpp/src/contexts.hpp
@@ -25,6 +25,7 @@
 #include "reconnect_options_impl.hpp"
 
 #include "proton/work_queue.hpp"
+#include "proton/transaction.hpp"
 #include "proton/message.hpp"
 
 #include "proton/object.h"
@@ -162,7 +163,7 @@ class transfer_context : public context {
     transfer_context() : user_data_(nullptr) {}
     static transfer_context& get(pn_delivery_t* s);
 
-    transaction* transaction_;
+    std::unique_ptr<transaction> transaction_;
     void* user_data_;
 };
 
diff --git a/cpp/src/messaging_adapter.cpp b/cpp/src/messaging_adapter.cpp
@@ -30,6 +30,7 @@
 #include "proton/receiver_options.hpp"
 #include "proton/sender.hpp"
 #include "proton/sender_options.hpp"
+#include "proton/target_options.hpp"
 #include "proton/session.hpp"
 #include "proton/tracker.hpp"
 #include "proton/transport.hpp"
@@ -69,6 +70,11 @@ void on_link_flow(messaging_handler& handler, pn_event_t* event) {
     // TODO: process session flow data, if no link-specific data, just return.
     if (!lnk) return;
     int state = pn_link_state(lnk);
+    if (pn_terminus_get_type(pn_link_remote_target(lnk))==PN_COORDINATOR) {
+        std::cout << "    on_link_flow, type: PN_COORDINATOR"  << std::endl;
+        return;
+
+    }
     if ((state&PN_LOCAL_ACTIVE) && (state&PN_REMOTE_ACTIVE)) {
         link_context& lctx = link_context::get(lnk);
         if (pn_link_is_sender(lnk)) {
@@ -115,8 +121,14 @@ void on_delivery(messaging_handler& handler, pn_event_t* event) {
     pn_delivery_t *dlv = pn_event_delivery(event);
     link_context& lctx = link_context::get(lnk);
     Tracing& ot = Tracing::getTracing();
+    if (pn_terminus_get_type(pn_link_remote_target(lnk))==PN_COORDINATOR) {
+        std::cout<< "    on_delivery: COOORINDATOR.. " << &handler << std::endl;
+        tracker t(make_wrapper<tracker>(dlv));
+        std::cout<< "    on_delivery: COOORINDATOR.. tracker" << &t << std::endl;
+        handler.on_tracker_settle(t);
+    }
 
-    if (pn_link_is_receiver(lnk)) {
+    else if (pn_link_is_receiver(lnk)) {
         delivery d(make_wrapper<delivery>(dlv));
         if (pn_delivery_aborted(dlv)) {
             pn_delivery_settle(dlv);
@@ -274,23 +286,73 @@ void on_link_local_open(messaging_handler& handler, pn_event_t* event) {
 
 void on_link_remote_open(messaging_handler& handler, pn_event_t* event) {
     auto lnk = pn_event_link(event);
-    // Currently don't implement (transaction) coordinator
+    int type = pn_terminus_get_type(pn_link_remote_target(lnk));
+    std::cout << "    on_link_remote_open, type:" << type << std::endl;
     if (pn_terminus_get_type(pn_link_remote_target(lnk))==PN_COORDINATOR) {
-      auto error = pn_link_condition(lnk);
-      pn_condition_set_name(error, "amqp:not-implemented");
-      pn_link_close(lnk);
+      auto cond = pn_link_condition(lnk);
+      if (pn_condition_is_set(cond)) {
+            std::cout<<"    Got condition on_link_remote_open(.PN_COORDINATOR): "
+                << pn_event_type_name(pn_event_type(event)) << " "
+                << pn_condition_get_name(cond) << " "
+                << pn_condition_get_description(cond) << std::endl;
+
+            pn_condition_set_name(cond, "amqp:on_link_remote_open:FAILED");
+            pn_link_close(lnk);
+            return;
+        }
+        std::cout<<"    IN on_link_remote_open(.PN_COORDINATOR) success " << std::endl;
+
+        // WHY???
+        // pn_terminus_copy(pn_link_source(lnk), pn_link_remote_source(lnk));
+        // pn_terminus_copy(pn_link_target(lnk), pn_link_remote_target(lnk));
+
+        // We need a new class?
+        // auto coordinator = pn_link_remote_target(lnk);
+
+
+            // proton::target_options to;
+            // std::vector<symbol> cap = {proton::symbol("amqp:local-transactions")};
+            // to.capabilities(cap);
+            // to.type(PN_COORDINATOR);
+
+            // proton::receiver_options ro;
+            // ro.name("txn-ctrl");
+            // ro.target(to);
+            // ro.handler(handler);
+            // receiver r(make_wrapper<receiver>(lnk));
+
+            // proton::receiver rcv = r.connection().open_receiver("does not matter", ro);
+              std::cout<<"    IN on_link_remote_open(.PN_COORDINATOR) have handler " << &handler << std::endl;
+
+            // handler.on_receiver_open(rcv);
+        // credit_topup(lnk);
+
+    // pn_delivery_t *dlv = pn_event_delivery(event);
+    //         tracker t(make_wrapper<tracker>(dlv));
+
+    //     // sender s(make_wrapper<sender>(lnk));
+    //     handler.on_tracker_settle(t);
+    // TODO: find what to do...
+        // HAHA.. treating coordinator like sender...
+    //   sender s(make_wrapper<sender>(lnk));
+    //   handler.on_sender_open(s);
+    // pn_link_close(lnk);
       return;
     }
     if (pn_link_state(lnk) & PN_LOCAL_UNINIT) { // Incoming link
         // Copy source and target from remote end.
+          std::cout<<"    Inside on_link_remote_open() .. PN_LOCAL_UNINIT " << std::endl;
+
         pn_terminus_copy(pn_link_source(lnk), pn_link_remote_source(lnk));
         pn_terminus_copy(pn_link_target(lnk), pn_link_remote_target(lnk));
     }
     if (pn_link_is_receiver(lnk)) {
+          std::cout<<"    Inside on_link_remote_open() .. pn_link_is_receiver " << std::endl;
       receiver r(make_wrapper<receiver>(lnk));
       handler.on_receiver_open(r);
       credit_topup(lnk);
     } else {
+          std::cout<<"    Inside on_link_remote_open() .. sender " << std::endl;
       sender s(make_wrapper<sender>(lnk));
       handler.on_sender_open(s);
     }
diff --git a/cpp/src/proactor_container_impl.cpp b/cpp/src/proactor_container_impl.cpp
@@ -34,6 +34,7 @@
 #include "proton/listener.h"
 #include "proton/proactor.h"
 #include "proton/transport.h"
+#include "proton/transaction.hpp"
 
 #include "contexts.hpp"
 #include "messaging_adapter.hpp"
@@ -49,7 +50,7 @@
 #include <random>
 
 // XXXX: Debug
-//#include <iostream>
+#include <iostream>
 
 namespace proton {
 
@@ -868,9 +869,11 @@ transaction container::impl::declare_transaction(proton::connection conn, proton
     class InternalTransactionHandler : public proton::messaging_handler {
         // TODO: auto_settle
         void on_tracker_settle(proton::tracker &t) override {
-            if(t.transaction()) {
-                //t.transaction()->handle_outcome(t);
-            }
+            std::cout<<"    [InternalTransactionHandler][on_tracker_settle] called with tracker.txn"
+                 << std::endl;
+            // if(t.transaction()) {
+                t.transaction().handle_outcome(t);
+            // }
         }
 
         // TODO: Add on_unhandled function
@@ -882,23 +885,28 @@ transaction container::impl::declare_transaction(proton::connection conn, proton
     proton::target_options t;
     std::vector<symbol> cap = {proton::symbol("amqp:local-transactions")};
     t.capabilities(cap);
-    // Type PN_COORDINATOR value is 3. It is a special target identifying a transaction coordinator.
-    // TODO: Change the type from int to enum.
-    t.type(3);
+    t.type(PN_COORDINATOR);
 
     proton::sender_options so;
     so.name("txn-ctrl");
     // Todo: Check the value, Or by deafult null?
     //so.source() ?
     so.target(t);
-    InternalTransactionHandler internal_handler; // internal_handler going out of scope. Fix it
+    // TODO: FIX STATIC
+    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;
+
     proton::sender s = conn.open_sender("does not matter", so);
 
     settle_before_discharge = false;
 
-    return mk_transaction_impl(s, handler, settle_before_discharge);
+    std::cout<<"    [declare_transaction] calling mk_transaction_impl" << std::endl;
+
+    auto txn = mk_transaction_impl(s, handler, settle_before_discharge);
+    std::cout<<"    [declare_transaction] txn address:" << &txn << std::endl;
 
+    return txn;
 }
 
 }
diff --git a/cpp/src/transaction.cpp b/cpp/src/transaction.cpp
diff --git a/cpp/src/transfer.cpp b/cpp/src/transfer.cpp
