feat/netty client

examples/src/main/java/com/influxdb/v3/ProxyExample.java

@@ -21,52 +21,96 @@
  */
 package com.influxdb.v3;
 
+import java.net.InetSocketAddress;
+import java.net.URI;
 import java.util.UUID;
+import java.util.function.Supplier;
 import java.util.stream.Stream;
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import io.grpc.HttpConnectProxiedSocketAddress;
+import io.grpc.ProxyDetector;
+import io.netty.handler.proxy.HttpProxyHandler;
 
 import com.influxdb.v3.client.InfluxDBClient;
 import com.influxdb.v3.client.Point;
 import com.influxdb.v3.client.PointValues;
 import com.influxdb.v3.client.config.ClientConfig;
+import com.influxdb.v3.client.config.NettyHttpClientConfig;
 
 public final class ProxyExample {
 
     private ProxyExample() {
     }
 
     public static void main(final String[] args) throws Exception {
-        // Run docker-compose.yml file to start Envoy proxy
+        // Run the docker-compose.yml file to start Envoy proxy,
+        // or start envoy proxy directly with the command `envoy-c envoy.yaml`
 
         String proxyUrl = "http://localhost:10000";
-        String sslRootsFilePath = "src/test/java/com/influxdb/v3/client/testdata/influxdb-certificate.pem";
+        String targetUrl = "http://localhost:8086";
+        String username = "username";
+        String password = "password";
+
+        NettyHttpClientConfig nettyHttpClientConfig = new NettyHttpClientConfig();
+
+        // Set proxy for write api
+        Supplier<HttpProxyHandler> writeApiProxy = () ->
+                new HttpProxyHandler(new InetSocketAddress("localhost", 10000), username, password);
+        nettyHttpClientConfig.configureChannelProxy(writeApiProxy);
+
+        // Set proxy for query api
+        ProxyDetector proxyDetector = createProxyDetector(targetUrl, proxyUrl, username, password);
+        nettyHttpClientConfig.configureManagedChannelProxy(proxyDetector);
+
         ClientConfig clientConfig = new ClientConfig.Builder()
                 .host(System.getenv("INFLUXDB_URL"))
                 .token(System.getenv("INFLUXDB_TOKEN").toCharArray())
                 .database(System.getenv("INFLUXDB_DATABASE"))
-                .proxyUrl(proxyUrl)
-                .sslRootsFilePath(sslRootsFilePath)
+                .nettyHttpClientConfig(nettyHttpClientConfig)
                 .build();
 
-        InfluxDBClient influxDBClient = InfluxDBClient.getInstance(clientConfig);
-        String testId = UUID.randomUUID().toString();
-        Point point = Point.measurement("My_Home")
-                .setTag("room", "Kitchen")
-                .setField("temp", 12.7)
-                .setField("hum", 37)
-                .setField("testId", testId);
-        influxDBClient.writePoint(point);
+        try (InfluxDBClient influxDBClient = InfluxDBClient.getInstance(clientConfig)) {
+            String testId = UUID.randomUUID().toString();
+            Point point = Point.measurement("My_Home")
+                    .setTag("room", "Kitchen")
+                    .setField("temp", 12.7)
+                    .setField("hum", 37)
+                    .setField("testId", testId);
+            influxDBClient.writePoint(point);
 
-        String query = String.format("SELECT * FROM \"My_Home\" WHERE \"testId\" = '%s'", testId);
-        try (Stream<PointValues> stream = influxDBClient.queryPoints(query)) {
-            stream.findFirst().ifPresent(values -> {
-                assert values.getTimestamp() != null;
-                System.out.printf("room[%s]: %s, temp: %3.2f, hum: %d",
-                        new java.util.Date(values.getTimestamp().longValue() / 1000000),
-                        values.getTag("room"),
-                        (Double) values.getField("temp"),
-                        (Long) values.getField("hum"));
-            });
+            String query = String.format("SELECT * FROM \"My_Home\" WHERE \"testId\" = '%s'", testId);
+            try (Stream<PointValues> stream = influxDBClient.queryPoints(query)) {
+                stream.findFirst().ifPresent(values -> {
+                    assert values.getTimestamp() != null;
+                    System.out.printf("room[%s]: %s, temp: %3.2f, hum: %d",
+                            new java.util.Date(values.getTimestamp().longValue() / 1000000),
+                            values.getTag("room"),
+                            (Double) values.getField("temp"),
+                            (Long) values.getField("hum"));
+                });
+            }
         }
     }
+
+    public static ProxyDetector createProxyDetector(@Nonnull final String targetUrl, @Nonnull final String proxyUrl,
+                                                    @Nullable final String username, @Nullable final String password) {
+        URI targetUri = URI.create(targetUrl);
+        URI proxyUri = URI.create(proxyUrl);
+        return (targetServerAddress) -> {
+            InetSocketAddress targetAddress = (InetSocketAddress) targetServerAddress;
+            if (targetUri.getHost().equals(targetAddress.getHostString())
+                    && targetUri.getPort() == targetAddress.getPort()) {
+                return HttpConnectProxiedSocketAddress.newBuilder()
+                        .setProxyAddress(new InetSocketAddress(proxyUri.getHost(), proxyUri.getPort()))
+                        .setTargetAddress(targetAddress)
+                        .setUsername(username)
+                        .setPassword(password)
+                        .build();
+            }
+            return null;
+        };
+    }
 }
 

examples/src/main/java/com/influxdb/v3/durable/DurableExample.java

@@ -21,6 +21,7 @@
  */
 package com.influxdb.v3.durable;
 
+import java.net.URISyntaxException;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.ExecutorService;
@@ -30,6 +31,7 @@
 import java.util.concurrent.locks.LockSupport;
 import java.util.logging.Logger;
 import java.util.stream.Stream;
+import javax.net.ssl.SSLException;
 
 import com.influxdb.v3.client.InfluxDBApiException;
 import com.influxdb.v3.client.InfluxDBClient;
@@ -124,8 +126,13 @@ public static void main(final String[] args) {
           }
 
           // borrow then return a client
-          InfluxDBClient client = clientPool.borrowClient();
-          try {
+            InfluxDBClient client = null;
+            try {
+                client = clientPool.borrowClient();
+            } catch (URISyntaxException | SSLException e) {
+                throw new RuntimeException(e);
+            }
+            try {
             logger.info(" [writeTaskPointsOK " + count + "] Writing " + points.size()
               + " points with client " + client.hashCode());
             client.writePoints(points);
@@ -159,8 +166,13 @@ public static void main(final String[] args) {
             }
           }
           // borrow a client from the pool
-          InfluxDBClient client = clientPool.borrowClient();
-          try {
+            InfluxDBClient client = null;
+            try {
+                client = clientPool.borrowClient();
+            } catch (URISyntaxException | SSLException e) {
+                throw new RuntimeException(e);
+            }
+            try {
             logger.info("[writeErrorRecover " + count + "] Writing " + lps.size()
               + " lps with client " + client.hashCode());
             client.writeRecords(lps);
@@ -183,9 +195,14 @@ public static void main(final String[] args) {
         int count = 0;
         while (!shutdownAll.get()) {
           // borrow a client from the pool
-          InfluxDBClient client = clientPool.borrowClient();
+            InfluxDBClient client = null;
+            try {
+                client = clientPool.borrowClient();
+            } catch (URISyntaxException | SSLException e) {
+                throw new RuntimeException(e);
+            }
 
-          // initiate the query and process the results
+            // initiate the query and process the results
           try (Stream<PointValues> pvs = client.queryPoints(query)) {
             logger.info("[queryOK " + count + "] with client " + client.hashCode()
               + ": query returned " + pvs.toArray().length + " records");
@@ -208,8 +225,13 @@ public static void main(final String[] args) {
         int count = 0;
         while (!shutdownAll.get()) {
           // borrow a client from the pool
-          InfluxDBClient client = clientPool.borrowClient();
-          // every third query attempt results in an error
+            InfluxDBClient client = null;
+            try {
+                client = clientPool.borrowClient();
+            } catch (URISyntaxException | SSLException e) {
+                throw new RuntimeException(e);
+            }
+            // every third query attempt results in an error
           String effectiveQuery = count > 0 && count % 3 == 0 ? badQuery : query;
 
           // attempt to execute the query and process the results

examples/src/main/java/com/influxdb/v3/durable/InfluxClientPool.java

@@ -21,10 +21,12 @@
  */
 package com.influxdb.v3.durable;
 
+import java.net.URISyntaxException;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Stack;
 import java.util.logging.Logger;
+import javax.net.ssl.SSLException;
 
 import com.influxdb.v3.client.InfluxDBClient;
 import com.influxdb.v3.client.config.ClientConfig;
@@ -75,7 +77,7 @@ public InfluxClientPool(final ClientConfig clientConfig, final int maxSize) {
    *
    * @return - An InfluxDBClient ready for use.
    */
-  public synchronized InfluxDBClient borrowClient() {
+  public synchronized InfluxDBClient borrowClient() throws URISyntaxException, SSLException {
     InfluxDBClient client;
     if (idlers.isEmpty()) {
       client = InfluxDBClient.getInstance(clientConfig);

pom.xml

@@ -165,6 +165,10 @@
                     <groupId>com.google.code.gson</groupId>
                     <artifactId>gson</artifactId>
                 </exclusion>
+                <exclusion>
+                    <groupId>io.netty</groupId>
+                    <artifactId>netty-handler-proxy</artifactId>
+                </exclusion>
             </exclusions>
         </dependency>
 
@@ -174,6 +178,12 @@
             <version>${netty-handler.version}</version>
         </dependency>
 
+        <dependency>
+            <groupId>io.netty</groupId>
+            <artifactId>netty-handler-proxy</artifactId>
+            <version>${netty-handler.version}</version>
+        </dependency>
+
         <dependency>
             <groupId>io.netty</groupId>
             <artifactId>netty-buffer</artifactId>
@@ -417,7 +427,7 @@
                                 **/target/**, **/*.jar, **/.git/**, **/.*, **/*.png, **/*.iml, **/*.bolt, .idea/**,
                                 **/*nightly*/**, **/.m2/**, LICENSE, **/*.md, **/.github/**, license_header.txt,
                                 release.properties/, **/pom.xml.releaseBackup, **/pom.xml.tag, **/semantic.yml,
-                                .circleci/config.yml, **/*.pem
+                                .circleci/config.yml, **/*.pem, **/*.p12
                             </excludes>
                         </licenseSet>
                     </licenseSets>

src/main/java/com/influxdb/v3/client/InfluxDBApiNettyException.java

@@ -0,0 +1,104 @@
+/*
+ * 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.client;
+
+import java.util.List;
+import javax.annotation.Nullable;
+
+import io.netty.handler.codec.http.HttpHeaders;
+
+/**
+ * The InfluxDBApiNettyException gets thrown whenever an error status is returned
+ * in the HTTP response.  It facilitates recovering from such errors whenever possible.
+ */
+public class InfluxDBApiNettyException extends InfluxDBApiException {
+
+    /**
+     * The HTTP headers associated with the error.
+     */
+    HttpHeaders headers;
+    /**
+     * The HTTP status code associated with the error.
+     */
+    int statusCode;
+
+    /**
+     * Construct a new InfluxDBApiNettyException with statusCode and headers.
+     *
+     * @param message    the detail message.
+     * @param headers    headers returned in the response.
+     * @param statusCode statusCode of the response.
+     */
+    public InfluxDBApiNettyException(
+            @Nullable final String message,
+            @Nullable final HttpHeaders headers,
+            final int statusCode) {
+        super(message);
+        this.headers = headers;
+        this.statusCode = statusCode;
+    }
+
+    /**
+     * Construct a new InfluxDBApiNettyException with statusCode and headers.
+     *
+     * @param cause      root cause of the exception.
+     * @param headers    headers returned in the response.
+     * @param statusCode status code of the response.
+     */
+    public InfluxDBApiNettyException(
+            @Nullable final Throwable cause,
+            @Nullable final HttpHeaders headers,
+            final int statusCode) {
+        super(cause);
+        this.headers = headers;
+        this.statusCode = statusCode;
+    }
+
+    /**
+     * Gets the HTTP headers property associated with the error.
+     *
+     * @return - the headers object.
+     */
+    public HttpHeaders headers() {
+        return headers;
+    }
+
+    /**
+     * Helper method to simplify retrieval of specific headers.
+     *
+     * @param name - name of the header.
+     * @return - value matching the header key, or null if the key does not exist.
+     */
+    public List<String> getHeader(final String name) {
+        return headers.getAll(name);
+    }
+
+    /**
+     * Gets the HTTP statusCode associated with the error.
+     *
+     * @return - the HTTP statusCode.
+     */
+    public int statusCode() {
+        return statusCode;
+    }
+
+}