Add a test file and few improvements

DreamPearl · DreamPearl · commit f04bc4358162 · 2025-10-30T18:44:18.000+05:30
diff --git a/cpp/CMakeLists.txt b/cpp/CMakeLists.txt
@@ -120,7 +120,6 @@ set(qpid-proton-cpp-source
   src/terminus.cpp
   src/timestamp.cpp
   src/tracker.cpp
-  src/transaction_handler.cpp
   src/transfer.cpp
   src/transport.cpp
   src/type_id.cpp
diff --git a/cpp/examples/tx_recv.cpp b/cpp/examples/tx_recv.cpp
@@ -29,7 +29,6 @@
 #include <proton/messaging_handler.hpp>
 #include <proton/receiver_options.hpp>
 #include <proton/source.hpp>
-#include <proton/transaction_handler.hpp>
 #include <proton/types.hpp>
 
 #include <iostream>
@@ -40,7 +39,7 @@
 #include <chrono>
 #include <thread>
 
-class tx_recv : public proton::messaging_handler, proton::transaction_handler {
+class tx_recv : public proton::messaging_handler {
   private:
     proton::receiver receiver;
     proton::session session;
@@ -68,23 +67,23 @@ class tx_recv : public proton::messaging_handler, proton::transaction_handler {
     }
 
     void on_session_open(proton::session &s) override {
-        std::cout << "New session is open" << std::endl;
-        s.transaction_declare(*this);
-        session = s;
+        if(!s.transaction_is_declared()) {
+            std::cout << "New session is open" << std::endl;
+            s.transaction_declare(*this);
+            session = s;
+        } else {
+            std::cout << "Transaction is declared: " << s.transaction_id() << std::endl;
+            receiver.add_credit(batch_size);
+        }
     }
 
-    void on_transaction_declare_failed(proton::session s) override {
-        std::cout << "Transaction declarion failed" << std::endl;
+    void on_session_error(proton::session &s) override {
+        std::cout << "Session error: " << s.error().what() << std::endl;
         s.connection().close();
         exit(-1);
     }
 
-    void on_transaction_declared(proton::session s) override {
-        std::cout << "Transaction is declared: " << s.transaction_id() << std::endl;
-        receiver.add_credit(batch_size);
-    }
-
-    void on_transaction_committed(proton::session s) override {
+    void on_session_transaction_committed(proton::session &s) override {
         std::cout << "Transaction commited" << std::endl;
         received += current_batch;
         current_batch = 0;
@@ -98,7 +97,7 @@ class tx_recv : public proton::messaging_handler, proton::transaction_handler {
         }
     }
 
-    void on_transaction_aborted(proton::session s) override {
+    void on_session_transaction_aborted(proton::session &s) override {
         std::cout << "Transaction aborted!" << std::endl;
         std::cout << "Re-delaring transaction now..." << std::endl;
         current_batch = 0;
diff --git a/cpp/examples/tx_send.cpp b/cpp/examples/tx_send.cpp
@@ -28,7 +28,6 @@
 #include <proton/messaging_handler.hpp>
 #include <proton/sender_options.hpp>
 #include <proton/types.hpp>
-#include <proton/transaction_handler.hpp>
 
 #include <iostream>
 #include <map>
@@ -38,7 +37,7 @@
 #include <chrono>
 #include <thread>
 
-class tx_send : public proton::messaging_handler, proton::transaction_handler {
+class tx_send : public proton::messaging_handler {
   private:
     proton::sender sender;
     std::string conn_url_;
@@ -65,27 +64,27 @@ class tx_send : public proton::messaging_handler, proton::transaction_handler {
     }
 
     void on_session_open(proton::session& s) override {
-        std::cout << "New session is open, declaring transaction now..." << std::endl;
-        s.transaction_declare(*this);
+        if(!s.transaction_is_declared()) {
+            std::cout << "New session is open, declaring transaction now..." << std::endl;
+            s.transaction_declare(*this);
+        } else {
+            std::cout << "Transaction is declared: " << s.transaction_id() << std::endl;
+            send();
+        }
     }
 
-    void on_transaction_declare_failed(proton::session s) override {
-        std::cout << "Transaction declarion failed" << std::endl;
+    void on_session_error(proton::session &s) override {
+        std::cout << "Session error: " << s.error().what() << std::endl;
         s.connection().close();
         exit(-1);
     }
 
-    void on_transaction_commit_failed(proton::session s) override {
+    void on_session_transaction_commit_failed(proton::session &s) override {
         std::cout << "Transaction commit failed!" << std::endl;
         s.connection().close();
         exit(-1);
     }
 
-    void on_transaction_declared(proton::session s) override {
-        std::cout << "Transaction is declared: " << s.transaction_id() << std::endl;
-        send();
-    }
-
     void on_sendable(proton::sender&) override {
         send();
     }
@@ -117,7 +116,7 @@ class tx_send : public proton::messaging_handler, proton::transaction_handler {
         }
     }
 
-    void on_transaction_committed(proton::session s) override {
+    void on_session_transaction_committed(proton::session &s) override {
         committed += current_batch;
         current_batch = 0;
         std::cout << "Transaction commited" << std::endl;
@@ -131,7 +130,7 @@ class tx_send : public proton::messaging_handler, proton::transaction_handler {
         }
     }
 
-    void on_transaction_aborted(proton::session s) override {
+    void on_session_transaction_aborted(proton::session &s) override {
         std::cout << "Transaction aborted!" << std::endl;
         std::cout << "Re-delaring transaction now..." << std::endl;
         current_batch = 0;
diff --git a/cpp/include/proton/fwd.hpp b/cpp/include/proton/fwd.hpp
@@ -55,7 +55,6 @@ class source_options;
 class ssl;
 class target_options;
 class tracker;
-class transaction_handler;
 class transport;
 class url;
 class void_function0;
diff --git a/cpp/include/proton/messaging_handler.hpp b/cpp/include/proton/messaging_handler.hpp
@@ -172,6 +172,15 @@ PN_CPP_CLASS_EXTERN messaging_handler {
     /// The remote peer closed the session with an error condition.
     PN_CPP_EXTERN virtual void on_session_error(session&);
 
+    /// Called when a local transaction is discharged successfully.
+    PN_CPP_EXTERN virtual void on_session_transaction_committed(session&);
+
+    /// Called when the commit of a local transaction fails.
+    PN_CPP_EXTERN virtual void on_session_transaction_commit_failed(session&);
+
+    /// Called when a local transaction is discharged unsuccessfully (aborted).
+    PN_CPP_EXTERN virtual void on_session_transaction_aborted(session&);
+
     /// The remote peer opened the link.
     PN_CPP_EXTERN virtual void on_receiver_open(receiver&);
 
diff --git a/cpp/include/proton/session.hpp b/cpp/include/proton/session.hpp
@@ -37,10 +37,6 @@ struct pn_session_t;
 
 namespace proton {
 
-/// @cond INTERNAL
-class transaction_impl;
-/// @endcond
-
 /// A container of senders and receivers.
 class
 PN_CPP_CLASS_EXTERN session : public internal::object<pn_session_t>, public endpoint {
@@ -109,7 +105,7 @@ PN_CPP_CLASS_EXTERN session : public internal::object<pn_session_t>, public endp
     /// Get user data from this session.
     PN_CPP_EXTERN void* user_data() const;
 
-    PN_CPP_EXTERN void transaction_declare(proton::transaction_handler &handler, bool settle_before_discharge = false);
+    PN_CPP_EXTERN void transaction_declare(proton::messaging_handler &handler, bool settle_before_discharge = false);
     PN_CPP_EXTERN bool transaction_is_declared();
     PN_CPP_EXTERN proton::binary transaction_id() const;
     PN_CPP_EXTERN void transaction_commit();
@@ -121,7 +117,6 @@ PN_CPP_CLASS_EXTERN session : public internal::object<pn_session_t>, public endp
     friend class internal::factory<session>;
     friend class sender;
     friend class session_iterator;
-    friend class transaction_impl;
     /// @endcond
 };
 
diff --git a/cpp/include/proton/transaction_handler.hpp b/cpp/include/proton/transaction_handler.hpp
diff --git a/cpp/src/contexts.hpp b/cpp/src/contexts.hpp
@@ -42,6 +42,7 @@ namespace proton {
 
 class proton_handler;
 class connector;
+class transaction_impl;
 
 namespace io {class link_namer;}
 
diff --git a/cpp/src/delivery.cpp b/cpp/src/delivery.cpp
@@ -36,12 +36,12 @@ namespace {
 void settle_delivery(pn_delivery_t* o, uint64_t state) {
     proton::session session = proton::make_wrapper(o).session();
     if(session.transaction_is_declared()) {
-       // Transactional disposition
+        // Transactional disposition
         auto disp = pn_transactional_disposition(pn_delivery_local(o));
         pn_transactional_disposition_set_id(disp, pn_bytes(session.transaction_id()));
         pn_transactional_disposition_set_outcome_type(disp, state);
     } else {
-    pn_delivery_update(o, state);
+        pn_delivery_update(o, state);
     }
     pn_delivery_settle(o);
 }
diff --git a/cpp/src/handler.cpp b/cpp/src/handler.cpp
@@ -65,6 +65,10 @@ void messaging_handler::on_session_open(session &s) {
         pn_session_open(unwrap(s));
     }
 }
+void messaging_handler::on_session_transaction_committed(session &) {}
+void messaging_handler::on_session_transaction_commit_failed(session &) {}
+void messaging_handler::on_session_transaction_aborted(session &) {}
+
 void messaging_handler::on_receiver_close(receiver &) {}
 void messaging_handler::on_receiver_error(receiver &l) { on_error(l.error()); }
 void messaging_handler::on_receiver_open(receiver &l) {
diff --git a/cpp/src/messaging_adapter.cpp b/cpp/src/messaging_adapter.cpp
@@ -119,29 +119,7 @@ void on_delivery(messaging_handler& handler, pn_event_t* event) {
     link_context& lctx = link_context::get(lnk);
     Tracing& ot = Tracing::getTracing();
 
-    if (pn_terminus_get_type(pn_link_remote_target(lnk)) == PN_COORDINATOR) {
-        if (pn_delivery_updated(dlv)) {
-            tracker t(make_wrapper<tracker>(dlv));
-            ot.on_settled_span(t);
-            switch (pn_delivery_remote_state(dlv)) {
-            case PN_ACCEPTED:
-                handler.on_tracker_accept(t);
-                break;
-            case PN_REJECTED:
-                handler.on_tracker_reject(t);
-                break;
-            case PN_RELEASED:
-            case PN_MODIFIED:
-                handler.on_tracker_release(t);
-                break;
-            }
-            if (t.settled()) {
-                handler.on_tracker_settle(t);
-                if (lctx.auto_settle)
-                    t.settle();
-            }
-        }
-    } else if (pn_link_is_receiver(lnk)) {
+    if (pn_link_is_receiver(lnk)) {
         delivery d(make_wrapper<delivery>(dlv));
         if (pn_delivery_aborted(dlv)) {
             pn_delivery_settle(dlv);
@@ -299,15 +277,6 @@ 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);
-    if (pn_terminus_get_type(pn_link_remote_target(lnk)) == PN_COORDINATOR) {
-      auto cond = pn_link_condition(lnk);
-      if (pn_condition_is_set(cond)) {
-            pn_condition_set_name(cond, "amqp:on_link_remote_open:FAILED");
-            pn_link_close(lnk);
-            return;
-        }
-      return;
-    }
     if (pn_link_state(lnk) & PN_LOCAL_UNINIT) { // Incoming link
         // Copy source and target from remote end.
         pn_terminus_copy(pn_link_source(lnk), pn_link_remote_source(lnk));
diff --git a/cpp/src/session.cpp b/cpp/src/session.cpp
diff --git a/cpp/src/transaction_handler.cpp b/cpp/src/transaction_handler.cpp
diff --git a/cpp/src/transaction_test.cpp b/cpp/src/transaction_test.cpp
diff --git a/cpp/tests.cmake b/cpp/tests.cmake

Original file line number	Diff line number	Diff line change
`@@ -65,6 +65,10 @@ void messaging_handler::on_session_open(session &s) {`
`65`	`65`	`pn_session_open(unwrap(s));`
`66`	`66`	`}`
`67`	`67`	`}`
	`68`	`+void messaging_handler::on_session_transaction_committed(session &) {}`
	`69`	`+void messaging_handler::on_session_transaction_commit_failed(session &) {}`
	`70`	`+void messaging_handler::on_session_transaction_aborted(session &) {}`
	`71`	`+`
`68`	`72`	`void messaging_handler::on_receiver_close(receiver &) {}`
`69`	`73`	`void messaging_handler::on_receiver_error(receiver &l) { on_error(l.error()); }`
`70`	`74`	`void messaging_handler::on_receiver_open(receiver &l) {`