Re #8: Working on Netty http pipeline.

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

@@ -0,0 +1,38 @@
+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();
+
+        ChannelFuture channelFuture = 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();
+
+        return channelFuture;
+    }
+}

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

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<SocketChannel> {
 
+    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<SocketChannel> {
         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);
     }
 }

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

@@ -0,0 +1,5 @@
+package pl.tkowalcz.tjahzi.http;
+
+public class HttpConnection {
+
+}

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

@@ -1,82 +1,99 @@
 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.ChannelFuture;
 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;
+import java.util.concurrent.TimeUnit;
 
 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 EventLoopGroup group;
+        private volatile ChannelFuture lokiConnection;
+//    private 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
         );
+
+        ThreadGroup threadGroup = new ThreadGroup("Tjahzi Loki client");
+        ThreadFactory threadFactory = r -> new Thread(threadGroup, r, "tjahzi-worker");
+        NioEventLoopGroup group = new NioEventLoopGroup(1, threadFactory);
+        this.group = group;
+
+        foobar(clientConfiguration);
+    }
+
+    private void foobar(ClientConfiguration clientConfiguration) {
+        lokiConnection = BootstrapUtil.initConnection(
+                group,
+                clientConfiguration
+        );
+
+        lokiConnection.addListener(future -> {
+            System.out.println("NettyHttpClient.foobar");
+            if (!future.isSuccess()) {
+                group.schedule(() -> {
+                            foobar(clientConfiguration);
+                        },
+                        1000, TimeUnit.MILLISECONDS);
+                // TODO: increase error counters
+
+            }
+        });
     }
 
     public void log(ByteBuf dataBuffer) {
-        ByteBuf output = PooledByteBufAllocator.DEFAULT.buffer();
-        snappy.encode(dataBuffer, output, dataBuffer.readableBytes());
+        log(dataBuffer, clientConfiguration.getMaxRetries());
+    }
 
-        FullHttpRequest request = new DefaultFullHttpRequest(
-                HttpVersion.HTTP_1_1,
-                HttpMethod.POST,
-                logEndpoint,
-                output
-        );
+    public void log(ByteBuf dataBuffer, int retries) {
+        lokiConnection.awaitUninterruptibly();
+        if (lokiConnection.isSuccess()) {
+            ByteBuf output = PooledByteBufAllocator.DEFAULT.buffer();
+            snappy.encode(dataBuffer, output, dataBuffer.readableBytes());
 
-        request.headers()
-                .add(headers)
-                .set(HttpHeaderNames.CONTENT_TYPE, PROTOBUF_MIME_TYPE)
-                .set(HttpHeaderNames.CONTENT_LENGTH, output.readableBytes())
-                .set(HttpHeaderNames.HOST, host);
+            FullHttpRequest request = new DefaultFullHttpRequest(
+                    HttpVersion.HTTP_1_1,
+                    HttpMethod.POST,
+                    clientConfiguration.getLogEndpoint(),
+                    output
+            );
 
-        channel.writeAndFlush(request);
+            request.headers()
+                    .add(headers)
+                    .set(HttpHeaderNames.CONTENT_TYPE, PROTOBUF_MIME_TYPE)
+                    .set(HttpHeaderNames.CONTENT_LENGTH, output.readableBytes())
+                    .set(HttpHeaderNames.HOST, clientConfiguration.getHost());
+
+            lokiConnection.channel().writeAndFlush(request);
+        } else {
+            // TODO: increase error counters
+
+            if (retries > 0) {
+                log(dataBuffer, retries - 1);
+            }
+
+            // Dropping message on the floor
+        }
     }
 
     @Override

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<FullHttpResponse> {
+
+    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");
+    }
+}

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

@@ -0,0 +1,95 @@
+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()
+                        .dynamicPort()
+                        .dynamicHttpsPort()
+        );
+
+//        wireMockServer.start();
+
+        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 shouldIncludeAdditionalHeaders() {
+        // Given
+//        wireMockServer.stubFor(
+//                get(urlEqualTo("/loki/api/v1/push"))
+//                        .willReturn(
+//                                aResponse().withStatus(200)
+//                        ));
+
+        ClientConfiguration clientConfiguration = ClientConfiguration.builder()
+                .withConnectionTimeoutMillis(10_000)
+                .withHost("localhost")
+                .withPort(12322)
+                .withMaxRetries(1)
+                .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"
+        );
+
+        // Then
+        await()
+                .atMost(Durations.TEN_MINUTES)
+                .untilAsserted(() ->
+                        wireMockServer.verify(
+                                postRequestedFor(urlMatching("/loki/api/v1/push"))
+                                        .withHeader("X-Scope-OrgID", equalTo("Circus"))
+                        ));
+    }
+}