incorporate recent development changes (3dcb6a78)

Author: Petr Matousek

src/api/qpid-proton/reactor/handler/ConnectorHandler.cpp

@@ -205,7 +205,7 @@ void ConnectorHandler::on_transport_close(transport &t) {
 
 void ConnectorHandler::on_session_error(session &s) {
     logger(error) << "The remote peer at " << broker_url.getHost() << ":" << broker_url.getPort() <<
-            " closed the session with an error condition";
+            " closed the session with an error condition: " + s.error().what();
     closeObjects();
 } 
 

src/api/qpid-proton/reactor/handler/TxReceiverHandler.cpp

@@ -133,51 +133,47 @@ int TxReceiverHandler::getBatchSize() const
 // reactor methods
 
 void TxReceiverHandler::on_session_open(session &s) {
-    logger(trace) << "[on_session_open] declare_txn started...";
-    s.transaction_declare(*this);
-    logger(trace) << "[on_session_open] declare_txn ended...";
-    logger(debug) << "[on_session_open] transaction batch size: " << batch_size;
+    if(!s.transaction_is_declared()) {
+        logger(trace) << "[on_session_open] New session is open";
+        s.transaction_declare(*this);
+    } else {
+        logger(trace) << "[on_session_open] Transaction is declared: " << s.transaction_id();
+        recv.add_credit(batch_size);
+        logger(debug) << "[on_session_open] Receiver credit: " << recv.credit();
+        if (count != 0 && processed + batch_size > count) {
+            batch_size = count % batch_size;
+	}
+    }
 }
 
-void TxReceiverHandler::on_transaction_declare_failed(session) {}
-
-void TxReceiverHandler::on_transaction_commit_failed(session s) {
-    logger(debug) << "[on_transaction_commit_failed] Transaction Commit Failed";
+void TxReceiverHandler::on_session_transaction_commit_failed(session &s) {
+    logger(debug) << "[on_session_transaction_commit_failed] Transaction Commit Failed";
     s.connection().close();
     exit(-1);
 }
 
-void TxReceiverHandler::on_transaction_declared(session s) {
-    // TODO python some weird magic around count 0, doesn't make much sense to me yet
-    // when fixes take care about all count checks ofr zero
-    if (count != 0 && processed + batch_size > count) {
-        batch_size = count % batch_size;
-    } else if (count != 0) {
-        batch_size = count;
-    }
-    logger(trace) << "[on_transaction_declared] txn called " << s.transaction_id();
-}
-
-void TxReceiverHandler::on_transaction_aborted(session s) {
+void TxReceiverHandler::on_session_transaction_aborted(session &s) {
     processed += current_batch;
     current_batch = 0;
-    logger(debug) << "[on_transaction_aborted] messages aborted, processed: " << processed;
+    logger(debug) << "[on_session_transaction_aborted] messages aborted, processed: " << processed;
     if (count == 0 || processed < count) {
+        logger(info) << "[on_session_transaction_aborted] re-declaring transaction";
         s.transaction_declare(*this);
     } else {
-        logger(info) << "[on_transaction_aborted] All messages processed";
+        logger(info) << "[on_session_transaction_aborted] All messages processed";
         s.connection().close();
     }
 }
 
-void TxReceiverHandler::on_transaction_committed(session s) {
+void TxReceiverHandler::on_session_transaction_committed(session &s) {
     processed += current_batch;
     current_batch = 0;
-    logger(debug) << "[on_transaction_committed] messages committed, processed: " << processed;
+    logger(debug) << "[on_session_transaction_committed] messages committed, processed: " << processed;
     if (count == 0 || processed < count) {
+        logger(info) << "[on_session_transaction_committed] re-declaring transaction";
         s.transaction_declare(*this);
     } else {
-        logger(info) << "[on_transaction_committed] All messages processed";
+        logger(info) << "[on_session_transaction_committed] All messages processed";
         s.connection().close();
     }
 }
@@ -367,14 +363,10 @@ void TxReceiverHandler::on_message(delivery &d, message &m)
 {
     logger(debug) << "[on_message] Processing received message";
 
-    // TODO legit?
     session s = d.session();
-
     d.accept();
     current_batch += 1;
 
-    logger(debug) << "[on_message] current batch: " << current_batch;
-
     if (log_msgs == "dict") {
         logger(trace) << "[on_message] Decoding message";
         ReactorDecoder decoder = ReactorDecoder(m);
@@ -436,8 +428,10 @@ void TxReceiverHandler::on_message(delivery &d, message &m)
 #endif
     }
 
+    logger(debug) << "[on_message] Receiver credit: " << recv.credit();
+    logger(debug) << "[on_message] Current batch: " << current_batch;
     if(current_batch == batch_size) {
-        logger(debug) << "[send] Transaction attempt: " << tx_action;
+        logger(debug) << "[on_message] Transaction attempt: " << tx_action;
         if (tx_action == "commit") {
             s.transaction_commit();
         } else if (tx_action == "rollback") {
@@ -459,7 +453,7 @@ void TxReceiverHandler::on_message(delivery &d, message &m)
         }
 
     } else if (count != 0 && processed + current_batch == count) {
-        logger(debug) << "[send] Transaction attempt (endloop): " << tx_endloop_action;
+        logger(debug) << "[on_message] Transaction attempt (endloop): " << tx_endloop_action;
         if (tx_endloop_action == "commit") {
             s.transaction_commit();
         } else if (tx_endloop_action == "rollback") {

src/api/qpid-proton/reactor/handler/TxReceiverHandler.h

@@ -142,10 +142,9 @@ class TxReceiverHandler : public ReceiverHandler {
     // reactor method
     void on_session_open(session &s);
     void on_transaction_declare_failed(session);
-    void on_transaction_commit_failed(session s);
-    void on_transaction_declared(session s);
-    void on_transaction_committed(session s);
-    void on_transaction_aborted(session s);
+    void on_session_transaction_commit_failed(session &s);
+    void on_session_transaction_committed(session &s);
+    void on_session_transaction_aborted(session &s);
 
     // overrides
     void on_container_start(container &c);

src/api/qpid-proton/reactor/handler/TxSenderHandler.cpp

@@ -124,8 +124,9 @@ void TxSenderHandler::checkIfCanSend() {
     }
 }
 
-void TxSenderHandler::send(session s)
+void TxSenderHandler::send()
 {
+    session s = sndr.session();
     logger(debug) << "[send] Preparing to send message";
     int credit = sndr.credit();
 
@@ -223,49 +224,39 @@ void TxSenderHandler::on_sendable(sender &s)
 {
     logger(trace) <<  "[on_sendable] transaction: " << &s;
     if (ready) {
-        send(s.session());
+        send();
     }
 }
 
-void TxSenderHandler::on_tracker_accept(tracker &t)
-{
-    logger(trace) << "[on_tracker_accept] Message accepted, confirmed message delivery: " << processed;
-}
-
 void TxSenderHandler::on_connection_close(connection &c)
 {
     current_batch = 0;
     logger(debug) << "[on_connection_close] Closing connection";
 }
 
-void TxSenderHandler::on_transaction_declared(session s) {
-    logger(trace) << "[on_transaction_declared] txn called " << s.transaction_id();
-    send(s);
-}
-
-void TxSenderHandler::on_transaction_committed(session s) {
-    logger(trace) << "[on_transaction_committed] Messages committed";
+void TxSenderHandler::on_session_transaction_committed(session &s) {
+    logger(trace) << "[on_session_transaction_committed] Messages committed";
     processed += current_batch;
-    logger(debug) << "[on_transaction_committed] Messages processed" << processed;
+    logger(debug) << "[on_session_transaction_committed] Messages processed: " << processed;
     if (processed == count) {
-        logger(trace) << "[on_transaction_committed] All messages processed";
+        logger(trace) << "[on_session_transaction_committed] All messages processed";
         s.connection().close();
     } else {
-        logger(trace) << "[on_transaction_committed] Declaring new transaction";
+        logger(trace) << "[on_session_transaction_committed] Declaring new transaction";
         current_batch = 0;
         s.transaction_declare(*this);
     }
 }
 
-void TxSenderHandler::on_transaction_aborted(session s) {
-    logger(trace) << "[on_transaction_aborted] Messages aborted";
+void TxSenderHandler::on_session_transaction_aborted(session &s) {
+    logger(trace) << "[on_session_transaction_aborted] Messages aborted";
     processed += current_batch;
-    logger(debug) << "[on_transaction_aborted] Messages processed" << processed;
+    logger(debug) << "[on_session_transaction_aborted] Messages processed: " << processed;
     if (processed == count) {
-        logger(trace) << "[on_transaction_aborted] All messages processed";
+        logger(trace) << "[on_session_transaction_aborted] All messages processed";
         s.connection().close();
     } else {
-        logger(trace) << "[on_transaction_aborted] Declaring new transaction";
+        logger(trace) << "[on_session_transaction_aborted] Declaring new transaction";
         current_batch = 0;
         s.transaction_declare(*this);
     }
@@ -276,9 +267,13 @@ void TxSenderHandler::on_sender_close(sender &s) {
 }
 
 void TxSenderHandler::on_session_open(session &s) {
-     logger(trace) << "[on_session_open] declare_txn started...";
-     s.transaction_declare(*this);
-     logger(trace) << "[on_session_open] declare_txn ended...";
+    if(!s.transaction_is_declared()) {
+        logger(trace) << "[on_session_open] New session is open";
+        s.transaction_declare(*this);
+    } else {
+        logger(trace) << "[on_session_open] Transaction is declared: " << s.transaction_id();
+        send();
+    }
 }
 
 void TxSenderHandler::on_container_start(container &c)
@@ -377,10 +372,8 @@ void TxSenderHandler::on_container_start(container &c)
 // #endif
 }
 
-void TxSenderHandler::on_transaction_declare_failed(session) {}
-
-void TxSenderHandler::on_transaction_commit_failed(session s) {
-    logger(error) << "[on_transaction_commit_failed] Transaction Commit Failed";
+void TxSenderHandler::on_session_transaction_commit_failed(session &s) {
+    logger(error) << "[on_session_transaction_commit_failed] Transaction Commit Failed";
     s.connection().close();
     exit(1);
 }

src/api/qpid-proton/reactor/handler/TxSenderHandler.h

@@ -111,21 +111,18 @@ class TxSenderHandler : public SenderHandler {
 
     // overrides
     void checkIfCanSend();
-    void send(session s);
+    void send();
 
     // reactor methods
     void on_sender_close(sender &s);
-    void on_transaction_declared(session s);
-    void on_transaction_committed(session s);
-    void on_transaction_aborted(session s);
-    void on_transaction_declare_failed(session s);
-    void on_transaction_commit_failed(session s);
+    void on_session_transaction_commit_failed(session &s);
+    void on_session_transaction_committed(session &s);
+    void on_session_transaction_aborted(session &s);
 
     // overrides
     void on_container_start(container &c);
     void on_session_open(session &s);
     void on_sendable(sender &s);
-    void on_tracker_accept(tracker &t);
     void on_connection_close(connection &c);
 
   private: