Merge branch 'main' into jd_2022_10_05_compile_as_java11

Author: jiridanek

cli-protonj2/pom.xml

@@ -42,13 +42,6 @@
         <jaeger-client.version>1.6.0</jaeger-client.version>
     </properties>
 
-    <repositories>
-        <repository>
-            <id>staging</id>
-            <url>https://repository.apache.org/content/repositories/orgapacheqpid-1218</url>
-        </repository>
-    </repositories>
-
     <!--    <dependencyManagement>-->
     <!--        <dependencies>-->
     <!--            <dependency>-->

cli-protonj2/src/main/java/com/redhat/mqe/CliProtonJ2Receiver.java

@@ -30,8 +30,11 @@
 import org.apache.qpid.protonj2.client.Receiver;
 import org.apache.qpid.protonj2.client.ReceiverOptions;
 import org.apache.qpid.protonj2.client.Sender;
+import org.apache.qpid.protonj2.client.Session;
+import org.apache.qpid.protonj2.client.exceptions.ClientException;
 import picocli.CommandLine;
 
+import java.io.IOException;
 import java.net.URI;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
@@ -109,14 +112,26 @@ public class CliProtonJ2Receiver extends CliProtonJ2SenderReceiver implements Ca
     private Integer duration;
 
     @CommandLine.Option(names = {"--duration-mode"}) // todo
-    private DurationMode durationMode;
+    private DurationModeReceiver durationMode;
 
     @CommandLine.Option(names = {"--ssn-ack-mode"})
     private SsnAckMode ssnAckMode;
 
+    @CommandLine.Option(names = {"--tx-size"})
+    private Integer txSize;
+
+    @CommandLine.Option(names = {"--tx-endloop-action"})
+    private TxAction txEndloopAction;
+
+    @CommandLine.Option(names = {"--tx-action"})
+    private TxAction txAction;
+
     @CommandLine.Option(names = {"--msg-content-to-file"})
     private String msgContentToFile;
 
+    @CommandLine.Option(names = {"--conn-reconnect"})
+    private String reconnectString = "false";
+
     public CliProtonJ2Receiver() {
         this.messageFormatter = new ProtonJ2MessageFormatter();
     }
@@ -125,8 +140,11 @@ public CliProtonJ2Receiver(ProtonJ2MessageFormatter messageFormatter) {
         this.messageFormatter = messageFormatter;
     }
 
+    /**
+     * This is the main function of the client, as called by the cli options handling library.
+     */
     @Override
