Merge pull request #35 from tkowalcz/28-no-connection-buffer-leak

Re #28: Added failure action to BlockingRetry so that we can dispose …

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);
     }
 }