passing tx-size fix, default batch size set to 10

Author: Petr Matousek

src/api/qpid-proton/reactor/TxReceivingClient.cpp

@@ -315,7 +315,7 @@ int TxReceivingClient::run(int argc, char **argv) const
         std::transform(recv_listen.begin(), recv_listen.end(), recv_listen.begin(), ::tolower);
     }
 
-    int recv_credit_window = -1;
+    int recv_credit_window = 0;
     if(options.is_set("recv-credit-window")) {
         recv_credit_window = atoi(options["recv-credit-window"].c_str());
     }
@@ -386,16 +386,16 @@ int TxReceivingClient::run(int argc, char **argv) const
         recv_listen,
         recv_listen_port,
         recv_credit_window,
-        recv_drain_after_credit_window
+        recv_drain_after_credit_window,
+        tx_action,
+        tx_endloop_action
     );
 
     if (selector != "") {
         handler.setSelector(selector);
     }
 
-    // TODO python defaults to 10
-    // int tx_size = 10;
-    int tx_size = 0;
+    int tx_size = 10;
     if (options.is_set("tx-size")) {
         tx_size = static_cast<int> (options.get("tx-size"));
     }

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

@@ -438,6 +438,9 @@ void TxReceiverHandler::on_message(delivery &d, message &m)
             s.transaction_commit();
         } else if (tx_action == "rollback") {
             s.transaction_abort();
+            if (count == 0) {
+                recv.connection().close();
+            }
         }
 
         if (tx_action == "none") {
@@ -453,17 +456,17 @@ void TxReceiverHandler::on_message(delivery &d, message &m)
         if (duration_time > 0 && duration_mode == "after-receive-action-tx-action") {
             // TODO: not implemented yet
         }
-
     } else if (count != 0 && processed + current_batch == count) {
         logger(debug) << "[on_message] Transaction attempt (endloop): " << tx_endloop_action;
         if (tx_endloop_action == "commit") {
             s.transaction_commit();
         } else if (tx_endloop_action == "rollback") {
             s.transaction_abort();
         } else {
-          recv.connection().close();
+            recv.connection().close();
         }
     }
+
 }
 
 void TxReceiverHandler::on_transport_close(transport &t) {

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

@@ -154,7 +154,7 @@ class TxReceiverHandler : public ReceiverHandler {
   private:
     typedef ReceiverHandler super;
 
-    int batch_size = 0;
+    int batch_size = 10;
     int current_batch = 0;
     int processed = 0;
     string tx_action = "commit";

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

@@ -128,7 +128,7 @@ class TxSenderHandler : public SenderHandler {
   private:
     typedef SenderHandler super;
 
-    int batch_size = 0;
+    int batch_size = 10;
     int current_batch = 0;
     int processed = 0;
     string tx_action = "commit";