allow write_chunk also to be routed via write_data

azkrishpy · azkrishpy · commit af32dd91995f · 2026-02-27T17:03:53.000-08:00
diff --git a/source/h1_encoder.c b/source/h1_encoder.c
@@ -152,15 +152,11 @@ static int s_scan_outgoing_headers(
 
 static int s_validate_manual_data_writes(const struct aws_h1_encoder_message *encoder_message, bool has_body_stream) {
 
-    /* Manual data writes require Content-Length header */
-    if (encoder_message->content_length == 0) {
-        AWS_LOGF_ERROR(AWS_LS_HTTP_STREAM, "id=static: Manual data writes require Content-Length header");
-        return aws_raise_error(AWS_ERROR_HTTP_INVALID_HEADER_FIELD);
-    }
-
-    /* Manual data writes cannot use chunked encoding */
-    if (encoder_message->has_chunked_encoding_header) {
-        AWS_LOGF_ERROR(AWS_LS_HTTP_STREAM, "id=static: Manual data writes cannot use Transfer-Encoding: chunked");
+    /* Manual data writes require either Content-Length or chunked encoding */
+    if (encoder_message->content_length == 0 && !encoder_message->has_chunked_encoding_header) {
+        AWS_LOGF_ERROR(
+            AWS_LS_HTTP_STREAM,
+            "id=static: Manual data writes require Content-Length header or Transfer-Encoding: chunked");
         return aws_raise_error(AWS_ERROR_HTTP_INVALID_HEADER_FIELD);
     }
 
@@ -277,7 +273,9 @@ int aws_h1_encoder_message_init_from_request(
 
     message->body = aws_input_stream_acquire(aws_http_message_get_body_stream(request));
     message->pending_chunk_list = pending_chunk_list;
-    message->pending_data_write_list = use_manual_data_writes ? pending_data_write_list : NULL;
+    /* Only set pending_data_write_list for Content-Length manual writes.
+     * Chunked manual writes go through pending_chunk_list instead (converted in s_stream_write_data). */
+    message->pending_data_write_list = NULL;
 
     struct aws_byte_cursor method;
     int err = aws_http_message_get_request_method(request, &method);
@@ -330,6 +328,11 @@ int aws_h1_encoder_message_init_from_request(
         if (err) {
             goto error;
         }
+        /* Content-Length manual writes use pending_data_write_list.
+         * Chunked manual writes go through pending_chunk_list instead. */
+        if (!message->has_chunked_encoding_header) {
+            message->pending_data_write_list = pending_data_write_list;
+        }
     }
 
     /* request-line: "{method} {uri} {version}\r\n" */
diff --git a/source/h1_stream.c b/source/h1_stream.c
@@ -383,20 +383,12 @@ static int s_stream_write_data(
         return aws_raise_error(AWS_ERROR_INVALID_ARGUMENT);
     }
 
-    struct aws_h1_data_write *data_write = aws_h1_data_write_new(stream_base->alloc, options);
-    if (!data_write) {
-        AWS_LOGF_ERROR(
-            AWS_LS_HTTP_STREAM,
-            "id=%p: Failed to create data write, error %d (%s).",
-            (void *)stream_base,
-            aws_last_error(),
-            aws_error_name(aws_last_error()));
-        return AWS_OP_ERR;
-    }
-
     int error_code = 0;
     bool should_schedule_task = false;
 
+    /* Check if this stream uses chunked encoding — if so, convert write_data into chunks */
+    bool is_chunked = false;
+
     { /* BEGIN CRITICAL SECTION */
         s_stream_lock_synced_data(stream);
 
@@ -425,14 +417,12 @@ static int s_stream_write_data(
             goto unlock;
         }
 
+        is_chunked = stream->synced_data.using_chunked_encoding;
+
         if (options->end_stream) {
             stream->synced_data.has_final_data_write = true;
         }
 
-        aws_linked_list_push_back(&stream->synced_data.pending_data_write_list, &data_write->node);
-        should_schedule_task = !stream->synced_data.is_cross_thread_work_task_scheduled;
-        stream->synced_data.is_cross_thread_work_task_scheduled = true;
-
     unlock:
         s_stream_unlock_synced_data(stream);
     } /* END CRITICAL SECTION */
@@ -444,10 +434,65 @@ static int s_stream_write_data(
             (void *)stream_base,
             error_code,
             aws_error_name(error_code));
-        aws_h1_data_write_destroy(data_write);
         return aws_raise_error(error_code);
     }
 
+    if (is_chunked) {
+        /* Convert write_data into chunk(s) for chunked encoding */
+        int64_t data_len = 0;
+        if (aws_input_stream_get_length(options->data, &data_len)) {
+            AWS_LOGF_ERROR(
+                AWS_LS_HTTP_STREAM,
+                "id=%p: Failed to get data stream length for chunked conversion",
+                (void *)stream_base);
+            return AWS_OP_ERR;
+        }
+
+        struct aws_http1_chunk_options chunk_opts = {
+            .chunk_data = options->data,
+            .chunk_data_size = (uint64_t)data_len,
+            .on_complete = options->on_complete,
+            .user_data = options->user_data,
+        };
+        if (aws_http1_stream_write_chunk(stream_base, &chunk_opts)) {
+            return AWS_OP_ERR;
+        }
+
+        /* If end_stream, also submit terminating zero-length chunk */
+        if (options->end_stream) {
+            struct aws_http1_chunk_options terminator = {
+                .chunk_data_size = 0,
+            };
+            if (aws_http1_stream_write_chunk(stream_base, &terminator)) {
+                return AWS_OP_ERR;
+            }
+        }
+
+        return AWS_OP_SUCCESS;
+    }
+
+    /* Content-Length path: create data write and push to pending list */
+    struct aws_h1_data_write *data_write = aws_h1_data_write_new(stream_base->alloc, options);
+    if (!data_write) {
+        AWS_LOGF_ERROR(
+            AWS_LS_HTTP_STREAM,
+            "id=%p: Failed to create data write, error %d (%s).",
+            (void *)stream_base,
+            aws_last_error(),
+            aws_error_name(aws_last_error()));
+        return AWS_OP_ERR;
+    }
+
+    { /* BEGIN CRITICAL SECTION */
+        s_stream_lock_synced_data(stream);
+
+        aws_linked_list_push_back(&stream->synced_data.pending_data_write_list, &data_write->node);
+        should_schedule_task = !stream->synced_data.is_cross_thread_work_task_scheduled;
+        stream->synced_data.is_cross_thread_work_task_scheduled = true;
+
+        s_stream_unlock_synced_data(stream);
+    } /* END CRITICAL SECTION */
+
     AWS_LOGF_TRACE(AWS_LS_HTTP_STREAM, "id=%p: Queued data write", (void *)stream_base);
 
     if (should_schedule_task) {
diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt
@@ -165,6 +165,8 @@ add_test_case(h1_client_write_data_after_final)
 add_test_case(h1_client_write_data_exceeds_content_length)
 add_test_case(h1_client_write_data_less_than_content_length)
 add_test_case(h1_client_unified_write_data_api)
+add_test_case(h1_client_write_data_chunked_single)
+add_test_case(h1_client_write_data_chunked_multiple)
 
 add_test_case(strutil_trim_http_whitespace)
 add_test_case(strutil_is_http_token)
diff --git a/tests/test_h1_client.c b/tests/test_h1_client.c
@@ -104,7 +104,7 @@ static int s_tester_init_ex(struct tester *tester, struct aws_allocator *alloc,
     tester->alloc = alloc;
 
     struct aws_logger_standard_options logger_options = {
-        .level = AWS_LOG_LEVEL_TRACE,
+        .level = AWS_LOG_LEVEL_DEBUG,
         .file = stderr,
     };
     ASSERT_SUCCESS(aws_logger_init_standard(&tester->logger, tester->alloc, &logger_options));
@@ -5607,3 +5607,106 @@ H1_CLIENT_TEST_CASE(h1_client_unified_write_data_api) {
     ASSERT_SUCCESS(s_tester_clean_up(&tester));
     return AWS_OP_SUCCESS;
 }
+
+/* Test write_data with chunked encoding - single write */
+H1_CLIENT_TEST_CASE(h1_client_write_data_chunked_single) {
+    (void)ctx;
+    struct tester tester;
+    ASSERT_SUCCESS(s_tester_init(&tester, allocator));
+
+    struct aws_http_message *request = s_new_default_chunked_put_request(allocator);
+
+    struct client_stream_tester stream_tester;
+    ASSERT_SUCCESS(client_stream_tester_init(
+        &stream_tester,
+        allocator,
+        &(struct client_stream_tester_options){
+            .request = request,
+            .connection = tester.connection,
+            .use_manual_data_writes = true,
+        }));
+
+    testing_channel_drain_queued_tasks(&tester.testing_channel);
+    aws_http_message_destroy(request);
+
+    struct aws_byte_cursor data = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("write more tests");
+    struct aws_input_stream *input_stream = aws_input_stream_new_from_cursor(allocator, &data);
+
+    struct aws_http_stream_write_data_options write_options = {
+        .data = input_stream,
+        .end_stream = true,
+    };
+
+    ASSERT_SUCCESS(aws_http_stream_write_data(stream_tester.stream, &write_options));
+    testing_channel_drain_queued_tasks(&tester.testing_channel);
+
+    const char *expected = "PUT /plan.txt HTTP/1.1\r\n"
+                           "Transfer-Encoding: chunked\r\n"
+                           "\r\n"
+                           "10\r\n"
+                           "write more tests"
+                           "\r\n"
+                           "0\r\n"
+                           "\r\n";
+    ASSERT_SUCCESS(testing_channel_check_written_messages_str(&tester.testing_channel, allocator, expected));
+
+    aws_input_stream_release(input_stream);
+    client_stream_tester_clean_up(&stream_tester);
+    ASSERT_SUCCESS(s_tester_clean_up(&tester));
+    return AWS_OP_SUCCESS;
+}
+
+/* Test write_data with chunked encoding - multiple writes */
+H1_CLIENT_TEST_CASE(h1_client_write_data_chunked_multiple) {
+    (void)ctx;
+    struct tester tester;
+    ASSERT_SUCCESS(s_tester_init(&tester, allocator));
+
+    struct aws_http_message *request = s_new_default_chunked_put_request(allocator);
+
+    struct client_stream_tester stream_tester;
+    ASSERT_SUCCESS(client_stream_tester_init(
+        &stream_tester,
+        allocator,
+        &(struct client_stream_tester_options){
+            .request = request,
+            .connection = tester.connection,
+            .use_manual_data_writes = true,
+        }));
+
+    testing_channel_drain_queued_tasks(&tester.testing_channel);
+    aws_http_message_destroy(request);
+
+    /* First write */
+    struct aws_byte_cursor data1 = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("write ");
+    struct aws_input_stream *stream1 = aws_input_stream_new_from_cursor(allocator, &data1);
+    struct aws_http_stream_write_data_options opts1 = {.data = stream1, .end_stream = false};
+    ASSERT_SUCCESS(aws_http_stream_write_data(stream_tester.stream, &opts1));
+
+    /* Second write with end_stream */
+    struct aws_byte_cursor data2 = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("more tests");
+    struct aws_input_stream *stream2 = aws_input_stream_new_from_cursor(allocator, &data2);
+    struct aws_http_stream_write_data_options opts2 = {.data = stream2, .end_stream = true};
+    ASSERT_SUCCESS(aws_http_stream_write_data(stream_tester.stream, &opts2));
+
+    testing_channel_drain_queued_tasks(&tester.testing_channel);
+
+    const char *expected = "PUT /plan.txt HTTP/1.1\r\n"
+                           "Transfer-Encoding: chunked\r\n"
+                           "\r\n"
+                           "6\r\n"
+                           "write "
+                           "\r\n"
+                           "A\r\n"
+                           "more tests"
+                           "\r\n"
+                           "0\r\n"
+                           "\r\n";
+    ASSERT_SUCCESS(testing_channel_check_written_messages_str(&tester.testing_channel, allocator, expected));
+
+    aws_input_stream_release(stream1);
+    aws_input_stream_release(stream2);
+    client_stream_tester_clean_up(&stream_tester);
+    ASSERT_SUCCESS(s_tester_clean_up(&tester));
+    return AWS_OP_SUCCESS;
+}
