refactoring

Author: Petr Matousek

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

@@ -157,8 +157,6 @@ void TxReceiverHandler::on_transaction_declared(session s) {
     }
     logger(trace) << "[on_transaction_declared] txn called " << (&s);
     logger(debug) << "[on_transaction_declared] txn is_empty " << (s.txn_is_empty());
-    // TODO
-    // tx = t;
 }
 
 void TxReceiverHandler::on_transaction_aborted(session s) {
@@ -168,15 +166,15 @@ void TxReceiverHandler::on_transaction_aborted(session s) {
     if (count == 0 || processed < count) {
         s.declare_transaction(*this);
     } else {
-        logger(info) << "[on_transaction_committed] All messages processed";
+        logger(info) << "[on_transaction_aborted] All messages processed";
         s.connection().close();
     }
 }
 
 void TxReceiverHandler::on_transaction_committed(session s) {
     processed += current_batch;
     current_batch = 0;
-    logger(debug) << "[on_transaction_aborted] messages committed, processed: " << processed;
+    logger(debug) << "[on_transaction_committed] messages committed, processed: " << processed;
     if (count == 0 || processed < count) {
         s.declare_transaction(*this);
     } else {
@@ -370,7 +368,9 @@ void TxReceiverHandler::on_message(delivery &d, message &m)
 {
     logger(debug) << "[on_message] Processing received message";
 
+    // TODO legit?
     session s = d.session();
+
     s.txn_accept(d);
     current_batch += 1;
 

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

@@ -110,18 +110,19 @@ int TxSenderHandler::getBatchSize() const
     return batch_size;
 }
 
-// void TxSenderHandler::checkIfCanSend() {
-//     if (processed < count) {
-//         work_q->schedule(interval, make_work(&TxSenderHandler::checkIfCanSend, this));
-// 
-//         if (sndr.credit() > 0) {
-//             logger(debug) << "[checkIfCanSend] Preparing to send message";
-//             send();
-//         } else {
-//             ready = true;
-//         }
-//     }
-// }
+void TxSenderHandler::checkIfCanSend() {
+    if (processed < count) {
+        work_q->schedule(interval, make_work(&TxSenderHandler::checkIfCanSend, this));
+
+        if (sndr.credit() > 0) {
+            logger(debug) << "[checkIfCanSend] Preparing to send message";
+            // TODO test w/ session defined
+            send();
+        } else {
+            ready = true;
+        }
+    }
+}
 
 void TxSenderHandler::send(session s)
 {
@@ -156,9 +157,7 @@ void TxSenderHandler::send(session s)
     logger(trace) << "[send] Current batch: " << current_batch;
     while (s.txn_is_declared() && sndr.credit() && (processed + current_batch) < count)
     {
-    logger(trace) << "[send] Sending messages through the link NAZDAR";
         s.txn_send(sndr, message_to_send);
-    logger(trace) << "[send] Sending messages through the link BAZAR";
         current_batch += 1;
 
         if (log_msgs == "dict") {
@@ -223,6 +222,7 @@ void TxSenderHandler::send(session s)
 
 void TxSenderHandler::on_sendable(sender &s)
 {
+    // TODO
     logger(trace) <<  "[on_sendable] IS THIS METHOD EVER CALLED IN TX MODE???";
     logger(trace) <<  "[on_sendable] transaction: " << &s;
     if (ready) {
@@ -245,7 +245,6 @@ void TxSenderHandler::on_transaction_declared(session s) {
     logger(trace) << "[on_transaction_declared] txn called " << (&s);
     logger(trace) << "[on_transaction_declared] txn is_empty " << (s.txn_is_empty())
                   << "\t" << s.txn_is_empty();
-    // tx = t;
     send(s);
 }
 
@@ -368,10 +367,11 @@ void TxSenderHandler::on_container_start(container &c)
 
         logger(trace) << "[on_container_start] Interval for duration: " << interval.milliseconds() << " ms";
     }
+
 // TODO
 // #if defined(__REACTOR_HAS_TIMER)
 //     work_q->schedule(duration::IMMEDIATE, make_work(&TxSenderHandler::timerEvent, this));
-// 
+//
 //     if (duration_time > 0 && duration_mode == "after-send") {
 //         work_q->schedule(duration::IMMEDIATE, make_work(&TxSenderHandler::checkIfCanSend, this));
 //     } else if (duration_time > 0 && duration_mode == "before-send") {
@@ -380,8 +380,6 @@ void TxSenderHandler::on_container_start(container &c)
 //         work_q->schedule(duration::IMMEDIATE, make_work(&TxSenderHandler::checkIfCanSend, this));
 //     }
 // #endif
-
-//    tx = transaction();
 }
 
 void TxSenderHandler::on_transaction_declare_failed(session) {}

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

@@ -115,7 +115,7 @@ class TxSenderHandler : public SenderHandler, transaction_handler {
 
     // overrides
     void checkIfCanSend();
-    void send(session sess);
+    void send(session s);
 
     // reactor methods
     void on_sender_close(sender &s);