diff --git a/tjahzi-core/pom.xml b/tjahzi-core/pom.xml
index 01d885c..0050a1a 100644
--- a/tjahzi-core/pom.xml
+++ b/tjahzi-core/pom.xml
@@ -52,6 +52,7 @@
org.assertj
assertj-core
3.15.0
+ test
@@ -61,6 +62,13 @@
test
+
+ org.apache.logging.log4j
+ log4j-slf4j-impl
+ 2.13.3
+ test
+
+
org.junit.jupiter
diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/BlockingRetry.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/BlockingRetry.java
index 46bc47c..72f2ef8 100644
--- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/BlockingRetry.java
+++ b/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 operation;
private int retries;
- BlockingRetry(Consumer operation, int retries) {
+ // VisibleForTesting
+ BlockingRetry(
+ Consumer operation,
+ ExponentialBackoffStrategy strategy,
+ int retries) {
this.operation = operation;
+ this.strategy = strategy;
this.retries = retries;
}
+ BlockingRetry(Consumer operation, int retries) {
+ this(
+ operation,
+ ExponentialBackoffStrategy.withDefault(),
+ retries
+ );
+ }
+
@Override
public void retry() {
if (retries > 0) {
diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/EventLoopGroupRetry.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/EventLoopGroupRetry.java
index 5533955..37707d9 100644
--- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/EventLoopGroupRetry.java
+++ b/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 operation;
- private final ExponentialBackoffStrategy strategy = ExponentialBackoffStrategy.withDefault();
+ private final ExponentialBackoffStrategy strategy;
- public EventLoopGroupRetry(NioEventLoopGroup group, Consumer operation) {
+ // VisibleForTesting
+ public EventLoopGroupRetry(
+ ScheduledExecutorService group,
+ Consumer operation,
+ ExponentialBackoffStrategy strategy) {
this.group = group;
this.operation = operation;
+ this.strategy = strategy;
+ }
+
+ public EventLoopGroupRetry(ScheduledExecutorService group, Consumer operation) {
+ this(
+ group,
+ operation,
+ ExponentialBackoffStrategy.withDefault()
+ );
}
@Override
diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpConnection.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpConnection.java
index 7b86930..e6d3b2a 100644
--- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpConnection.java
+++ b/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();
diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/NettyHttpClient.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/NettyHttpClient.java
index a808438..27282cc 100644
--- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/NettyHttpClient.java
+++ b/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,
diff --git a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/HeadersTest.java b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/HeadersTest.java
index 800cd1d..b165b48 100644
--- a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/HeadersTest.java
+++ b/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)
));
diff --git a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemTest.java b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemTest.java
index b0f53c7..17e813c 100644
--- a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemTest.java
+++ b/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();
diff --git a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/ReconnectTest.java b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/ReconnectTest.java
index 00ef198..73f51ab 100644
--- a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/ReconnectTest.java
+++ b/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"))
+ )
+ );
+ }
}
diff --git a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/http/BlockingRetryTest.java b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/http/BlockingRetryTest.java
new file mode 100644
index 0000000..7d497e9
--- /dev/null
+++ b/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);
+ }
+}
diff --git a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/http/EventLoopGroupRetryTest.java b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/http/EventLoopGroupRetryTest.java
new file mode 100644
index 0000000..1f72760
--- /dev/null
+++ b/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);
+ });
+ }
+}