Re #8: Added tests and check if channel().isActive() before using it.

tjahzi-core/pom.xml

@@ -52,6 +52,7 @@
             <groupId>org.assertj</groupId>
             <artifactId>assertj-core</artifactId>
             <version>3.15.0</version>
+            <scope>test</scope>
         </dependency>
 
         <dependency>
@@ -61,6 +62,13 @@
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <groupId>org.apache.logging.log4j</groupId>
+            <artifactId>log4j-slf4j-impl</artifactId>
+            <version>2.13.3</version>
+            <scope>test</scope>
+        </dependency>
+
         <!-- Testcontainers -->
         <dependency>
             <groupId>org.junit.jupiter</groupId>

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

@@ -4,16 +4,29 @@ import java.util.function.Consumer;
 
 class BlockingRetry implements Retry {
 
-    private final ExponentialBackoffStrategy strategy = ExponentialBackoffStrategy.withDefault();
+    private final ExponentialBackoffStrategy strategy;
     private final Consumer<BlockingRetry> operation;
 
     private int retries;
 
-    BlockingRetry(Consumer<BlockingRetry> operation, int retries) {
+    // VisibleForTesting
+    BlockingRetry(
+            Consumer<BlockingRetry> operation,
+            ExponentialBackoffStrategy strategy,
+            int retries) {
         this.operation = operation;
+        this.strategy = strategy;
         this.retries = retries;
     }
 
+    BlockingRetry(Consumer<BlockingRetry> operation, int retries) {
+        this(
+                operation,
+                ExponentialBackoffStrategy.withDefault(),
+                retries
+        );
+    }
+
     @Override
     public void retry() {
         if (retries > 0) {

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

@@ -1,19 +1,31 @@
 package pl.tkowalcz.tjahzi.http;
 
-import io.netty.channel.nio.NioEventLoopGroup;
-
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 
 class EventLoopGroupRetry implements Retry, Runnable {
 
-    private final NioEventLoopGroup group;
+    private final ScheduledExecutorService group;
     private final Consumer<EventLoopGroupRetry> operation;
-    private final ExponentialBackoffStrategy strategy = ExponentialBackoffStrategy.withDefault();
+    private final ExponentialBackoffStrategy strategy;
 
-    public EventLoopGroupRetry(NioEventLoopGroup group, Consumer<EventLoopGroupRetry> operation) {
+    // VisibleForTesting
+    public EventLoopGroupRetry(
+            ScheduledExecutorService group,
+            Consumer<EventLoopGroupRetry> operation,
+            ExponentialBackoffStrategy strategy) {
         this.group = group;
         this.operation = operation;
+        this.strategy = strategy;
+    }
+
+    public EventLoopGroupRetry(ScheduledExecutorService group, Consumer<EventLoopGroupRetry> operation) {
+        this(
+                group,
+                operation,
+                ExponentialBackoffStrategy.withDefault()
+        );
     }
 
     @Override

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

@@ -54,9 +54,11 @@ public class HttpConnection implements Closeable {
     }
 
     private void execute(FullHttpRequest request, Retry retry) {
-        lokiConnection.awaitUninterruptibly();
-        if (lokiConnection.isSuccess()) {
-            lokiConnection.channel().writeAndFlush(request);
+        ChannelFuture stableReference = this.lokiConnection;
+
+        stableReference.awaitUninterruptibly();
+        if (stableReference.isSuccess() && stableReference.channel().isActive()) {
+            stableReference.channel().writeAndFlush(request);
         } else {
             // TODO: increase error counters
             retry.retry();

tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/NettyHttpClient.java

@@ -15,9 +15,7 @@ public class NettyHttpClient implements Closeable {
     private final HttpHeaders headers;
 
     private final Snappy snappy = new Snappy();
-
-    //        private volatile ChannelFuture lokiConnection;
-    private HttpConnection lokiConnection;
+    private final HttpConnection lokiConnection;
 
     public NettyHttpClient(
             ClientConfiguration clientConfiguration,

tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/HeadersTest.java

@@ -54,7 +54,7 @@ class HeadersTest {
     void shouldIncludeAdditionalHeaders() {
         // Given
         wireMockServer.stubFor(
-                get(urlEqualTo("/loki/api/v1/push"))
+                post(urlEqualTo("/loki/api/v1/push"))
                         .willReturn(
                                 aResponse().withStatus(200)
                         ));
@@ -101,7 +101,7 @@ class HeadersTest {
     void shouldHandleCaseWithNoAdditionalHeaders() {
         // Given
         wireMockServer.stubFor(
-                get(urlEqualTo("/loki/api/v1/push"))
+                post(urlEqualTo("/loki/api/v1/push"))
                         .willReturn(
                                 aResponse().withStatus(200)
                         ));

tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemTest.java

@@ -55,8 +55,7 @@ class LoggingSystemTest {
 
         NettyHttpClient httpClient = HttpClientFactory.defaultFactory()
                 .getHttpClient(
-                        clientConfiguration,
-                        new String[0]
+                        clientConfiguration
                 );
 
         TjahziInitializer initializer = new TjahziInitializer();

tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/ReconnectTest.java

@@ -31,7 +31,7 @@ public class ReconnectTest {
         );
 
         wireMockServer.stubFor(
-                get(urlEqualTo("/loki/api/v1/push"))
+                post(urlEqualTo("/loki/api/v1/push"))
                         .willReturn(
                                 aResponse().withStatus(200)
                         ));
@@ -54,7 +54,7 @@ public class ReconnectTest {
     }
 
     @Test
-    void shouldIncludeAdditionalHeaders() {
+    void shouldEventuallyReconnectIfLokiWasDownWhenStarting() {
         // Given
         ClientConfiguration clientConfiguration = ClientConfiguration.builder()
                 .withConnectionTimeoutMillis(10_000)
@@ -91,4 +91,64 @@ public class ReconnectTest {
                         )
                 );
     }
+
+    @Test
+    void shouldReconnectIfLokiFailed() {
+        // Given
+        ClientConfiguration clientConfiguration = ClientConfiguration.builder()
+                .withConnectionTimeoutMillis(10_000)
+                .withHost("localhost")
+                .withPort(12322)
+                .withMaxRetries(10)
+                .build();
+
+        NettyHttpClient httpClient = HttpClientFactory.defaultFactory()
+                .getHttpClient(clientConfiguration);
+
+        wireMockServer.start();
+
+        loggingSystem = initializer.createLoggingSystem(
+                httpClient,
+                1024 * 1024,
+                false
+        );
+
+        TjahziLogger logger = loggingSystem.createLogger();
+        logger.log(
+                System.currentTimeMillis(),
+                Map.of(),
+                "Test"
+        );
+
+        await()
+                .atMost(Durations.FIVE_SECONDS)
+                .untilAsserted(() ->
+                        wireMockServer.verify(
+                                1,
+                                postRequestedFor(urlMatching("/loki/api/v1/push"))
+                        )
+                );
+
+        wireMockServer.stop();
+        await().until(() -> !wireMockServer.isRunning());
+
+        // When
+        logger.log(
+                System.currentTimeMillis(),
+                Map.of(),
+                "Test"
+        );
+
+        wireMockServer.start();
+
+        // Then
+        await()
+                .atMost(Durations.TEN_MINUTES)
+                .untilAsserted(() ->
+                        wireMockServer.verify(
+                                1,
+                                postRequestedFor(urlMatching("/loki/api/v1/push"))
+                        )
+                );
+    }
 }

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

@@ -0,0 +1,51 @@
+package pl.tkowalcz.tjahzi.http;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class BlockingRetryTest {
+
+    @Test
+    void shouldRetryAndInvokeOperation() {
+        // Given
+        AtomicInteger retryCounter = new AtomicInteger();
+
+        BlockingRetry retry = new BlockingRetry(
+                __ -> {
+                    retryCounter.incrementAndGet();
+                    __.retry();
+                },
+                new ExponentialBackoffStrategy(1, 1, 1),
+                5
+        );
+
+        // When
+        retry.retry();
+
+        // Then
+        assertThat(retryCounter).hasValue(5);
+    }
+
+    @Test
+    void shouldNotRetryOnItsOwn() {
+        // Given
+        AtomicInteger retryCounter = new AtomicInteger();
+
+        BlockingRetry retry = new BlockingRetry(
+                __ -> {
+                    retryCounter.incrementAndGet();
+                },
+                new ExponentialBackoffStrategy(1, 1, 1),
+                5
+        );
+
+        // When
+        retry.retry();
+
+        // Then
+        assertThat(retryCounter).hasValue(1);
+    }
+}

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

@@ -0,0 +1,73 @@
+package pl.tkowalcz.tjahzi.http;
+
+import io.netty.channel.nio.NioEventLoopGroup;
+import org.awaitility.Durations;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+class EventLoopGroupRetryTest {
+
+    private NioEventLoopGroup eventLoopGroup;
+
+    @BeforeEach
+    void setUp() {
+        eventLoopGroup = new NioEventLoopGroup(Executors.defaultThreadFactory());
+    }
+
+    @AfterEach
+    void tearDown() {
+        eventLoopGroup.shutdownGracefully();
+    }
+
+    @Test
+    void shouldRetryAndInvokeOperation() {
+        // Given
+        AtomicInteger retryCounter = new AtomicInteger();
+
+        EventLoopGroupRetry retry = new EventLoopGroupRetry(
+                eventLoopGroup,
+                __ -> {
+                    retryCounter.incrementAndGet();
+                    __.retry();
+                },
+                new ExponentialBackoffStrategy(250, 1000, 2)
+        );
+
+        // When
+        retry.retry();
+
+        // Then
+        await().atMost(Durations.TEN_SECONDS).untilAsserted(() -> {
+            assertThat(retryCounter).hasValueGreaterThan(3);
+        });
+    }
+
+    @Test
+    void shouldNotRetryOnItsOwn() {
+        // Given
+        AtomicInteger retryCounter = new AtomicInteger();
+
+        EventLoopGroupRetry retry = new EventLoopGroupRetry(
+                eventLoopGroup,
+                __ -> {
+                    retryCounter.incrementAndGet();
+                },
+                new ExponentialBackoffStrategy(1, 1, 1)
+        );
+
+        // When
+        retry.retry();
+
+        // Then
+        await().atMost(Durations.TEN_SECONDS).untilAsserted(() -> {
+            assertThat(retryCounter).hasValue(1);
+        });
+    }
+}