Call HttpURLConnection.disconnect when closing response, and avoid double buffering

Author: zhicwu

clickhouse-client/src/main/java/com/clickhouse/client/data/ClickHouseLZ4InputStream.java

@@ -90,7 +90,11 @@ private boolean readFully(byte[] b, int off, int len) throws IOException {
     }
 
     public ClickHouseLZ4InputStream(InputStream stream) {
-        super(null);
+        this(stream, null);
+    }
+
+    public ClickHouseLZ4InputStream(InputStream stream, Runnable afterClose) {
+        super(afterClose);
 
         this.decompressor = factory.fastDecompressor();
         this.stream = ClickHouseChecker.nonNull(stream, "InputStream");

clickhouse-http-client/src/main/java/com/clickhouse/client/http/ClickHouseHttpConnection.java

@@ -20,6 +20,7 @@
 import com.clickhouse.client.ClickHouseCompression;
 import com.clickhouse.client.ClickHouseConfig;
 import com.clickhouse.client.ClickHouseCredentials;
+import com.clickhouse.client.ClickHouseInputStream;
 import com.clickhouse.client.ClickHouseNode;
 import com.clickhouse.client.ClickHouseRequest;
 import com.clickhouse.client.ClickHouseUtils;
@@ -203,6 +204,14 @@ protected ClickHouseHttpConnection(ClickHouseNode server, ClickHouseRequest<?> r
         this.defaultHeaders = Collections.unmodifiableMap(map);
     }
 
+    protected void closeQuietly() {
+        try {
+            close();
+        } catch (Exception e) {
+            // ignore
+        }
+    }
+
     protected String getBaseUrl() {
         String baseUrl;
         int index = url.indexOf('?');
@@ -236,23 +245,30 @@ protected OutputStream getRequestOutputStream(OutputStream out) throws IOExcepti
         return out;
     }
 
-    protected InputStream getResponseInputStream(InputStream in) throws IOException {
+    protected ClickHouseInputStream getResponseInputStream(InputStream in) throws IOException {
+        Runnable afterClose = null;
+        if (!isReusable()) {
+            afterClose = this::closeQuietly;
+        }
+        ClickHouseInputStream chInput;
         if (config.isCompressServerResponse()) {
             // TODO support more algorithms
             ClickHouseCompression algorithm = config.getCompressAlgorithmForServerResponse();
             switch (algorithm) {
                 case GZIP:
-                    in = new GZIPInputStream(in);
+                    chInput = ClickHouseInputStream.of(new GZIPInputStream(in), config.getMaxBufferSize(), afterClose);
                     break;
                 case LZ4:
-                    in = new ClickHouseLZ4InputStream(in);
+                    chInput = new ClickHouseLZ4InputStream(in, afterClose);
                     break;
                 default:
                     throw new UnsupportedOperationException("Unsupported compression algorithm: " + algorithm);
             }
+        } else {
+            chInput = ClickHouseInputStream.of(in, config.getMaxBufferSize(), afterClose);
         }
 
-        return in;
+        return chInput;
     }
 
     /**

clickhouse-http-client/src/main/java/com/clickhouse/client/http/ClickHouseHttpResponse.java

@@ -1,6 +1,5 @@
 package com.clickhouse.client.http;
 
-import java.io.InputStream;
 import java.io.Serializable;
 import java.util.HashMap;
 import java.util.Map;
@@ -39,16 +38,6 @@ private static long getLongValue(Map<String, String> map, String key) {
 
     protected final ClickHouseResponseSummary summary;
 
-    protected void closeConnection() {
-        if (!connection.isReusable()) {
-            try {
-                connection.close();
-            } catch (Exception e) {
-                // ignore
-            }
-        }
-    }
-
     protected ClickHouseConfig getConfig(ClickHouseRequest<?> request) {
         ClickHouseConfig config = request.getConfig();
         if (format != null && format != config.getFormat()) {
@@ -61,14 +50,14 @@ protected ClickHouseConfig getConfig(ClickHouseRequest<?> request) {
         return config;
     }
 
-    public ClickHouseHttpResponse(ClickHouseHttpConnection connection, InputStream input, String serverDisplayName,
-            String queryId, String summary, ClickHouseFormat format, TimeZone timeZone) {
+    public ClickHouseHttpResponse(ClickHouseHttpConnection connection, ClickHouseInputStream input,
+            String serverDisplayName, String queryId, String summary, ClickHouseFormat format, TimeZone timeZone) {
         if (connection == null || input == null) {
             throw new IllegalArgumentException("Non-null connection and input stream are required");
         }
 
         this.connection = connection;
-        this.input = ClickHouseInputStream.of(input, connection.config.getMaxBufferSize(), this::closeConnection);
+        this.input = input;
 
         this.serverDisplayName = !ClickHouseChecker.isNullOrEmpty(serverDisplayName) ? serverDisplayName
                 : connection.server.getHost();

clickhouse-http-client/src/main/java/com/clickhouse/client/http/HttpUrlConnectionImpl.java

@@ -117,8 +117,8 @@ private void checkResponse(HttpURLConnection conn) throws IOException {
             // TODO get exception from response header, for example:
             // X-ClickHouse-Exception-Code: 47
             StringBuilder builder = new StringBuilder();
-            try (Reader reader = new BufferedReader(
-                    new InputStreamReader(getResponseInputStream(conn.getErrorStream()), StandardCharsets.UTF_8))) {
+            try (Reader reader = new InputStreamReader(getResponseInputStream(conn.getErrorStream()),
+                    StandardCharsets.UTF_8)) {
                 int c = 0;
                 while ((c = reader.read()) != -1) {
                     builder.append((char) c);

clickhouse-http-client/src/main/java11/com/clickhouse/client/http/HttpClientConnectionImpl.java

@@ -76,8 +76,7 @@ private HttpResponse<InputStream> checkResponse(HttpResponse<InputStream> r) thr
             // TODO get exception from response header, for example:
             // X-ClickHouse-Exception-Code: 47
             StringBuilder builder = new StringBuilder();
-            try (Reader reader = new BufferedReader(
-                    new InputStreamReader(getResponseInputStream(r.body()), StandardCharsets.UTF_8))) {
+            try (Reader reader = new InputStreamReader(getResponseInputStream(r.body()), StandardCharsets.UTF_8)) {
                 int c = 0;
                 while ((c = reader.read()) != -1) {
                     builder.append((char) c);
@@ -278,5 +277,6 @@ public boolean ping(int timeout) {
 
     @Override
     public void close() {
+        // nothing
     }
 }