-    public Integer call() throws Exception { // your business logic goes here...
+    public Integer call() throws Exception {
         String prefix = "";
         if (!broker.startsWith("amqp://") && !broker.startsWith("amqps://")) {
             prefix = "amqp://";
@@ -163,6 +181,9 @@ public Integer call() throws Exception { // your business logic goes here...
         }
 
         final ConnectionOptions options = new ConnectionOptions();
+        if (stringToBool(reconnectString)) {
+            options.reconnectEnabled(true);
+        }
         options.user(connUsername);
         options.password(connPassword);
         for (AuthMechanism mech : connAuthMechanisms) {
@@ -189,26 +210,33 @@ public Integer call() throws Exception { // your business logic goes here...
             }
         }
 
-        try (Connection connection = client.connect(serverHost, serverPort, options)) {
+        boolean transacted = txSize != null || txAction != null || txEndloopAction != null;
+
+        try (Connection connection = client.connect(serverHost, serverPort, options);
+             Session session = connection.openSession()) {
             Receiver receiver;
             if (stringToBool(durableSubscriberString)) {
-                receiver = connection.openDurableReceiver(address, durableSubscriberName, receiverOptions);
+                receiver = session.openDurableReceiver(address, durableSubscriberName, receiverOptions);
             } else {
-                receiver = connection.openReceiver(address, receiverOptions);
+                receiver = session.openReceiver(address, receiverOptions);
             }
 
             if (stringToBool(subscriberUnsubscribeString)) {
                 receiver.close();
                 return 0;
             }
 
+            if (transacted) {
+                session.beginTransaction();
+            }
+
+            int i = 0;
             double initialTimestamp = Utils.getTime();
-            for (int i = 0; i < count; i++) {
+            while (true) {
 
-//                if (durationMode == DurationMode.sleepBeforeReceive) {
-//                    LOG.trace("Sleeping before receive");
-//                    Utils.sleepUntilNextIteration(initialTimestamp, msgCount, duration, i + 1);
-//                }
+                if (durationMode == DurationModeReceiver.beforeReceive) {
+                    Utils.sleepUntilNextIteration(initialTimestamp, count, duration, i + 1);
+                }
 
                 final Delivery delivery;
                 if (timeout == 0) {
@@ -221,9 +249,8 @@ public Integer call() throws Exception { // your business logic goes here...
                     break;
                 }
 
-                if (durationMode == DurationMode.afterReceive) {
-//                    LOG.trace("Sleeping after receive");
-                    Utils.sleepUntilNextIteration(initialTimestamp, count, duration, i + 1); // todo possibly it is i, different loop here
+                if (durationMode == DurationModeReceiver.afterReceive) {
+                    Utils.sleepUntilNextIteration(initialTimestamp, count, duration, i + 1);
                 }
 
                 if (processReplyTo && delivery.message().replyTo() != null) {
@@ -235,42 +262,51 @@ public Integer call() throws Exception { // your business logic goes here...
                     }
                 }
 
-                int messageFormat = delivery.messageFormat();
-                Message<Object> message = delivery.message();
-
                 // todo, is this what we mean?
                 if (ssnAckMode != null && ssnAckMode == SsnAckMode.client) {
                     delivery.accept();
                 }
 
-                Map<String, Object> messageDict = messageFormatter.formatMessage(address, message, stringToBool(msgContentHashedString));
-                if (msgContentToFile != null) {
-                    // todo?
-                    Path file = Paths.get(msgContentToFile + "_" + i);
-                    Files.write(file, message.body().toString().getBytes(StandardCharsets.UTF_8));
-                }
-                switch(out) {
-                    case python:
-                        switch (logMsgs) {
-                            case dict:
-                                messageFormatter.printMessageAsPython(messageDict);
-                                break;
-                            case interop:
-                                messageFormatter.printMessageAsPython(messageDict);
-                                break;
+                outputReceivedMessage(i, delivery);
+                i++;
+
+                if (txSize != null && txSize != 0) {
+                    if (i % txSize == 0) {
+                        if (txAction != null) {
+                            switch (txAction) {
+                                case commit:
+                                    session.commitTransaction();
+                                    break;
+                                case rollback:
+                                    session.rollbackTransaction();
+                                    break;
+                            }
+
+                            session.beginTransaction();
+
+                            if (durationMode == DurationModeReceiver.afterReceiveTxAction) {
+                                Utils.sleepUntilNextIteration(initialTimestamp, i, duration, i + 1);
+                            }
                         }
+                    }
+                }
+
+                if (i == count) { // not i > count; --count=0 needs to disable the break
+                    break;
+                }
+            }
+
+            if (txEndloopAction != null) {
+                switch (txEndloopAction) {
+                    case commit:
+                        session.commitTransaction();
                         break;
-                    case json:
-                        switch (logMsgs) {
-                            case dict:
-                                messageFormatter.printMessageAsJson(messageDict);
-                                break;
-                            case interop:
-                                messageFormatter.printMessageAsJson(messageDict);
-                                break;
-                        }
+                    case rollback:
+                        session.rollbackTransaction();
                         break;
                 }
+            } else if (transacted) {
+                session.rollbackTransaction();
             }
 
             if (stringToBool(durableSubscriberString)) {
@@ -283,4 +319,36 @@ public Integer call() throws Exception { // your business logic goes here...
         return 0;
     }
 
+    private void outputReceivedMessage(int i, Delivery delivery) throws ClientException, IOException {
+        Message<Object> message = delivery.message();
+        int messageFormat = delivery.messageFormat();
+        Map<String, Object> messageDict = messageFormatter.formatMessage(address, message, stringToBool(msgContentHashedString));
+        if (msgContentToFile != null) {
+            // todo?
+            Path file = Paths.get(msgContentToFile + "_" + i);
+            Files.write(file, message.body().toString().getBytes(StandardCharsets.UTF_8));
+        }
+        switch (out) {
+            case python:
+                switch (logMsgs) {
+                    case dict:
+                        messageFormatter.printMessageAsPython(messageDict);
+                        break;
+                    case interop:
+                        messageFormatter.printMessageAsPython(messageDict);
+                        break;
+                }
+                break;
+            case json:
+                switch (logMsgs) {
+                    case dict:
+                        messageFormatter.printMessageAsJson(messageDict);
+                        break;
+                    case interop:
+                        messageFormatter.printMessageAsJson(messageDict);
+                        break;
+                }
+                break;
+        }
+    }
 }