Unified write data API

include/aws/http/private/h1_encoder.h

@@ -8,6 +8,8 @@
 #include <aws/http/private/http_impl.h>
 #include <aws/http/private/request_response_impl.h>
 
+struct aws_h1_data_write;
+
 struct aws_h1_chunk {
     struct aws_allocator *allocator;
     struct aws_input_stream *data;
@@ -44,6 +46,13 @@ struct aws_h1_encoder_message {
     /* Pointer to chunked_trailer, used for chunked_trailer. */
     struct aws_h1_trailer *trailer;
 
+    /* Pointer to list of `struct aws_h1_data_write`, used for manual data writes with Content-Length.
+     * List is owned by aws_h1_stream. */
+    struct aws_linked_list *pending_data_write_list;
+
+    /* Current data write being processed (for manual data writes with Content-Length) */
+    struct aws_h1_data_write *current_data_write;
+
     /* If non-zero, length of unchunked body to send */
     uint64_t content_length;
     bool has_connection_close_header;
@@ -64,6 +73,9 @@ enum aws_h1_encoder_state {
     AWS_H1_ENCODER_STATE_CHUNK_BODY,
     AWS_H1_ENCODER_STATE_CHUNK_END,
     AWS_H1_ENCODER_STATE_CHUNK_TRAILER,
+    /* The _DATA_WRITE_ states support the write_data() API (manual data writes with Content-Length) */
+    AWS_H1_ENCODER_STATE_DATA_WRITE_NEXT,
+    AWS_H1_ENCODER_STATE_DATA_WRITE_BODY,
     AWS_H1_ENCODER_STATE_DONE,
 };
 
@@ -104,7 +116,9 @@ int aws_h1_encoder_message_init_from_request(
     struct aws_h1_encoder_message *message,
     struct aws_allocator *allocator,
     const struct aws_http_message *request,
-    struct aws_linked_list *pending_chunk_list);
+    struct aws_linked_list *pending_chunk_list,
+    struct aws_linked_list *pending_data_write_list,
+    bool use_manual_data_writes);
 
 int aws_h1_encoder_message_init_from_response(
     struct aws_h1_encoder_message *message,
@@ -138,6 +152,10 @@ bool aws_h1_encoder_is_message_in_progress(const struct aws_h1_encoder *encoder)
 AWS_HTTP_API
 bool aws_h1_encoder_is_waiting_for_chunks(const struct aws_h1_encoder *encoder);
 
+/* Return true if the encoder is stuck waiting for more data writes to be added to the current message */
+AWS_HTTP_API
+bool aws_h1_encoder_is_waiting_for_data_writes(const struct aws_h1_encoder *encoder);
+
 AWS_EXTERN_C_END
 
 #endif /* AWS_HTTP_H1_ENCODER_H */

include/aws/http/private/h1_stream.h

@@ -22,6 +22,15 @@ enum aws_h1_stream_api_state {
     AWS_H1_STREAM_API_STATE_COMPLETE,
 };
 
+struct aws_h1_data_write {
+    struct aws_allocator *allocator;
+    struct aws_input_stream *data;
+    aws_http_stream_write_complete_fn *on_complete;
+    void *user_data;
+    struct aws_linked_list_node node;
+    bool is_end_stream;
+};
+
 struct aws_h1_stream {
     struct aws_http_stream base;
 
@@ -66,9 +75,21 @@ struct aws_h1_stream {
          * Only body data (not headers, etc) counts against the stream's flow-control window. */
         uint64_t stream_window;
 
+        /* List of `struct aws_h1_data_write` which have been moved from synced_data for processing */
+        struct aws_linked_list pending_data_write_list;
+
+        /* Whether the stream is using manual data writes instead of input_stream */
+        bool using_manual_data_writes : 1;
+
+        /* Whether the final data write (with is_end_stream=true) has been received */
+        bool has_final_data_write : 1;
+
         /* Whether a "request handler" stream has a response to send.
          * Has mirror variable in synced_data */
         bool has_outgoing_response : 1;
+
+        /* Total bytes written so far for Content-Length validation */
+        uint64_t incremental_content_written;
     } thread_data;
 
     /* Any thread may touch this data, but the connection's lock must be held.
@@ -83,6 +104,10 @@ struct aws_h1_stream {
          * but haven't yet moved to thread_data.encoder_message.pending_chunk_list where the encoder will find them. */
         struct aws_linked_list pending_chunk_list;
 
+        /* List of `struct aws_h1_data_write` which have been submitted by user,
+         * but haven't yet moved to thread_data.pending_data_write_list where the encoder will find them. */
+        struct aws_linked_list pending_data_write_list;
+
         /* trailing headers which have been submitted by user,
          * but haven't yet moved to thread_data.encoder_message where the encoder will find them. */
         struct aws_h1_trailer *pending_trailer;
@@ -102,9 +127,15 @@ struct aws_h1_stream {
         /* Whether the outgoing message is using chunked encoding */
         bool using_chunked_encoding : 1;
 
+        /* Whether the stream is using manual data writes instead of input_stream */
+        bool using_manual_data_writes : 1;
+
         /* Whether the final 0 length chunk has already been sent */
         bool has_final_chunk : 1;
 
+        /* Whether the final data write (with is_end_stream=true) has been received */
+        bool has_final_data_write : 1;
+
         /* Whether the chunked trailer has already been sent */
         bool has_added_trailer : 1;
     } synced_data;
@@ -123,4 +154,15 @@ void aws_h1_stream_cancel(struct aws_http_stream *stream, int error_code);
 
 int aws_h1_stream_send_response(struct aws_h1_stream *stream, struct aws_http_message *response);
 
+struct aws_h1_data_write *aws_h1_data_write_new(
+    struct aws_allocator *allocator,
+    const struct aws_http_stream_write_data_options *options);
+
+void aws_h1_data_write_destroy(struct aws_h1_data_write *data_write);
+
+void aws_h1_data_write_complete_and_destroy(
+    struct aws_h1_data_write *data_write,
+    struct aws_http_stream *http_stream,
+    int error_code);
+
 #endif /* AWS_HTTP_H1_STREAM_H */

include/aws/http/private/request_response_impl.h

@@ -25,9 +25,7 @@ struct aws_http_stream_vtable {
     int (*http2_reset_stream)(struct aws_http_stream *http2_stream, uint32_t http2_error);
     int (*http2_get_received_error_code)(struct aws_http_stream *http2_stream, uint32_t *http2_error);
     int (*http2_get_sent_error_code)(struct aws_http_stream *http2_stream, uint32_t *http2_error);
-    int (*http2_write_data)(
-        struct aws_http_stream *http2_stream,
-        const struct aws_http2_stream_write_data_options *options);
+    int (*write_data)(struct aws_http_stream *stream, const struct aws_http_stream_write_data_options *options);
 };
 
 /**

include/aws/http/request_response.h

@@ -300,6 +300,14 @@ struct aws_http_make_request_options {
     /* Callback for when the request/response stream is completely destroyed. */
     aws_http_on_stream_destroy_fn *on_destroy;
 
+    /**
+     * When true, request body data will be provided over time via `aws_http_stream_write_data()`.
+     * The stream will only be polled for writing when data has been supplied.
+     * Works with both HTTP/1.1 (Content-Length) and HTTP/2.
+     * When false (default), the entire request body is read from the input stream immediately.
+     */
+    bool use_manual_data_writes;
+
     /**
      * When using HTTP/2, request body data will be provided over time. The stream will only be polled for writing
      * when data has been supplied via `aws_http2_stream_write_data`
@@ -471,33 +479,28 @@ struct aws_http1_chunk_options {
 typedef aws_http_stream_write_complete_fn aws_http2_stream_write_data_complete_fn;
 
 /**
- * Encoding options for manual H2 data frame writes
+ * Common fields for write_data options structures.
+ * This macro allows protocol-specific options to share the same base fields.
  */
-struct aws_http2_stream_write_data_options {
-    /**
-     * The data to be sent.
-     * Optional.
-     * If not set, input stream with length 0 will be used.
-     */
-    struct aws_input_stream *data;
-
-    /**
-     * Set true when it's the last chunk to be sent.
-     * After a write with end_stream, no more data write will be accepted.
-     */
-    bool end_stream;
+#define AWS_HTTP_STREAM_WRITE_DATA_OPTIONS_FIELDS                                                                      \
+    struct aws_input_stream *data;                                                                                     \
+    bool end_stream;                                                                                                   \
+    aws_http_stream_write_complete_fn *on_complete;                                                                    \
+    void *user_data;
 
-    /**
-     * Invoked when the data stream is no longer in use, whether or not it was successfully sent.
-     * Optional.
-     * See `aws_http2_stream_write_data_complete_fn`.
-     */
-    aws_http2_stream_write_data_complete_fn *on_complete;
+/**
+ * Unified options for writing data to an HTTP stream.
+ * Works with both HTTP/1.1 (with Content-Length) and HTTP/2.
+ */
+struct aws_http_stream_write_data_options {
+    AWS_HTTP_STREAM_WRITE_DATA_OPTIONS_FIELDS
+};
 
-    /**
-     * User provided data passed to the on_complete callback on its invocation.
-     */
-    void *user_data;
+/**
+ * Encoding options for manual H2 data frame writes
+ */
+struct aws_http2_stream_write_data_options {
+    AWS_HTTP_STREAM_WRITE_DATA_OPTIONS_FIELDS
 };
 
 #define AWS_HTTP_REQUEST_HANDLER_OPTIONS_INIT                                                                          \
@@ -902,6 +905,27 @@ AWS_HTTP_API int aws_http1_stream_write_chunk(
     struct aws_http_stream *http1_stream,
     const struct aws_http1_chunk_options *options);
 
+/**
+ * Write data to an HTTP stream in a protocol-agnostic way.
+ * Works with both HTTP/1.1 (with Content-Length) and HTTP/2.
+ *
+ * For HTTP/1.1: The request must have a Content-Length header and must NOT have a body stream set.
+ *               The stream must have specified `use_manual_data_writes` during request creation.
+ * For HTTP/2: The stream must have specified `http2_use_manual_data_writes` during request creation.
+ *
+ * For client streams, activate() must be called before any data is written.
+ * For server streams, the response must be submitted before any data is written.
+ * A write with end_stream set to true will complete the stream and prevent any further writes.
+ *
+ * @return AWS_OP_SUCCESS if the write was queued
+ *         AWS_OP_ERROR indicating the attempt raised an error code.
+ *              AWS_ERROR_INVALID_STATE will be raised for invalid usage.
+ *              AWS_ERROR_HTTP_STREAM_HAS_COMPLETED will be raised if the stream ended for reasons behind the scenes.
+ */
+AWS_HTTP_API int aws_http_stream_write_data(
+    struct aws_http_stream *stream,
+    const struct aws_http_stream_write_data_options *options);
+
 /**
  * The stream must have specified `http2_use_manual_data_writes` during request creation.
  * For client streams, activate() must be called before any frames are submitted.

source/h1_connection.c

@@ -1072,14 +1072,17 @@ static void s_write_outgoing_stream(struct aws_h1_connection *connection, bool f
      * The outgoing stream task will be kicked off again when user adds more data (new stream, new chunk, etc) */
     struct aws_h1_stream *outgoing_stream = s_update_outgoing_stream_ptr(connection);
     bool waiting_for_chunks = aws_h1_encoder_is_waiting_for_chunks(&connection->thread_data.encoder);
-    if (!outgoing_stream || waiting_for_chunks) {
+    bool waiting_for_data_writes = aws_h1_encoder_is_waiting_for_data_writes(&connection->thread_data.encoder);
+    if (!outgoing_stream || waiting_for_chunks || waiting_for_data_writes) {
         if (!first_try) {
             AWS_LOGF_TRACE(
                 AWS_LS_HTTP_CONNECTION,
-                "id=%p: Outgoing stream task stopped. outgoing_stream=%p waiting_for_chunks:%d",
+                "id=%p: Outgoing stream task stopped. outgoing_stream=%p waiting_for_chunks:%d "
+                "waiting_for_data_writes:%d",
                 (void *)&connection->base,
                 outgoing_stream ? (void *)&outgoing_stream->base : NULL,
-                waiting_for_chunks);
+                waiting_for_chunks,
+                waiting_for_data_writes);
         }
         connection->thread_data.is_outgoing_stream_task_active = false;
         return;