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
new file mode 100644
index 0000000..72f2ef8
--- /dev/null
+++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/BlockingRetry.java
@@ -0,0 +1,44 @@
+package pl.tkowalcz.tjahzi.http;
+
+import java.util.function.Consumer;
+
+class BlockingRetry implements Retry {
+
+ private final ExponentialBackoffStrategy strategy;
+ private final Consumer operation;
+
+ private 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) {
+ retries--;
+
+ try {
+ Thread.sleep(strategy.getAsLong(), 0);
+ } catch (InterruptedException ignore) {
+ return;
+ }
+
+ operation.accept(this);
+ }
+ }
+}
diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/BootstrapUtil.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/BootstrapUtil.java
new file mode 100644
index 0000000..c750967
--- /dev/null
+++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/BootstrapUtil.java
@@ -0,0 +1,36 @@
+package pl.tkowalcz.tjahzi.http;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.buffer.PooledByteBufAllocator;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.socket.nio.NioSocketChannel;
+
+public class BootstrapUtil {
+
+ public static ChannelFuture initConnection(
+ EventLoopGroup group,
+ ClientConfiguration clientConfiguration
+ ) {
+ Bootstrap bootstrap = new Bootstrap();
+ ResponseHandler responseHandler = new ResponseHandler();
+
+ return bootstrap.group(group)
+ .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
+ .option(ChannelOption.SO_KEEPALIVE, true)
+ .option(ChannelOption.TCP_NODELAY, false)
+ .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, clientConfiguration.getConnectionTimeoutMillis())
+ .channel(NioSocketChannel.class)
+ .handler(
+ new HttpClientInitializer(
+ responseHandler,
+ clientConfiguration.getRequestTimeoutMillis()
+ )
+ )
+ .remoteAddress(
+ clientConfiguration.getHost(),
+ clientConfiguration.getPort()
+ ).connect();
+ }
+}
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
new file mode 100644
index 0000000..37707d9
--- /dev/null
+++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/EventLoopGroupRetry.java
@@ -0,0 +1,44 @@
+package pl.tkowalcz.tjahzi.http;
+
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
+
+class EventLoopGroupRetry implements Retry, Runnable {
+
+ private final ScheduledExecutorService group;
+ private final Consumer operation;
+ private final ExponentialBackoffStrategy strategy;
+
+ // 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
+ public void run() {
+ operation.accept(this);
+ }
+
+ @Override
+ public void retry() {
+ group.schedule(
+ this,
+ strategy.getAsLong(),
+ TimeUnit.MILLISECONDS
+ );
+ }
+}
diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ExponentialBackoffStrategy.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ExponentialBackoffStrategy.java
new file mode 100644
index 0000000..4a46091
--- /dev/null
+++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ExponentialBackoffStrategy.java
@@ -0,0 +1,38 @@
+package pl.tkowalcz.tjahzi.http;
+
+import java.util.function.LongSupplier;
+
+public class ExponentialBackoffStrategy implements LongSupplier {
+
+ private final long maximumBackoffMillis;
+ private final double multiplier;
+
+ private long nextBackoff;
+
+ public ExponentialBackoffStrategy(
+ long initialBackoffMillis,
+ long maximumBackoffMillis,
+ double multiplier
+ ) {
+ this.maximumBackoffMillis = maximumBackoffMillis;
+ this.multiplier = multiplier;
+
+ this.nextBackoff = initialBackoffMillis;
+ }
+
+ @Override
+ public long getAsLong() {
+ long result = nextBackoff;
+
+ nextBackoff = (long) Math.min(nextBackoff * multiplier, maximumBackoffMillis);
+ return result;
+ }
+
+ public static ExponentialBackoffStrategy withDefault() {
+ return new ExponentialBackoffStrategy(
+ 250,
+ 30_000,
+ 3
+ );
+ }
+}
diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpClientFactory.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpClientFactory.java
index bb4b57a..062639f 100644
--- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpClientFactory.java
+++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpClientFactory.java
@@ -6,7 +6,7 @@ public class HttpClientFactory {
public NettyHttpClient getHttpClient(
ClientConfiguration clientConfiguration,
- String[] additionalHeaders) {
+ String... additionalHeaders) {
return new NettyHttpClient(
clientConfiguration,
additionalHeaders
diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpClientInitializer.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpClientInitializer.java
index 4a2d2ee..c8d447a 100644
--- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpClientInitializer.java
+++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpClientInitializer.java
@@ -10,9 +10,13 @@ import io.netty.handler.timeout.IdleStateHandler;
class HttpClientInitializer extends ChannelInitializer {
+ private static final int BYTES_IN_MEGABYTE = 1024 * 1024;
+
+ private final ResponseHandler responseHandler;
private final int requestTimeoutMillis;
- HttpClientInitializer(int requestTimeoutMillis) {
+ HttpClientInitializer(ResponseHandler responseHandler, int requestTimeoutMillis) {
+ this.responseHandler = responseHandler;
this.requestTimeoutMillis = requestTimeoutMillis;
}
@@ -21,7 +25,8 @@ class HttpClientInitializer extends ChannelInitializer {
ChannelPipeline p = ch.pipeline();
p.addLast(new IdleStateHandler(requestTimeoutMillis, 0, 60));
p.addLast(new HttpClientCodec());
- p.addLast(new HttpObjectAggregator(1048576));
+ p.addLast(new HttpObjectAggregator(BYTES_IN_MEGABYTE));
p.addLast(new HttpContentDecompressor());
+ p.addLast(responseHandler);
}
}
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
new file mode 100644
index 0000000..e6d3b2a
--- /dev/null
+++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpConnection.java
@@ -0,0 +1,73 @@
+package pl.tkowalcz.tjahzi.http;
+
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.handler.codec.http.FullHttpRequest;
+
+import java.io.Closeable;
+import java.util.concurrent.ThreadFactory;
+
+public class HttpConnection implements Closeable {
+
+ private final ClientConfiguration clientConfiguration;
+ private final NioEventLoopGroup group;
+
+ private volatile ChannelFuture lokiConnection;
+
+ public HttpConnection(ClientConfiguration clientConfiguration) {
+ this.clientConfiguration = clientConfiguration;
+
+ ThreadGroup threadGroup = new ThreadGroup("Tjahzi Loki client");
+ ThreadFactory threadFactory = r -> new Thread(threadGroup, r, "tjahzi-worker");
+
+ this.group = new NioEventLoopGroup(1, threadFactory);
+
+ EventLoopGroupRetry retry = new EventLoopGroupRetry(group, this::recreateConnection);
+ recreateConnection(retry);
+ }
+
+ private void recreateConnection(Retry retry) {
+ lokiConnection = BootstrapUtil.initConnection(
+ group,
+ clientConfiguration
+ );
+
+ lokiConnection.addListener((ChannelFuture future) -> {
+ if (future.isSuccess()) {
+ future
+ .channel()
+ .closeFuture()
+ .addListener(__ -> retry.retry());
+ } else {
+ retry.retry();
+ }
+ });
+ }
+
+ public void execute(FullHttpRequest request) {
+ BlockingRetry blockingRetry = new BlockingRetry(
+ retry -> execute(request, retry),
+ clientConfiguration.getMaxRetries()
+ );
+
+ execute(request, blockingRetry);
+ }
+
+ private void execute(FullHttpRequest request, Retry retry) {
+ ChannelFuture stableReference = this.lokiConnection;
+
+ stableReference.awaitUninterruptibly();
+ if (stableReference.isSuccess() && stableReference.channel().isActive()) {
+ stableReference.channel().writeAndFlush(request);
+ } else {
+ // TODO: increase error counters
+ retry.retry();
+ // Dropping message on the floor
+ }
+ }
+
+ @Override
+ public void close() {
+ group.shutdownGracefully();
+ }
+}
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 39c4ee5..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
@@ -1,62 +1,32 @@
package pl.tkowalcz.tjahzi.http;
-import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator;
-import io.netty.channel.Channel;
-import io.netty.channel.ChannelOption;
-import io.netty.channel.EventLoopGroup;
-import io.netty.channel.nio.NioEventLoopGroup;
-import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.compression.Snappy;
import io.netty.handler.codec.http.*;
import java.io.Closeable;
-import java.util.concurrent.ThreadFactory;
public class NettyHttpClient implements Closeable {
public static final String PROTOBUF_MIME_TYPE = "application/x-protobuf";
- private final EventLoopGroup group;
- private final String logEndpoint;
-
- private final Channel channel;
- private final String host;
+ private final ClientConfiguration clientConfiguration;
private final HttpHeaders headers;
private final Snappy snappy = new Snappy();
+ private final HttpConnection lokiConnection;
public NettyHttpClient(
ClientConfiguration clientConfiguration,
String[] additionalHeaders) {
- host = clientConfiguration.getHost();
- logEndpoint = clientConfiguration.getLogEndpoint();
-
- ThreadGroup threadGroup = new ThreadGroup("Tjahzi Loki client");
- ThreadFactory threadFactory = r -> new Thread(threadGroup, r, "tjahzi-worker");
- group = new NioEventLoopGroup(1, threadFactory);
-
- Bootstrap bootstrap = new Bootstrap();
- channel = bootstrap.group(group)
- .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
- .option(ChannelOption.SO_KEEPALIVE, true)
- .option(ChannelOption.TCP_NODELAY, false)
- .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, clientConfiguration.getConnectionTimeoutMillis())
- .channel(NioSocketChannel.class)
- .handler(new HttpClientInitializer(clientConfiguration.getRequestTimeoutMillis()))
- .remoteAddress(
- clientConfiguration.getHost(),
- clientConfiguration.getPort()
- )
- .connect()
- .syncUninterruptibly()
- .channel();
-
- headers = new ReadOnlyHttpHeaders(
+ this.clientConfiguration = clientConfiguration;
+ this.headers = new ReadOnlyHttpHeaders(
true,
additionalHeaders
);
+
+ lokiConnection = new HttpConnection(clientConfiguration);
}
public void log(ByteBuf dataBuffer) {
@@ -66,7 +36,7 @@ public class NettyHttpClient implements Closeable {
FullHttpRequest request = new DefaultFullHttpRequest(
HttpVersion.HTTP_1_1,
HttpMethod.POST,
- logEndpoint,
+ clientConfiguration.getLogEndpoint(),
output
);
@@ -74,13 +44,13 @@ public class NettyHttpClient implements Closeable {
.add(headers)
.set(HttpHeaderNames.CONTENT_TYPE, PROTOBUF_MIME_TYPE)
.set(HttpHeaderNames.CONTENT_LENGTH, output.readableBytes())
- .set(HttpHeaderNames.HOST, host);
+ .set(HttpHeaderNames.HOST, clientConfiguration.getHost());
- channel.writeAndFlush(request);
+ lokiConnection.execute(request);
}
@Override
public void close() {
- group.shutdownGracefully();
+ lokiConnection.close();
}
}
diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ResponseHandler.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ResponseHandler.java
new file mode 100644
index 0000000..5a2e355
--- /dev/null
+++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ResponseHandler.java
@@ -0,0 +1,29 @@
+package pl.tkowalcz.tjahzi.http;
+
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.handler.codec.http.FullHttpResponse;
+
+@ChannelHandler.Sharable
+class ResponseHandler extends SimpleChannelInboundHandler {
+
+ ResponseHandler() {
+ super(false);
+ }
+
+ @Override
+ protected void channelRead0(ChannelHandlerContext ctx, FullHttpResponse msg) {
+ System.out.println("msg = " + msg);
+ }
+
+ @Override
+ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+ System.out.println("cause = " + cause);
+ }
+
+ @Override
+ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+ System.out.println("ResponseHandler.channelInactive");
+ }
+}
diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/Retry.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/Retry.java
new file mode 100644
index 0000000..5b53020
--- /dev/null
+++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/Retry.java
@@ -0,0 +1,6 @@
+package pl.tkowalcz.tjahzi.http;
+
+public interface Retry {
+
+ void retry();
+}
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 1c7ed40..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)
));
@@ -69,10 +69,8 @@ class HeadersTest {
NettyHttpClient httpClient = HttpClientFactory.defaultFactory()
.getHttpClient(
clientConfiguration,
- new String[]{
- "X-Scope-OrgID", "Circus",
- "C", "Control"
- }
+ "X-Scope-OrgID", "Circus",
+ "C", "Control"
);
loggingSystem = initializer.createLoggingSystem(
@@ -103,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)
));
@@ -117,8 +115,7 @@ class HeadersTest {
NettyHttpClient httpClient = HttpClientFactory.defaultFactory()
.getHttpClient(
- clientConfiguration,
- new String[0]
+ clientConfiguration
);
loggingSystem = initializer.createLoggingSystem(
@@ -163,11 +160,9 @@ class HeadersTest {
NettyHttpClient httpClient = HttpClientFactory.defaultFactory()
.getHttpClient(
clientConfiguration,
- new String[]{
- "content-type", "text/plain",
- "content-length", "5232423423",
- "host", "remote"
- }
+ "content-type", "text/plain",
+ "content-length", "5232423423",
+ "host", "remote"
);
loggingSystem = initializer.createLoggingSystem(
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
new file mode 100644
index 0000000..73f51ab
--- /dev/null
+++ b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/ReconnectTest.java
@@ -0,0 +1,154 @@
+package pl.tkowalcz.tjahzi;
+
+import com.github.tomakehurst.wiremock.WireMockServer;
+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.http.ClientConfiguration;
+import pl.tkowalcz.tjahzi.http.HttpClientFactory;
+import pl.tkowalcz.tjahzi.http.NettyHttpClient;
+
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import static com.github.tomakehurst.wiremock.client.WireMock.*;
+import static com.github.tomakehurst.wiremock.core.WireMockConfiguration.wireMockConfig;
+import static org.awaitility.Awaitility.await;
+
+public class ReconnectTest {
+
+ private WireMockServer wireMockServer;
+
+ private TjahziInitializer initializer;
+ private LoggingSystem loggingSystem;
+
+ @BeforeEach
+ void setUp() {
+ wireMockServer = new WireMockServer(
+ wireMockConfig()
+ .port(12322)
+ );
+
+ wireMockServer.stubFor(
+ post(urlEqualTo("/loki/api/v1/push"))
+ .willReturn(
+ aResponse().withStatus(200)
+ ));
+
+ initializer = new TjahziInitializer();
+ }
+
+ @AfterEach
+ void tearDown() {
+ wireMockServer.stop();
+
+ if (loggingSystem != null) {
+ loggingSystem.close(
+ (int) TimeUnit.SECONDS.toMillis(10),
+ System.out::println
+ );
+
+ loggingSystem = null;
+ }
+ }
+
+ @Test
+ void shouldEventuallyReconnectIfLokiWasDownWhenStarting() {
+ // Given
+ ClientConfiguration clientConfiguration = ClientConfiguration.builder()
+ .withConnectionTimeoutMillis(10_000)
+ .withHost("localhost")
+ .withPort(12322)
+ .withMaxRetries(10)
+ .build();
+
+ NettyHttpClient httpClient = HttpClientFactory.defaultFactory()
+ .getHttpClient(clientConfiguration);
+
+ loggingSystem = initializer.createLoggingSystem(
+ httpClient,
+ 1024 * 1024,
+ false
+ );
+
+ // When
+ TjahziLogger logger = loggingSystem.createLogger();
+ logger.log(
+ System.currentTimeMillis(),
+ Map.of(),
+ "Test"
+ );
+
+ wireMockServer.start();
+
+ // Then
+ await()
+ .atMost(Durations.TEN_MINUTES)
+ .untilAsserted(() ->
+ wireMockServer.verify(
+ postRequestedFor(urlMatching("/loki/api/v1/push"))
+ )
+ );
+ }
+
+ @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/ResourcesCleanupOnCloseTest.java b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/ResourcesCleanupOnCloseTest.java
index 53a8932..aadf88b 100644
--- a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/ResourcesCleanupOnCloseTest.java
+++ b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/ResourcesCleanupOnCloseTest.java
@@ -54,10 +54,8 @@ class ResourcesCleanupOnCloseTest {
NettyHttpClient httpClient = HttpClientFactory.defaultFactory()
.getHttpClient(
clientConfiguration,
- new String[]{
- "X-Scope-OrgID", "Circus",
- "C", "Control"
- }
+ "X-Scope-OrgID", "Circus",
+ "C", "Control"
);
LoggingSystem loggingSystem = initializer.createLoggingSystem(
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);
+ });
+ }
+}
diff --git a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/http/ExponentialBackoffStrategyTest.java b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/http/ExponentialBackoffStrategyTest.java
new file mode 100644
index 0000000..1245681
--- /dev/null
+++ b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/http/ExponentialBackoffStrategyTest.java
@@ -0,0 +1,74 @@
+package pl.tkowalcz.tjahzi.http;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class ExponentialBackoffStrategyTest {
+
+ @Test
+ void shouldReturnInitialBackoffOnFirstTry() {
+ // Given
+ int expected = 1000;
+
+ ExponentialBackoffStrategy backoff = new ExponentialBackoffStrategy(
+ expected,
+ 10_000,
+ 2
+ );
+
+ // When
+ long actual = backoff.getAsLong();
+
+ // Then
+ assertThat(actual).isEqualTo(expected);
+ }
+
+ @Test
+ void shouldNotGoOverMaximumBackoff() {
+ // Given
+ int expected = 10_000;
+
+ ExponentialBackoffStrategy backoff = new ExponentialBackoffStrategy(
+ 4000,
+ expected,
+ 2
+ );
+
+ // When
+ for (int i = 0; i < 1000; i++) {
+ backoff.getAsLong();
+ }
+
+ long actual = backoff.getAsLong();
+
+ // Then
+ assertThat(actual).isEqualTo(expected);
+ }
+
+ @Test
+ void shouldIncreaseBackoffExponentially() {
+ // Given
+ ExponentialBackoffStrategy backoff = new ExponentialBackoffStrategy(
+ 250,
+ 10_000,
+ 3
+ );
+
+ ArrayList actual = new ArrayList<>();
+
+ // When
+ actual.add(backoff.getAsLong());
+ actual.add(backoff.getAsLong());
+ actual.add(backoff.getAsLong());
+ actual.add(backoff.getAsLong());
+ actual.add(backoff.getAsLong());
+
+ // Then
+ assertThat(actual).containsExactly(
+ 250L, 750L, 2250L, 6750L, 10_000L
+ );
+ }
+}