feat: netty for rest client

Author: NguyenHoangSon96

examples/src/main/java/com/influxdb/v3/netty/rest/ClientChannelInitializer.java

@@ -0,0 +1,63 @@
+/*
+ * The MIT License
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+ * THE SOFTWARE.
+ */
+package com.influxdb.v3.netty.rest;
+
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.handler.codec.http.FullHttpResponse;
+import io.netty.handler.codec.http.HttpClientCodec;
+import io.netty.handler.codec.http.HttpObjectAggregator;
+import io.netty.handler.ssl.SslContext;
+import io.netty.util.concurrent.Promise;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+public class ClientChannelInitializer extends ChannelInitializer<SocketChannel> {
+
+    private final SslContext sslCtx;
+
+    private final Promise<FullHttpResponse> promise;
+
+    private final String host;
+
+    private final Integer port;
+
+    public ClientChannelInitializer(@Nonnull String host, @Nonnull Integer port, @Nonnull Promise<FullHttpResponse> promise, @Nullable SslContext sslCtx) {
+        this.sslCtx = sslCtx;
+        this.promise = promise;
+        this.host = host;
+        this.port = port;
+    }
+
+    @Override
+    public void initChannel(SocketChannel ch) {
+        ChannelPipeline p = ch.pipeline();
+        if (sslCtx != null) {
+            p.addLast("ssl", sslCtx.newHandler(ch.alloc(), host, port));
+        }
+        p.addLast(new HttpClientCodec());
+        p.addLast(new HttpObjectAggregator(1048576));
+        p.addLast(new ClientHandler(this.promise));
+    }
+}

examples/src/main/java/com/influxdb/v3/netty/rest/ClientHandler.java

@@ -0,0 +1,65 @@
+/*
+ * The MIT License
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+ * THE SOFTWARE.
+ */
+package com.influxdb.v3.netty.rest;
+
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.handler.codec.http.*;
+import io.netty.util.CharsetUtil;
+import io.netty.util.concurrent.Promise;
+
+public class ClientHandler extends SimpleChannelInboundHandler<FullHttpResponse> {
+
+    private final Promise<FullHttpResponse> promise;
+
+    public ClientHandler(Promise<FullHttpResponse> promise) {
+        this.promise = promise;
+    }
+
+    @Override
+    public void channelRead0(ChannelHandlerContext ctx, FullHttpResponse msg) {
+//        if (msg instanceof HttpResponse) {
+//            HttpResponse response = (HttpResponse) msg;
+//            System.err.println("{ START CONTENT");
+//        }
+//        if (msg instanceof HttpContent) {
+//            HttpContent content = (HttpContent) msg;
+//            System.err.print(content.content().toString(CharsetUtil.UTF_8));
+//            System.err.flush();
+//
+//            if (content instanceof LastHttpContent) {
+//                System.err.println("} END OF CONTENT");
+//            }
+//        }
+//        System.out.println(msg);
+        this.promise.trySuccess(msg.retain());
+
+    }
+
+    @Override
+    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
+        this.promise.tryFailure(cause);
+        cause.printStackTrace();
+        ctx.close();
+    }
+
+}

examples/src/main/java/com/influxdb/v3/netty/rest/Main.java

@@ -0,0 +1,90 @@
+/*
+ * The MIT License
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+ * THE SOFTWARE.
+ */
+package com.influxdb.v3.netty.rest;
+
+import com.influxdb.v3.client.InfluxDBClient;
+import com.influxdb.v3.client.Point;
+import com.influxdb.v3.client.config.ClientConfig;
+import io.netty.handler.codec.http.FullHttpResponse;
+import io.netty.handler.codec.http.HttpMethod;
+
+import javax.net.ssl.SSLException;
+import java.net.URISyntaxException;
+import java.util.UUID;
+import java.util.concurrent.ExecutionException;
+
+
+public class Main {
+    public static void main(String[] args) throws URISyntaxException, SSLException, ExecutionException, InterruptedException {
+        ClientConfig clientConfig = configCloud();
+
+        var testId = UUID.randomUUID().toString();
+        // Temporarily use `ClientConfig` as a constructor argument.
+        try (RestClient restClient = new RestClient(clientConfig)) {
+
+            // Get server version.
+            System.out.println("Server version: " + restClient.getServerVersion());
+
+            // Write data
+            System.out.println("Write data with testId " + testId);
+            var p = Point.measurement("cpu_sonnh")
+                    .setTag("host", "server1")
+                    .setField("usage_idle", 90.0f)
+                    .setField("testId", testId);
+            var lineProtocol = p.toLineProtocol();
+            restClient.write(lineProtocol);
+
+            // Read data
+            System.out.println("Read data with testId " + testId);
+            String query = String.format("SELECT * FROM \"cpu_sonnh\" WHERE \"testId\" = '%s'", testId);
+            InfluxDBClient influxDBClient = InfluxDBClient.getInstance(clientConfig);
+            var stream = influxDBClient.queryPoints(query);
+            stream.findFirst().ifPresent(pointValues -> System.out.println("Field usage_idle: " + pointValues.getField("usage_idle")));
+
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    public static ClientConfig configCloud() {
+        String url = System.getenv("TESTING_INFLUXDB_URL");
+        String token = System.getenv("TESTING_INFLUXDB_TOKEN");
+        String database = System.getenv("TESTING_INFLUXDB_DATABASE");
+        return new ClientConfig.Builder()
+                .host(url)
+                .token(token.toCharArray())
+                .database(database)
+                .build();
+    }
+
+    // This is a docker container ran from scripts/influxdb-setup.sh, get the token and database, url from that script
+    public static ClientConfig configLocal() {
+        String url = "localhost";
+        String token = "apiv3_sMYBS-vRxl6UDMylb7m2u64G6R7g61jlGL76XnUJY3EaN4MD0tZd4DZOBhe6j-dYtoVhrC6PqGgI9Xiv8d3Psw";
+        String database = System.getenv("bucket0");
+        return new ClientConfig.Builder()
+                .host(url)
+                .token(token.toCharArray())
+                .database(database)
+                .build();
+    }
+}