Removed coordinator class and improved STATES handling.

Author: DreamPearl

cpp/examples/CMakeLists.txt

@@ -61,8 +61,7 @@ foreach(example
     service_bus
     multithreaded_client
     multithreaded_client_flow_control
-    tx_send
-    tx_recv)
+    tx_send)
   add_executable(${example} ${example}.cpp)
   target_link_libraries(${example} Proton::cpp Threads::Threads)
 endforeach()

cpp/examples/simple_recv.cpp

@@ -43,10 +43,11 @@ class simple_recv : public proton::messaging_handler {
     proton::receiver receiver;
     int expected;
     int received;
+    bool verbose;
 
   public:
-    simple_recv(const std::string &s, const std::string &u, const std::string &p, int c) :
-        url(s), user(u), password(p), expected(c), received(0) {}
+    simple_recv(const std::string &s, const std::string &u, const std::string &p, int c, bool verbose) :
+        url(s), user(u), password(p), expected(c), received(0), verbose(verbose) {}
 
     void on_container_start(proton::container &c) override {
         proton::connection_options co;
@@ -61,6 +62,9 @@ class simple_recv : public proton::messaging_handler {
         }
 
         if (expected == 0 || received < expected) {
+            if (verbose) {
+                std::cout << msg << ": ";
+            }
             std::cout << msg.body() << std::endl;
             received++;
 
@@ -77,18 +81,20 @@ int main(int argc, char **argv) {
     std::string user;
     std::string password;
     int message_count = 100;
+    bool verbose;
     example::options opts(argc, argv);
 
     opts.add_value(address, 'a', "address", "connect to and receive from URL", "URL");
     opts.add_value(message_count, 'm', "messages", "receive COUNT messages", "COUNT");
     opts.add_value(user, 'u', "user", "authenticate as USER", "USER");
     opts.add_value(password, 'p', "password", "authenticate with PASSWORD", "PASSWORD");
+    opts.add_flag(verbose, 'v', "verbose", "show whole message contents");
 
 
     try {
         opts.parse();
 
-        simple_recv recv(address, user, password, message_count);
+        simple_recv recv(address, user, password, message_count, verbose);
         proton::container(recv).run();
 
         return 0;

cpp/examples/tx_recv.cpp

@@ -38,17 +38,14 @@
 
 class tx_send : public proton::messaging_handler, proton::transaction_handler {
   private:
-    proton::sender sender; 
+    proton::sender sender;
     std::string url;
     int total;
     int batch_size;
     int sent;
     int batch_index = 0;
     int current_batch = 0;
     int committed = 0;
-    int confirmed = 0;
-
-    proton::session session;
 
   public:
     tx_send(const std::string &s, int c, int b):
@@ -59,35 +56,34 @@ class tx_send : public proton::messaging_handler, proton::transaction_handler {
     }
 
     void on_session_open(proton::session &s) override {
-        session = s;
-        std::cout << "    [on_session_open] declare_txn started..." << std::endl;
+        std::cout << "New session is open, declaring transaction now..." << std::endl;
         s.declare_transaction(*this);
-        std::cout << "    [on_session_open] declare_txn ended..." << std::endl;
     }
 
-    void on_transaction_declare_failed(proton::session) {}
+    void on_transaction_declare_failed(proton::session s) {
+        std::cout << "Transaction declarion failed" << std::endl;
+        s.connection().close();
+        exit(-1);
+    }
+
     void on_transaction_commit_failed(proton::session s) {
-        std::cout << "Transaction Commit Failed" << std::endl;
+        std::cout << "Transaction commit failed!" << std::endl;
         s.connection().close();
         exit(-1);
     }
 
     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 " << (s.txn_is_empty())
-                  << "\t" << std::endl;
-        send(sender);
+        std::cout << "Transaction is declared" << std::endl;
+        send();
     }
 
-    void on_sendable(proton::sender &s) override {
-        std::cout << "    [OnSendable] session: " << &session
-                  << std::endl;
-        send(s);
+    void on_sendable(proton::sender&) override {
+        send();
     }
 
-    void send(proton::sender &s) {
+    void send() {
         static int unique_id = 10000;
+        proton::session session = sender.session();
         while (session.txn_is_declared() && sender.credit() &&
                (committed + current_batch) < total) {
             proton::message msg;
@@ -96,47 +92,42 @@ class tx_send : public proton::messaging_handler, proton::transaction_handler {
 
             msg.id(unique_id++);
             msg.body(m);
-            std::cout << "##### [example] transaction send msg: "  << msg
-                      << std::endl;
+            std::cout << "Sending: " << msg << std::endl;
             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) {
+                    std::cout << "Commiting transaction..." << std::endl;
                     session.txn_commit();
                 } else {
+                    std::cout << "Aborting transaction..." << std::endl;
                     session.txn_abort();
-                }       
+                }
                 batch_index++;
             }
         }
     }
 
-    void on_tracker_accept(proton::tracker &t) override {
-        confirmed += 1;
-        std::cout << "    [example] on_tracker_accept:" << confirmed
-                  << std::endl;
-    }
-
     void on_transaction_committed(proton::session s) override {
         committed += current_batch;
         current_batch = 0;
-        std::cout<<"    [OnTxnCommitted] Committed:"<< committed<< std::endl;
+        std::cout << "Transaction commited" << std::endl;
         if(committed == total) {
-            std::cout << "All messages committed" << std::endl;
+            std::cout << "All messages committed, closing connection." << std::endl;
             s.connection().close();
         }
         else {
-            std::cout << "redlcaring txn " << std::endl;
-            session.declare_transaction(*this);
+            std::cout << "Re-declaring transaction now..." << std::endl;
+            s.declare_transaction(*this);
         }
     }
 
     void on_transaction_aborted(proton::session s) override {
-        std::cout << "Meesages Aborted ....." << std::endl;
+        std::cout << "Transaction aborted!" << std::endl;
+        std::cout << "Re-delaring transaction now..." << std::endl;
         current_batch = 0;
-        session.declare_transaction(*this);
+        s.declare_transaction(*this);
     }
 
     void on_sender_close(proton::sender &s) override {

cpp/examples/tx_send.cpp

@@ -131,7 +131,7 @@ PN_CPP_CLASS_EXTERN session : public internal::object<pn_session_t>, public endp
 };
 
 /// @cond INTERNAL
-    
+
 /// An iterator of sessions.
 class session_iterator : public internal::iter_base<session, session_iterator> {
  public:
@@ -145,7 +145,7 @@ class session_iterator : public internal::iter_base<session, session_iterator> {
 typedef internal::iter_range<session_iterator> session_range;
 
 /// @endcond
-    
+
 } // proton
 
 #endif // PROTON_SESSION_HPP

cpp/include/proton/session.hpp

@@ -65,26 +65,6 @@ class target : public terminus {
     /// @endcond
 };
 
-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