Bind out the write data API (#981)

Author: azkrishpy

src/main/java/software/amazon/awssdk/crt/http/Http2ClientConnection.java

@@ -187,12 +187,49 @@ public void updateConnectionWindow(long incrementSize) {
     @Override
     public Http2Stream makeRequest(HttpRequestBase request, HttpStreamBaseResponseHandler streamHandler)
             throws CrtRuntimeException {
+        return makeRequest(request, streamHandler, false);
+    }
+
+    /**
+     * Schedules an HttpRequest on the Native EventLoop for this
+     * HttpClientConnection. The HTTP/1.1 request will be transformed to HTTP/2
+     * request under the hood.
+     *
+     * @param request       The Request to make to the Server.
+     * @param streamHandler The Stream Handler to be called from the Native
+     *                      EventLoop
+     * @param useManualDataWrites When {@code true}, request body data is provided via
+     *        {@link HttpStreamBase#writeData} instead of from the request's
+     *        {@link HttpRequestBodyStream}.
+     *
+     *        <p>By design, CRT supports setting both a body stream and enabling manual
+     *        writes for HTTP/2, but this is not recommended. Body streams are intended
+     *        for requests whose payload is available in full at the time of sending. If
+     *        the stream does not signal end-of-stream promptly, the event loop will
+     *        busy-wait (hot-loop) for more data, wasting CPU time. Manual writes avoid
+     *        this by letting the caller control when data is sent; the event loop only
+     *        processes the request when {@link HttpStreamBase#writeData} is called and is
+     *        free to service other requests in the meantime.
+     *
+     *        <p>When both a body stream and manual writes are enabled, the body stream is
+     *        sent as the first DATA frame and the connection then waits asynchronously for
+     *        subsequent {@code writeData()} calls. However, if the body stream has not
+     *        signalled end-of-stream, the event loop will keep getting scheduled for
+     *        requesting more data until it completes.
+     * @throws CrtRuntimeException if stream creation fails
+     * @return The Http2Stream that represents this Request/Response Pair. It can be
+     *         closed at any time during the request/response, but must be closed by
+     *         the user thread making this request when it's done.
+     */
+    public Http2Stream makeRequest(HttpRequestBase request, HttpStreamBaseResponseHandler streamHandler,
+            boolean useManualDataWrites) throws CrtRuntimeException {
         if (isNull()) {
             throw new IllegalStateException("Http2ClientConnection has been closed, can't make requests on it.");
         }
 
         Http2Stream stream = http2ClientConnectionMakeRequest(getNativeHandle(), request.marshalForJni(),
-                request.getBodyStream(), new HttpStreamResponseHandlerNativeAdapter(streamHandler));
+                request.getBodyStream(), new HttpStreamResponseHandlerNativeAdapter(streamHandler),
+                useManualDataWrites);
         return stream;
     }
 
@@ -204,7 +241,8 @@ public Http2Stream makeRequest(HttpRequestBase request, HttpStreamBaseResponseHa
      ******************************************************************************/
 
     private static native Http2Stream http2ClientConnectionMakeRequest(long connectionBinding, byte[] marshalledRequest,
-            HttpRequestBodyStream bodyStream, HttpStreamResponseHandlerNativeAdapter responseHandler)
+            HttpRequestBodyStream bodyStream, HttpStreamResponseHandlerNativeAdapter responseHandler,
+            boolean useManualDataWrites)
             throws CrtRuntimeException;
 
     private static native void http2ClientConnectionUpdateSettings(long connectionBinding,

src/main/java/software/amazon/awssdk/crt/http/HttpClientConnection.java

@@ -41,22 +41,58 @@ protected HttpClientConnection(long connectionBinding) {
      */
     public HttpStream makeRequest(HttpRequest request, HttpStreamResponseHandler streamHandler)
             throws CrtRuntimeException {
+        return makeRequest(request, streamHandler, false);
+    }
+
+    /**
+     * Schedules an HttpRequest on the Native EventLoop for this HttpClientConnection specific to HTTP/1.1 connection.
+     *
+     * @param request The Request to make to the Server.
+     * @param streamHandler The Stream Handler to be called from the Native EventLoop
+     * @param useManualDataWrites When {@code true}, request body data is provided via
+     *        {@link HttpStreamBase#writeData} instead of from the request's
+     *        {@link HttpRequestBodyStream}.
+     *
+     *        <p>By design, CRT does not support setting both a body stream and enabling
+     *        manual writes for HTTP/1.1. Body streams are intended for requests whose
+     *        payload is available in full at the time of sending. If the stream does not
+     *        signal end-of-stream promptly, the event loop will busy-wait (hot-loop) for
+     *        more data, wasting CPU time. Manual writes avoid this by letting the caller
+     *        control when data is sent; the event loop only processes the request when
+     *        {@link HttpStreamBase#writeData} is called and is free to service other
+     *        requests in the meantime.
+     *
+     *        <p>If the request was created with an {@link HttpRequestBodyStream} and this
+     *        parameter is {@code true}, an {@link IllegalStateException} is thrown
+     *        immediately.
+     * @throws CrtRuntimeException if stream creation fails
+     * @return The HttpStream that represents this Request/Response Pair. It can be closed at any time during the
+     *          request/response, but must be closed by the user thread making this request when it's done.
+     */
+    public HttpStream makeRequest(HttpRequest request, HttpStreamResponseHandler streamHandler,
+            boolean useManualDataWrites) throws CrtRuntimeException, IllegalStateException {
         if (isNull()) {
             throw new IllegalStateException("HttpClientConnection has been closed, can't make requests on it.");
         }
         if (getVersion() == HttpVersion.HTTP_2) {
             throw new IllegalArgumentException("HTTP/1 only method called on an HTTP/2 connection.");
         }
+        if (useManualDataWrites && request.getBodyStream() != null) {
+            throw new IllegalStateException(
+                "Cannot use manual data writes with a body stream on an HTTP/1.1 request. "
+                + "Either remove the body stream or set useManualDataWrites to false.");
+        }
         HttpStreamBase stream = httpClientConnectionMakeRequest(getNativeHandle(),
                 request.marshalForJni(),
                 request.getBodyStream(),
-                new HttpStreamResponseHandlerNativeAdapter(streamHandler));
+                new HttpStreamResponseHandlerNativeAdapter(streamHandler),
+                useManualDataWrites);
 
         return (HttpStream)stream;
     }
 
     /**
-     * Schedules an HttpRequestBase on the Native EventLoop for this HttpClientConnection applies to both HTTP/2 and HTTP/1.1 connection.
+     * Schedules an HttpRequestBase on the Native EventLoop for this HttpClientConnection. Applies to both HTTP/2 and HTTP/1.1 connections.
      *
      * @param request The Request to make to the Server.
      * @param streamHandler The Stream Handler to be called from the Native EventLoop
@@ -65,13 +101,30 @@ public HttpStream makeRequest(HttpRequest request, HttpStreamResponseHandler str
      *          request/response, but must be closed by the user thread making this request when it's done.
      */
     public HttpStreamBase makeRequest(HttpRequestBase request, HttpStreamBaseResponseHandler streamHandler) throws CrtRuntimeException {
+        return makeRequest(request, streamHandler, false);
+    }
+
+    /**
+     * Schedules an HttpRequestBase on the Native EventLoop for this HttpClientConnection. Applies to both HTTP/2 and HTTP/1.1 connections.
+     *
+     * @param request The Request to make to the Server.
+     * @param streamHandler The Stream Handler to be called from the Native EventLoop
+     * @param useManualDataWrites When true, request body data will be provided via
+     *        {@link HttpStreamBase#writeData} instead of from the request's body stream.
+     * @throws CrtRuntimeException if stream creation fails
+     * @return The HttpStream that represents this Request/Response Pair. It can be closed at any time during the
+     *          request/response, but must be closed by the user thread making this request when it's done.
+     */
+    public HttpStreamBase makeRequest(HttpRequestBase request, HttpStreamBaseResponseHandler streamHandler,
+            boolean useManualDataWrites) throws CrtRuntimeException {
         if (isNull()) {
             throw new IllegalStateException("HttpClientConnection has been closed, can't make requests on it.");
         }
         HttpStreamBase stream = httpClientConnectionMakeRequest(getNativeHandle(),
                 request.marshalForJni(),
                 request.getBodyStream(),
-                new HttpStreamResponseHandlerNativeAdapter(streamHandler));
+                new HttpStreamResponseHandlerNativeAdapter(streamHandler),
+                useManualDataWrites);
 
         return stream;
     }
@@ -172,7 +225,8 @@ public static boolean isErrorRetryable(HttpException exception) {
     private static native HttpStreamBase httpClientConnectionMakeRequest(long connectionBinding,
                                                                      byte[] marshalledRequest,
                                                                      HttpRequestBodyStream bodyStream,
-                                                                     HttpStreamResponseHandlerNativeAdapter responseHandler) throws CrtRuntimeException;
+                                                                     HttpStreamResponseHandlerNativeAdapter responseHandler,
+                                                                     boolean useManualDataWrites) throws CrtRuntimeException;
 
     private static native void httpClientConnectionShutdown(long connectionBinding) throws CrtRuntimeException;
     private static native boolean httpClientConnectionIsOpen(long connectionBinding) throws CrtRuntimeException;

src/main/java/software/amazon/awssdk/crt/http/HttpStream.java

@@ -40,7 +40,10 @@ public interface HttpStreamWriteChunkCompletionCallback {
      *                                request stream.
      * @param chunkCompletionCallback Invoked upon the data being flushed to the
      *                                wire or an error occurring.
+     * @deprecated Use {@link HttpStreamBase#writeData(byte[], boolean, HttpStreamWriteDataCompletionCallback)} instead.
+     *             writeData() works for both HTTP/1.1 and HTTP/2, whereas writeChunk() is HTTP/1.1 only.
      */
+    @Deprecated
     public void writeChunk(final byte[] chunkData, boolean isFinalChunk,
             final HttpStreamWriteChunkCompletionCallback chunkCompletionCallback) {
         if (isNull()) {
@@ -71,7 +74,10 @@ public void writeChunk(final byte[] chunkData, boolean isFinalChunk,
      * @param isFinalChunk if set to true, this will terminate the request stream.
      * @return completable future which will complete upon the data being flushed to
      *         the wire or an error occurring.
+     * @deprecated Use {@link HttpStreamBase#writeData(byte[], boolean)} instead.
+     *             writeData() works for both HTTP/1.1 and HTTP/2, whereas writeChunk() is HTTP/1.1 only.
      */
+    @Deprecated
     public CompletableFuture<Void> writeChunk(final byte[] chunkData, boolean isFinalChunk) {
         CompletableFuture<Void> completionFuture = new CompletableFuture<>();
 

src/main/java/software/amazon/awssdk/crt/http/HttpStreamBase.java

@@ -5,7 +5,11 @@
 
 package software.amazon.awssdk.crt.http;
 
+import software.amazon.awssdk.crt.CRT;
 import software.amazon.awssdk.crt.CrtResource;
+import software.amazon.awssdk.crt.CrtRuntimeException;
+
+import java.util.concurrent.CompletableFuture;
 
 /**
  * An base class represents a single Http Request/Response for both HTTP/1.1 and
@@ -112,6 +116,63 @@ public void cancel() {
         }
     }
 
+    /**
+     * Completion interface for writing data to an http stream.
+     */
+    public interface HttpStreamWriteDataCompletionCallback {
+        void onWriteDataCompleted(int errorCode);
+    }
+
+    /**
+     * Write data to an HTTP stream. Works for both HTTP/1.1 and HTTP/2.
+     * The stream must have been created with {@code useManualDataWrites = true}.
+     * You must call activate() before using this function.
+     *
+     * @param data       data to send, or null to write zero bytes. Pass null with
+     *                   endStream=true to signal end-of-body without sending additional data.
+     * @param endStream  if true, this is the last data to be sent on this stream.
+     * @param completionCallback invoked when the data has been flushed or an error occurs.
+     */
+    public void writeData(final byte[] data, boolean endStream,
+            final HttpStreamWriteDataCompletionCallback completionCallback) {
+        if (isNull()) {
+            throw new IllegalStateException("HttpStream has been closed.");
+        }
+        if (completionCallback == null) {
+            throw new IllegalArgumentException("You must supply a completionCallback");
+        }
+
+        int error = httpStreamBaseWriteData(getNativeHandle(), data, endStream, completionCallback);
+        if (error != 0) {
+            int lastError = CRT.awsLastError();
+            throw new CrtRuntimeException(lastError);
+        }
+    }
+
+    /**
+     * Write data to an HTTP stream. Works for both HTTP/1.1 and HTTP/2.
+     * The stream must have been created with {@code useManualDataWrites = true}.
+     * You must call activate() before using this function.
+     *
+     * @param data      data to send, or null to write zero bytes. Pass null with
+     *                  endStream=true to signal end-of-body without sending additional data.
+     * @param endStream if true, this is the last data to be sent on this stream.
+     * @return completable future which completes when data is flushed or an error occurs.
+     */
+    public CompletableFuture<Void> writeData(final byte[] data, boolean endStream) {
+        CompletableFuture<Void> completionFuture = new CompletableFuture<>();
+
+        writeData(data, endStream, (errorCode) -> {
+            if (errorCode == 0) {
+                completionFuture.complete(null);
+            } else {
+                completionFuture.completeExceptionally(new CrtRuntimeException(errorCode));
+            }
+        });
+
+        return completionFuture;
+    }
+
     /*******************************************************************************
      * Native methods
      ******************************************************************************/
@@ -125,4 +186,7 @@ public void cancel() {
     private static native int httpStreamBaseGetResponseStatusCode(long http_stream);
 
     private static native void httpStreamBaseCancelDefaultError(long http_stream);
+
+    private static native int httpStreamBaseWriteData(long http_stream, byte[] data, boolean endStream,
+            HttpStreamWriteDataCompletionCallback completionCallback);
 }

src/main/resources/META-INF/native-image/software.amazon.awssdk/crt/aws-crt/jni-config.json

@@ -761,6 +761,17 @@
       }
     ]
   },
+  {
+    "name": "software.amazon.awssdk.crt.http.HttpStreamBase$HttpStreamWriteDataCompletionCallback",
+    "methods": [
+      {
+        "name": "onWriteDataCompleted",
+        "parameterTypes": [
+          "int"
+        ]
+      }
+    ]
+  },
   {
     "name": "software.amazon.awssdk.crt.http.HttpStreamMetrics",
     "methods": [