[TH2-5050] Decrease timeout for sending on each attempt

src/main/java/com/exactpro/th2/FixHandler.java

@@ -55,8 +55,13 @@
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.ReadWriteLock;
 import java.util.concurrent.locks.ReentrantLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.function.Consumer;
+
 import kotlin.jvm.functions.Function1;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.lang3.exception.ExceptionUtils;
@@ -65,6 +70,9 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.annotation.concurrent.GuardedBy;
+import javax.annotation.concurrent.ThreadSafe;
+
 import static com.exactpro.th2.conn.dirty.fix.FixByteBufUtilKt.findField;
 import static com.exactpro.th2.conn.dirty.fix.FixByteBufUtilKt.findLastField;
 import static com.exactpro.th2.conn.dirty.fix.FixByteBufUtilKt.firstField;
@@ -165,6 +173,8 @@ public class FixHandler implements AutoCloseable, IHandler {
 
     private final AtomicReference<Future<?>> heartbeatTimer = new AtomicReference<>(CompletableFuture.completedFuture(null));
     private final AtomicReference<Future<?>> testRequestTimer = new AtomicReference<>(CompletableFuture.completedFuture(null));
+
+    private final TimeoutSendHandler timeoutSendHandler;
     private Future<?> reconnectRequestTimer = CompletableFuture.completedFuture(null);
     private volatile IChannel channel;
     protected FixHandlerSettings settings;
@@ -235,9 +245,11 @@ public FixHandler(IHandlerContext context) {
         if (settings.getHeartBtInt() <= 0) throw new IllegalArgumentException("HeartBtInt cannot be negative or zero");
         if (settings.getTestRequestDelay() <= 0) throw new IllegalArgumentException("TestRequestDelay cannot be negative or zero");
         if (settings.getDisconnectRequestDelay() <= 0) throw new IllegalArgumentException("DisconnectRequestDelay cannot be negative or zero");
-        if (settings.getConnectionTimeoutOnSend() <= 0) {
-            throw new IllegalArgumentException("connectionTimeoutOnSend must be greater than zero");
-        }
+        this.timeoutSendHandler = new TimeoutSendHandler(
+                settings.getConnectionTimeoutOnSend(),
+                settings.getMinConnectionTimeoutOnSend(),
+                context::send
+        );
     }
 
     @Override
@@ -276,7 +288,7 @@ private CompletableFuture<MessageID> send(@NotNull ByteBuf body, @NotNull Map<St
             try {
                 disconnect(!isUngracefulDisconnect);
                 enabled.set(false);
-                channel.open().get(settings.getConnectionTimeoutOnSend(), TimeUnit.MILLISECONDS);
+                timeoutSendHandler.getWithTimeout(channel.open());
             } catch (Exception e) {
                 context.send(CommonUtil.toErrorEvent(String.format("Error while ending session %s by user logout. Is graceful disconnect: %b", channel.getSessionAlias(), !isUngracefulDisconnect), e));
             }
@@ -286,15 +298,16 @@ private CompletableFuture<MessageID> send(@NotNull ByteBuf body, @NotNull Map<St
         // TODO: probably, this should be moved to the core part
         // But those changes will break API
         // So, let's keep it here for now
-        long deadline = System.currentTimeMillis() + settings.getConnectionTimeoutOnSend();
+        long deadline = timeoutSendHandler.getDeadline();
+        long currentTimeout = timeoutSendHandler.getCurrentTimeout();
 
         if (!channel.isOpen()) {
             try {
-                channel.open().get(settings.getConnectionTimeoutOnSend(), TimeUnit.MILLISECONDS);
+                timeoutSendHandler.getWithTimeout(channel.open());
             } catch (TimeoutException e) {
                 ExceptionUtils.rethrow(new TimeoutException(
                         String.format("could not open connection before timeout %d mls elapsed",
-                                settings.getConnectionTimeoutOnSend())));
+                                currentTimeout)));
             } catch (Exception e) {
                 ExceptionUtils.rethrow(e);
             }
@@ -310,7 +323,7 @@ private CompletableFuture<MessageID> send(@NotNull ByteBuf body, @NotNull Map<St
             if (System.currentTimeMillis() > deadline) {
                 // The method should have checked exception in signature...
                 ExceptionUtils.rethrow(new TimeoutException(String.format("session was not established within %d mls",
-                        settings.getConnectionTimeoutOnSend())));
+                        currentTimeout)));
             }
         }
 
@@ -1167,4 +1180,89 @@ private void debug(String message, Object... args) {
             LOGGER.debug("{} - {}: {}", channel.getSessionGroup(), channel.getSessionAlias(), String.format(message, args));
         }
     }
