Re #28: Added failure action to BlockingRetry so that we can dispose of the http request (and ByteBuf with request content). Additionally simplified these retry classes so stats counters are updated outside.

core/src/main/java/pl/tkowalcz/tjahzi/http/BlockingRetry.java

@@ -1,40 +1,39 @@
 package pl.tkowalcz.tjahzi.http;
 
-import pl.tkowalcz.tjahzi.stats.MonitoringModule;
-
 import java.util.function.Consumer;
 
 class BlockingRetry implements Retry {
 
-    private final ExponentialBackoffStrategy strategy;
     private final Consumer<BlockingRetry> operation;
-    private final MonitoringModule monitoringModule;
+    private final Runnable failureAction;
 
+    private final ExponentialBackoffStrategy strategy;
     private int retries;
 
     // VisibleForTesting
     BlockingRetry(
             Consumer<BlockingRetry> operation,
+            Runnable failureAction,
             ExponentialBackoffStrategy strategy,
-            int retries,
-            MonitoringModule monitoringModule
+            int retries
     ) {
         this.operation = operation;
+        this.failureAction = failureAction;
         this.strategy = strategy;
 
         this.retries = retries;
-        this.monitoringModule = monitoringModule;
     }
 
     BlockingRetry(
             Consumer<BlockingRetry> operation,
-            int retries,
-            MonitoringModule monitoringModule) {
+            Runnable failureAction,
+            int retries
+    ) {
         this(
                 operation,
+                failureAction,
                 ExponentialBackoffStrategy.withDefault(),
-                retries,
-                monitoringModule
+                retries
         );
     }
 
@@ -42,7 +41,6 @@ class BlockingRetry implements Retry {
     public void retry() {
         if (retries > 0) {
             retries--;
-            monitoringModule.incrementRetriedHttpRequests();
 
             try {
                 Thread.sleep(strategy.getAsLong(), 0);
@@ -52,7 +50,7 @@ class BlockingRetry implements Retry {
 
             operation.accept(this);
         } else {
-            monitoringModule.incrementFailedHttpRequests();
+            failureAction.run();
         }
     }
 }

core/src/main/java/pl/tkowalcz/tjahzi/http/EventLoopGroupRetry.java

@@ -1,7 +1,5 @@
 package pl.tkowalcz.tjahzi.http;
 
-import pl.tkowalcz.tjahzi.stats.MonitoringModule;
-
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
@@ -12,38 +10,30 @@ class EventLoopGroupRetry implements Retry, Runnable {
     private final Consumer<EventLoopGroupRetry> operation;
     private final ExponentialBackoffStrategy strategy;
 
-    private final MonitoringModule monitoringModule;
-
     // VisibleForTesting
     public EventLoopGroupRetry(
             ScheduledExecutorService group,
             Consumer<EventLoopGroupRetry> operation,
-            ExponentialBackoffStrategy strategy,
-            MonitoringModule monitoringModule
+            ExponentialBackoffStrategy strategy
     ) {
         this.group = group;
         this.operation = operation;
         this.strategy = strategy;
-
-        this.monitoringModule = monitoringModule;
     }
 
     public EventLoopGroupRetry(
             ScheduledExecutorService group,
-            Consumer<EventLoopGroupRetry> operation,
-            MonitoringModule monitoringModule
+            Consumer<EventLoopGroupRetry> operation
     ) {
         this(
                 group,
                 operation,
-                ExponentialBackoffStrategy.withDefault(),
-                monitoringModule
+                ExponentialBackoffStrategy.withDefault()
         );
     }
 
     @Override
     public void run() {
-        monitoringModule.incrementHttpConnectAttempts();
         operation.accept(this);
     }
 

core/src/main/java/pl/tkowalcz/tjahzi/http/HttpConnection.java

@@ -27,8 +27,10 @@ public class HttpConnection implements Closeable {
 
         EventLoopGroupRetry retry = new EventLoopGroupRetry(
                 group,
-                this::recreateConnection,
-                monitoringModule
+                __ -> {
+                    monitoringModule.incrementHttpConnectAttempts();
+                    recreateConnection(__);
+                }
         );
         recreateConnection(retry);
     }
@@ -54,9 +56,15 @@ public class HttpConnection implements Closeable {
 
     public void execute(FullHttpRequest request) {
         BlockingRetry blockingRetry = new BlockingRetry(
-                retry -> execute(request, retry),
-                clientConfiguration.getMaxRetries(),
-                monitoringModule
+                retry -> {
+                    monitoringModule.incrementRetriedHttpRequests();
+                    execute(request, retry);
+                },
+                () -> {
+                    monitoringModule.incrementFailedHttpRequests();
+                    request.release();
+                },
+                clientConfiguration.getMaxRetries()
         );
 
         execute(request, blockingRetry);

core/src/test/java/pl/tkowalcz/tjahzi/http/BlockingRetryTest.java

@@ -1,7 +1,6 @@
 package pl.tkowalcz.tjahzi.http;
 
 import org.junit.jupiter.api.Test;
-import pl.tkowalcz.tjahzi.stats.StandardMonitoringModule;
 
 import java.util.concurrent.atomic.AtomicInteger;
 
@@ -10,43 +9,62 @@ import static org.assertj.core.api.Assertions.assertThat;
 class BlockingRetryTest {
 
     @Test
-    void shouldRetryAndInvokeOperation() {
+    void shouldInvokeOperation() {
         // Given
-        AtomicInteger retryCounter = new AtomicInteger();
-        StandardMonitoringModule monitoringModule = new StandardMonitoringModule();
+        AtomicInteger invocationCounter = new AtomicInteger();
+        AtomicInteger failureCounter = new AtomicInteger();
 
         BlockingRetry retry = new BlockingRetry(
-                __ -> {
-                    retryCounter.incrementAndGet();
-                    __.retry();
-                },
+                __ -> invocationCounter.incrementAndGet(),
+                failureCounter::incrementAndGet,
                 new ExponentialBackoffStrategy(1, 1, 1),
-                5,
-                monitoringModule
+                5
         );
 
         // When
         retry.retry();
 
         // Then
-        assertThat(retryCounter).hasValue(5);
-        assertThat(monitoringModule.getRetriedHttpRequests()).isEqualTo(5);
-        assertThat(monitoringModule.getFailedHttpRequests()).isEqualTo(1);
+        assertThat(invocationCounter).hasValue(1);
+        assertThat(failureCounter).hasValue(0);
+    }
+
+    @Test
+    void shouldRetryAndInvokeFailureCallback() {
+        // Given
+        AtomicInteger invocationCounter = new AtomicInteger();
+        AtomicInteger failureCounter = new AtomicInteger();
+
+        BlockingRetry retry = new BlockingRetry(
+                __ -> {
+                    invocationCounter.incrementAndGet();
+                    __.retry();
+                },
+                failureCounter::incrementAndGet,
+                new ExponentialBackoffStrategy(1, 1, 1),
+                5
+        );
+
+        // When
+        retry.retry();
+
+        // Then
+        assertThat(invocationCounter).hasValue(5);
+        assertThat(failureCounter).hasValue(1);
     }
 
     @Test
     void shouldNotRetryOnItsOwn() {
         // Given
         AtomicInteger retryCounter = new AtomicInteger();
-        StandardMonitoringModule monitoringModule = new StandardMonitoringModule();
+        AtomicInteger failureCounter = new AtomicInteger();
 
         BlockingRetry retry = new BlockingRetry(
-                __ -> {
-                    retryCounter.incrementAndGet();
+                __ -> retryCounter.incrementAndGet(),
+                () -> {
                 },
                 new ExponentialBackoffStrategy(1, 1, 1),
-                5,
-                monitoringModule
+                5
         );
 
         // When
@@ -54,7 +72,6 @@ class BlockingRetryTest {
 
         // Then
         assertThat(retryCounter).hasValue(1);
-        assertThat(monitoringModule.getRetriedHttpRequests()).isEqualTo(1);
-        assertThat(monitoringModule.getFailedHttpRequests()).isEqualTo(0);
+        assertThat(failureCounter).hasValue(0);
     }
 }

core/src/test/java/pl/tkowalcz/tjahzi/http/EventLoopGroupRetryTest.java

@@ -5,7 +5,6 @@ import org.awaitility.Durations;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
-import pl.tkowalcz.tjahzi.stats.StandardMonitoringModule;
 
 import java.util.concurrent.Executors;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -16,12 +15,10 @@ import static org.awaitility.Awaitility.await;
 class EventLoopGroupRetryTest {
 
     private NioEventLoopGroup eventLoopGroup;
-    private StandardMonitoringModule monitoringModule;
 
     @BeforeEach
     void setUp() {
         eventLoopGroup = new NioEventLoopGroup(Executors.defaultThreadFactory());
-        monitoringModule = new StandardMonitoringModule();
     }
 
     @AfterEach
@@ -33,7 +30,6 @@ class EventLoopGroupRetryTest {
     void shouldRetryAndInvokeOperation() {
         // Given
         AtomicInteger retryCounter = new AtomicInteger();
-        StandardMonitoringModule monitoringModule = new StandardMonitoringModule();
 
         EventLoopGroupRetry retry = new EventLoopGroupRetry(
                 eventLoopGroup,
@@ -41,8 +37,7 @@ class EventLoopGroupRetryTest {
                     retryCounter.incrementAndGet();
                     __.retry();
                 },
-                new ExponentialBackoffStrategy(250, 1000, 2),
-                monitoringModule
+                new ExponentialBackoffStrategy(250, 1000, 2)
         );
 
         // When
@@ -52,8 +47,6 @@ class EventLoopGroupRetryTest {
         await().atMost(Durations.TEN_SECONDS).untilAsserted(() -> {
             assertThat(retryCounter).hasValueGreaterThan(3);
         });
-
-        assertThat(monitoringModule.getHttpConnectAttempts()).isGreaterThanOrEqualTo(3);
     }
 
     @Test
@@ -66,8 +59,7 @@ class EventLoopGroupRetryTest {
                 __ -> {
                     retryCounter.incrementAndGet();
                 },
-                new ExponentialBackoffStrategy(1, 1, 1),
-                monitoringModule
+                new ExponentialBackoffStrategy(1, 1, 1)
         );
 
         // When
@@ -77,7 +69,5 @@ class EventLoopGroupRetryTest {
         await().atMost(Durations.TEN_SECONDS).untilAsserted(() -> {
             assertThat(retryCounter).hasValue(1);
         });
-
-        assertThat(monitoringModule.getHttpConnectAttempts()).isEqualTo(1);
     }
 }