Has errors

Author: DreamPearl

cpp/examples/tx_send.cpp

@@ -49,7 +49,7 @@ class tx_send : public proton::messaging_handler, proton::transaction_handler {
     int confirmed = 0;
 
     proton::session session;
-    proton::transaction transaction;
+    // proton::transaction transaction;
   public:
     tx_send(const std::string &s, int c, int b):
         url(s), total(c), batch_size(b), sent(0) {}
@@ -65,32 +65,31 @@ class tx_send : 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] Session: " << (&s)
                   << std::endl;
-        std::cout << "[on_transaction_declared] txn is_empty " << (t.is_empty())
-                  << "\t" << transaction.is_empty() << std::endl;
-        transaction = t;
+        std::cout << "[on_transaction_declared] txn is_empty " << (s.txn_is_empty())
+                  << "\t" << std::endl;
 
         send(sender);
     }
 
     void on_sendable(proton::sender &s) override {
-        std::cout << "    [OnSendable] transaction: " << &transaction
+        std::cout << "    [OnSendable] session: " << &session
                   << std::endl;
         send(s);
     }
 
     void send(proton::sender &s) {
         static int unique_id = 10000;
-        while (!transaction.is_empty() && sender.credit() &&
+        while (!session.txn_is_empty() && sender.credit() &&
                (committed + current_batch) < total) {
             proton::message msg;
             std::map<std::string, int> m;
@@ -100,18 +99,17 @@ class tx_send : public proton::messaging_handler, proton::transaction_handler {
             msg.body(m);
             std::cout << "##### [example] transaction send msg: " << msg
                       << std::endl;
-            transaction.send(sender, msg);
+            session.txn_send(sender, msg);
             current_batch += 1;
             if(current_batch == batch_size)
             {
                 std::cout << " >> Txn attempt commit" << std::endl;
                 if (batch_index % 2 == 0) {
-                    transaction.commit();
+                    session.txn_commit();
                 } else {
-                    transaction.abort();
+                    session.txn_abort();
                 }
-
-                transaction = proton::transaction();
+    // TODO: Only one transaction is permitted per session.
                 batch_index++;
             }
         }
@@ -123,20 +121,20 @@ class tx_send : public proton::messaging_handler, proton::transaction_handler {
                   << std::endl;
     }
 
-    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 == total) {
             std::cout << "All messages committed" << std::endl;
-            t.connection().close();
+            s.connection().close();
         }
         else {
             session.declare_transaction(*this);
         }
     }
 
-    void on_transaction_aborted(proton::transaction t) override {
+    void on_transaction_aborted(proton::session s) override {
         std::cout << "Meesages Aborted ....." << std::endl;
         current_batch = 0;
         session.declare_transaction(*this);
@@ -150,7 +148,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 = 6;
+    int message_count = 3;
     int batch_size = 3;
     example::options opts(argc, argv);
 

cpp/include/proton/session.hpp

@@ -29,13 +29,15 @@
 #include "./sender.hpp"
 
 #include <string>
+#include <iostream>
 
 /// @file
 /// @copybrief proton::session
 
 struct pn_session_t;
 
 namespace proton {
+  class transaction_impl;
 
 /// A container of senders and receivers.
 class
@@ -105,11 +107,32 @@ 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 transaction declare_transaction(proton::transaction_handler &handler, bool settle_before_discharge = false);
+    PN_CPP_EXTERN void declare_transaction(proton::transaction_handler &handler, bool settle_before_discharge = false);
+
+
+    // static transaction mk_transaction_impl(sender &s, transaction_handler &h,
+                                          //  bool f);
+    // PN_CPP_EXTERN transaction(transaction_impl *impl);
+    
+    // PN_CPP_EXTERN transaction();
+    // PN_CPP_EXTERN ~transaction();
+    PN_CPP_EXTERN bool txn_is_empty();
+    PN_CPP_EXTERN void txn_commit();
+    PN_CPP_EXTERN void txn_abort();
+    PN_CPP_EXTERN void txn_declare();
+    PN_CPP_EXTERN void txn_handle_outcome(proton::tracker);
+    PN_CPP_EXTERN proton::tracker txn_send(proton::sender s, proton::message msg);
+    PN_CPP_EXTERN void txn_accept(delivery &t);
+    PN_CPP_EXTERN proton::connection txn_connection() const;
+
+    // PN_CPP_EXTERN session_context& get_session_context();
+
+    // transaction _txn;
 
     /// @cond INTERNAL
   friend class internal::factory<session>;
   friend class session_iterator;
+  friend class transaction_impl;
     /// @endcond
 };
 
@@ -124,6 +147,39 @@ class session_iterator : public internal::iter_base<session, session_iterator> {
     PN_CPP_EXTERN session_iterator operator++();
 };
 
+
+class transaction_impl {
+  public:
+    proton::sender txn_ctrl;
+    proton::transaction_handler *handler = nullptr;
+    proton::binary id;
+    proton::tracker _declare;
+    proton::tracker _discharge;
+    bool failed = false;
+    std::vector<proton::tracker> pending;
+
+    void commit();
+    void abort();
+    void declare();
+    proton::tracker send(proton::sender s, proton::message msg);
+
+    void discharge(bool failed);
+    void release_pending();
+    void accept(delivery &d);
+    void update(tracker &d, uint64_t state);
+    void set_id(binary _id);
+
+    proton::tracker send_ctrl(proton::symbol descriptor, proton::value _value);
+    void handle_outcome(proton::tracker t);
+    transaction_impl(proton::sender &_txn_ctrl,
+                     proton::transaction_handler &_handler,
+                     bool _settle_before_discharge);
+
+    // delete copy and assignment operator to ensure no copy of this object is
+    // every made transaction_impl(const transaction_impl&) = delete;
+    // transaction_impl&  operator=(const transaction_impl&) = delete;
+};
+
 /// A range of sessions.
 typedef internal::iter_range<session_iterator> session_range;
 

cpp/include/proton/transaction.hpp

@@ -71,54 +71,54 @@ class transaction_impl;
 //     // transaction_impl& operator=(const transaction_impl&) = delete;
 // };
 
-class
-PN_CPP_CLASS_EXTERN transaction {
-  private:
-    //   PN_CPP_EXTERN transaction(proton::sender& _txn_ctrl,
-    //   proton::transaction_handler& _handler, bool _settle_before_discharge);
+// class
+// PN_CPP_CLASS_EXTERN transaction {
+//   private:
+//     //   PN_CPP_EXTERN transaction(proton::sender& _txn_ctrl,
+//     //   proton::transaction_handler& _handler, bool _settle_before_discharge);
 
-    static transaction mk_transaction_impl(sender &s, transaction_handler &h,
-                                           bool f);
-    PN_CPP_EXTERN transaction(transaction_impl *impl);
-    transaction_impl *_impl;
+//     static transaction mk_transaction_impl(sender &s, transaction_handler &h,
+//                                            bool f);
+//     PN_CPP_EXTERN transaction(transaction_impl *impl);
+//     transaction_impl *_impl;
 
-  public:
-    // TODO:
-    // PN_CPP_EXTERN transaction(transaction &o);
-    PN_CPP_EXTERN transaction();
-    PN_CPP_EXTERN ~transaction();
-    PN_CPP_EXTERN bool is_empty();
-    PN_CPP_EXTERN void commit();
-    PN_CPP_EXTERN void abort();
-    PN_CPP_EXTERN void declare();
-    PN_CPP_EXTERN void handle_outcome(proton::tracker);
-    PN_CPP_EXTERN proton::tracker send(proton::sender s, proton::message msg);
-    PN_CPP_EXTERN void accept(delivery &t);
-    PN_CPP_EXTERN proton::connection connection() const;
-
-  friend class transaction_impl;
-  friend class session;
-};
+//   public:
+//     // TODO:
+//     // PN_CPP_EXTERN transaction(transaction &o);
+//     PN_CPP_EXTERN transaction();
+//     PN_CPP_EXTERN ~transaction();
+//     PN_CPP_EXTERN bool is_empty();
+//     PN_CPP_EXTERN void commit();
+//     PN_CPP_EXTERN void abort();
+//     PN_CPP_EXTERN void declare();
+//     PN_CPP_EXTERN void handle_outcome(proton::tracker);
+//     PN_CPP_EXTERN proton::tracker send(proton::sender s, proton::message msg);
+//     PN_CPP_EXTERN void accept(delivery &t);
+//     PN_CPP_EXTERN proton::connection connection() const;
+
+//   friend class transaction_impl;
+//   friend class session;
+// };
 
 class
 PN_CPP_CLASS_EXTERN transaction_handler {
   public:
     PN_CPP_EXTERN virtual ~transaction_handler();
 
     /// Called when a local transaction is declared.
-    PN_CPP_EXTERN virtual void on_transaction_declared(transaction);
+    PN_CPP_EXTERN virtual void on_transaction_declared(session);
 
     /// Called when a local transaction is discharged successfully.
-    PN_CPP_EXTERN virtual void on_transaction_committed(transaction);
+    PN_CPP_EXTERN virtual void on_transaction_committed(session);
 
     /// Called when a local transaction is discharged unsuccessfully (aborted).
-    PN_CPP_EXTERN virtual void on_transaction_aborted(transaction);
+    PN_CPP_EXTERN virtual void on_transaction_aborted(session);
 
     /// Called when a local transaction declare fails.
-    PN_CPP_EXTERN virtual void on_transaction_declare_failed(transaction);
+    PN_CPP_EXTERN virtual void on_transaction_declare_failed(session);
 
     /// Called when the commit of a local transaction fails.
-    PN_CPP_EXTERN virtual void on_transaction_commit_failed(transaction);
+    PN_CPP_EXTERN virtual void on_transaction_commit_failed(session);
 };
 
 } // namespace proton

cpp/include/proton/transfer.hpp

@@ -95,9 +95,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;

cpp/src/contexts.hpp

@@ -25,7 +25,7 @@
 #include "reconnect_options_impl.hpp"
 
 #include "proton/work_queue.hpp"
-#include "proton/transaction.hpp"
+#include "proton/session.hpp"
 #include "proton/message.hpp"
 
 #include "proton/object.h"
@@ -42,7 +42,7 @@ namespace proton {
 
 class proton_handler;
 class connector;
-class transaction;
+// class transaction;
 
 namespace io {class link_namer;}
 
@@ -153,7 +153,7 @@ class session_context : public context {
   public:
     session_context() : handler(0), user_data_(nullptr) {}
     static session_context& get(pn_session_t* s);
-
+    std::unique_ptr<transaction_impl> _txn_impl;
     messaging_handler* handler;
     void* user_data_;
 };
@@ -163,7 +163,7 @@ class transfer_context : public context {
     transfer_context() : user_data_(nullptr) {}
     static transfer_context& get(pn_delivery_t* s);
 
-    std::unique_ptr<transaction> transaction_;
+    // std::unique_ptr<transaction> transaction_;
     void* user_data_;
 };