+
+    @ThreadSafe
+    private static class TimeoutSendHandler {
+        private final Consumer<Event> eventConsumer;
+        private final long maxTimeout;
+        private final long minTimeout;
+        private final ReadWriteLock lock = new ReentrantReadWriteLock();
+        @GuardedBy("lock")
+        private long currentTimeout;
+        @GuardedBy("lock")
+        private int attempts;
+
+        TimeoutSendHandler(long maxTimeout, long minTimeout, Consumer<Event> eventConsumer) {
+            if (maxTimeout < minTimeout) {
+                throw new IllegalArgumentException("max timeout must be greater than min timeout");
+            }
+            if (maxTimeout <= 0) {
+                throw new IllegalArgumentException("connectionTimeoutOnSend must be greater than zero");
+            }
+            if (minTimeout <= 0) {
+                throw new IllegalArgumentException("minConnectionTimeoutOnSend must be greater than zero");
+            }
+            this.maxTimeout = maxTimeout;
+            this.minTimeout = minTimeout;
+            this.eventConsumer = requireNonNull(eventConsumer, "event consumer");
+            currentTimeout = maxTimeout;
+        }
+
+        public void getWithTimeout(Future<?> future) throws ExecutionException, InterruptedException, TimeoutException {
+            try {
+                long timeout = getCurrentTimeout();
+
+                future.get(timeout, TimeUnit.MILLISECONDS);
+
+                int attempts;
+                lock.writeLock().lock();
+                try {
+                    attempts = this.attempts;
+                    this.attempts = 0;
+                    currentTimeout = maxTimeout;
+                } finally {
+                    lock.writeLock().unlock();
+                }
+                if (attempts > 0) {
+                    generateEvent(attempts);
+                }
+            } catch (ExecutionException | InterruptedException | TimeoutException ex) {
+                lock.writeLock().lock();
+                try {
+                    attempts += 1;
+                    currentTimeout = Math.max(minTimeout, currentTimeout / 2);
+                } finally {
+                    lock.writeLock().unlock();
+                }
+                throw ex;
+            }
+        }
+
+        private void generateEvent(int attempts) {
+            eventConsumer.accept(
+                    Event.start().endTimestamp()
+                            .status(Event.Status.FAILED)
+                            .name("Message sending attempt successful after " + attempts + " failed attempt(s)")
+                            .type("MessageSendingAttempts")
+            );
+        }
+
+        public long getCurrentTimeout() {
+            lock.readLock().lock();
+            try {
+                return currentTimeout;
+            } finally {
+                lock.readLock().unlock();
+            }
+        }
+
+        public long getDeadline() {
+            lock.readLock().lock();
+            try {
+                return System.currentTimeMillis() + currentTimeout;
+            } finally {
+                lock.readLock().unlock();
+            }
+        }
+    }
 }

src/main/java/com/exactpro/th2/FixHandlerSettings.java

@@ -75,6 +75,8 @@ public class FixHandlerSettings implements IHandlerSettings {
      */
     private long connectionTimeoutOnSend = DEFAULT_CONNECTION_TIMEOUT_ON_SEND;
 
+    private long minConnectionTimeoutOnSend = 1_000;
+
     @JsonDeserialize(using = DateTimeFormatterDeserializer.class)
     private DateTimeFormatter sendingDateTimeFormat = DateTimeFormatter.ofPattern("yyyyMMdd-HH:mm:ss.SSSSSSSSS");
 
@@ -317,4 +319,12 @@ public long getConnectionTimeoutOnSend() {
     public void setConnectionTimeoutOnSend(long connectionTimeoutOnSend) {
         this.connectionTimeoutOnSend = connectionTimeoutOnSend;
     }
+
+    public long getMinConnectionTimeoutOnSend() {
+        return minConnectionTimeoutOnSend;
+    }
+
+    public void setMinConnectionTimeoutOnSend(long minConnectionTimeoutOnSend) {
+        this.minConnectionTimeoutOnSend = minConnectionTimeoutOnSend;
+    }
 }

src/test/java/com/exactpro/th2/FixHandlerSendTimeoutTest.java

@@ -59,6 +59,7 @@ void sendTimeoutOnConnectionOpen() {
         settings.setPort(42);
         settings.setHost("localhost");
         settings.setConnectionTimeoutOnSend(300); // 300 millis
+        settings.setMinConnectionTimeoutOnSend(100);
         Mockito.when(contextMock.getSettings())
                 .thenReturn(settings);
         var fixHandler = new FixHandler(contextMock);
@@ -100,6 +101,7 @@ void sendTimeoutOnSessionEnabled() {
         settings.setPort(42);
         settings.setHost("localhost");
         settings.setConnectionTimeoutOnSend(300); // 300 millis
+        settings.setMinConnectionTimeoutOnSend(100);
         LocalTime currentTime = LocalTime.now(ZoneOffset.UTC);
         int deltaMinutes = currentTime.isAfter(LocalTime.NOON)
                 ? -1