Revert changes made on streaming

ClickHouse · Dec 8, 2021 · e7b0e6a · e7b0e6a
1 parent e9a3bc7
commit e7b0e6a
Show file tree

Hide file tree

Showing 4 changed files with 12 additions and 20 deletions.
diff --git a/clickhouse-client/src/main/java/com/clickhouse/client/ClickHouseInputStream.java b/clickhouse-client/src/main/java/com/clickhouse/client/ClickHouseInputStream.java
@@ -40,7 +40,7 @@ private void ensureOpen() throws IOException {
                 throw new IOException("Stream has been closed");
             }
 
-            if (buffer == null || (buffer != EMPTY && buffer.limit() > 0 && !buffer.hasRemaining())) {
+            if (buffer == null || (buffer != EMPTY && !buffer.hasRemaining())) {
                 updateBuffer();
             }
         }
@@ -65,17 +65,9 @@ private int updateBuffer() throws IOException {
 
         @Override
         public int available() throws IOException {
-            if (closed || buffer == EMPTY) {
-                return 0;
-            }
+            ensureOpen();
 
-            int available = 0;
-            if (buffer == null || (buffer.limit() > 0 && !buffer.hasRemaining())) {
-                available = updateBuffer();
-            } else {
-                available = buffer.remaining();
-            }
-            return available;
+            return buffer.remaining();
         }
 
         @Override
@@ -95,7 +87,7 @@ public void close() throws IOException {
         public byte readByte() throws IOException {
             ensureOpen();
 
-            if (buffer == EMPTY || buffer.limit() == 0) {
+            if (buffer == EMPTY) {
                 close();
                 throw new EOFException();
             }
@@ -107,7 +99,7 @@ public byte readByte() throws IOException {
         public int read() throws IOException {
             ensureOpen();
 
-            if (buffer == EMPTY || buffer.limit() == 0) {
+            if (buffer == EMPTY) {
                 return -1;
             }
 
@@ -120,7 +112,7 @@ public int read(byte[] b, int off, int len) throws IOException {
 
             int counter = 0;
             while (len > 0) {
-                if (buffer == EMPTY || buffer.limit() == 0) {
+                if (buffer == EMPTY) {
                     return counter > 0 ? counter : -1;
                 }
 

diff --git a/clickhouse-client/src/main/java/com/clickhouse/client/data/ClickHouseLZ4InputStream.java b/clickhouse-client/src/main/java/com/clickhouse/client/data/ClickHouseLZ4InputStream.java
@@ -28,10 +28,10 @@ public class ClickHouseLZ4InputStream extends ClickHouseInputStream {
     private boolean closed;
 
     private boolean checkNext() throws IOException {
-        if (currentBlock == null) {
+        if (currentBlock == null || !currentBlock.hasRemaining()) {
             currentBlock = readNextBlock();
         }
-        return currentBlock != null && currentBlock.hasRemaining();
+        return currentBlock != null;
     }
 
     // every block is:

diff --git a/...use-http-client/src/main/java11/com/clickhouse/client/http/ClickHouseResponseHandler.java b/...use-http-client/src/main/java11/com/clickhouse/client/http/ClickHouseResponseHandler.java
@@ -9,7 +9,6 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.LinkedBlockingDeque;
-import java.util.concurrent.TimeUnit;
 import java.util.concurrent.Flow.Subscription;
 import java.util.concurrent.atomic.AtomicBoolean;
 

diff --git a/clickhouse-http-client/src/main/java11/com/clickhouse/client/http/DefaultHttpConnection.java b/clickhouse-http-client/src/main/java11/com/clickhouse/client/http/DefaultHttpConnection.java
@@ -183,14 +183,15 @@ private ClickHouseHttpResponse postString(HttpRequest.Builder reqBuilder, String
         reqBuilder.POST(HttpRequest.BodyPublishers.ofString(sql));
         HttpResponse<InputStream> r;
         try {
-            // r = httpClient.send(reqBuilder.build(), responseInfo -> new
-            // ExtendedResponseInputStream());
-            r = httpClient.send(reqBuilder.build(),
+            CompletableFuture<HttpResponse<InputStream>> f = httpClient.sendAsync(reqBuilder.build(),
                     responseInfo -> new ClickHouseResponseHandler(config.getMaxBufferSize(),
                             config.getSocketTimeout()));
+            r = f.get();
         } catch (InterruptedException e) {
             Thread.currentThread().interrupt();
             throw new IOException("Thread was interrupted when posting request or receiving response", e);
+        } catch (ExecutionException e) {
+            throw new IOException("Failed to post query", e);
         }
         return buildResponse(r);
     }