Closes #345, Introduces KustoRequest wrapper object and async methods

data/src/main/java/com/microsoft/azure/kusto/data/BaseClient.java

@@ -10,10 +10,12 @@
 import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import reactor.core.publisher.Mono;
 
 import java.io.IOException;
 import java.io.InputStream;
 import java.lang.invoke.MethodHandles;
+import java.util.Objects;
 import java.util.Optional;
 
 public abstract class BaseClient implements Client, StreamingClient {
@@ -30,33 +32,50 @@ public BaseClient(HttpClient httpClient) {
     }
 
     protected String post(HttpRequest request) throws DataServiceException {
-
-        // Todo: Add async version of this method
-
         // Execute and get the response
         try (HttpResponse response = httpClient.sendSync(request, Context.NONE)) {
-            String responseBody = Utils.isGzipResponse(response) ? Utils.gzipedInputToString(response.getBodyAsBinaryData().toStream())
-                    : response.getBodyAsBinaryData().toString();
-
-            if (responseBody != null) {
-                switch (response.getStatusCode()) {
-                    case HttpStatus.OK:
-                        return responseBody;
-                    case HttpStatus.TOO_MANY_REQS:
-                        throw new ThrottleException(request.getUrl().toString());
-                    default:
-                        throw createExceptionFromResponse(request.getUrl().toString(), response, null, responseBody);
-                }
-            }
+            return processResponseBody(response);
         }
+    }
+
+    protected Mono<String> postAsync(HttpRequest request) {
+        // Execute and get the response
+        return httpClient.send(request)
+                .onErrorContinue((err, src) -> LOGGER.error("Error coming from src {}", src, err))
+                .flatMap(this::processResponseAsync)
+                .onErrorContinue((err, src) -> LOGGER.error("Error coming from src {}", src, err));
+    }
 
-        return null;
+    public Mono<String> processResponseAsync(HttpResponse response) {
+        try {
+            return Mono.just(Objects.requireNonNull(processResponseBody(response)));
+        } catch (Exception e) {
+            return Mono.error(new RuntimeException("Error processing response", e));
+        }
+    }
+
+    private String processResponseBody(HttpResponse response) throws DataServiceException {
+        String responseBody = Utils.isGzipResponse(response) ? Utils.gzipedInputToString(response.getBodyAsBinaryData().toStream())
+                : response.getBodyAsBinaryData().toString();
+
+        if (responseBody == null) {
+            return null;
+        }
+        switch (response.getStatusCode()) {
+            case HttpStatus.OK:
+                return responseBody;
+            case HttpStatus.TOO_MANY_REQS:
+                throw new ThrottleException(response.getRequest().getUrl().toString());
+            default:
+                throw createExceptionFromResponse(response.getRequest().getUrl().toString(), response, null, responseBody);
+        }
     }
 
     protected InputStream postToStreamingOutput(HttpRequest request) throws DataServiceException {
         return postToStreamingOutput(request, 0);
     }
 
+    // Todo: Implement async version of this method
     protected InputStream postToStreamingOutput(HttpRequest request, int redirectCount) throws DataServiceException {
 
         boolean returnInputStream = false;
@@ -65,7 +84,6 @@ protected InputStream postToStreamingOutput(HttpRequest request, int redirectCou
         HttpResponse httpResponse = null;
         try {
 
-            // Todo: Implement async version of this method
             httpResponse = httpClient.sendSync(request, Context.NONE);
 
             int responseStatusCode = httpResponse.getStatusCode();

data/src/main/java/com/microsoft/azure/kusto/data/Client.java

@@ -5,30 +5,36 @@
 
 import com.microsoft.azure.kusto.data.exceptions.DataClientException;
 import com.microsoft.azure.kusto.data.exceptions.DataServiceException;
+import reactor.core.publisher.Mono;
 
-import java.io.Closeable;
+public interface Client {
 
-public interface Client extends Closeable {
-    KustoOperationResult execute(String command) throws DataServiceException, DataClientException;
+    Mono<KustoOperationResult> executeQueryAsync(String database, String command, ClientRequestProperties properties);
 
-    KustoOperationResult execute(String database, String command) throws DataServiceException, DataClientException;
+    Mono<KustoOperationResult> executeMgmtAsync(String database, String command, ClientRequestProperties properties);
 
-    KustoOperationResult execute(String database, String command, ClientRequestProperties properties) throws DataServiceException, DataClientException;
+    Mono<String> executeToJsonAsync(String database, String command, ClientRequestProperties properties);
 
+    @Deprecated
     KustoOperationResult executeQuery(String command) throws DataServiceException, DataClientException;
 
+    @Deprecated
     KustoOperationResult executeQuery(String database, String command) throws DataServiceException, DataClientException;
 
     KustoOperationResult executeQuery(String database, String command, ClientRequestProperties properties) throws DataServiceException, DataClientException;
 
+    @Deprecated
     KustoOperationResult executeMgmt(String command) throws DataServiceException, DataClientException;
 
+    @Deprecated
     KustoOperationResult executeMgmt(String database, String command) throws DataServiceException, DataClientException;
 
     KustoOperationResult executeMgmt(String database, String command, ClientRequestProperties properties) throws DataServiceException, DataClientException;
 
+    @Deprecated
     String executeToJsonResult(String database) throws DataServiceException, DataClientException;
 
+    @Deprecated
     String executeToJsonResult(String database, String command) throws DataServiceException, DataClientException;
 
     String executeToJsonResult(String database, String command, ClientRequestProperties properties) throws DataServiceException, DataClientException;