hack to streaming data

Author: TingDaoK

include/aws/s3/private/s3_parallel_input_stream.h

@@ -99,6 +99,8 @@ struct aws_parallel_input_stream *aws_parallel_input_stream_new_from_file(
     struct aws_allocator *allocator,
     struct aws_byte_cursor file_name);
 
+const char *aws_parallel_input_stream_get_file_path(struct aws_parallel_input_stream *stream);
+
 AWS_EXTERN_C_END
 AWS_POP_SANE_WARNING_LEVEL
 

include/aws/s3/private/s3_request.h

@@ -138,6 +138,9 @@ struct aws_s3_request {
     /* Request body to use when sending the request. The contents of this body will be re-used if a request is
      * retried.*/
     struct aws_byte_buf request_body;
+    struct aws_input_stream *request_stream;
+    bool parallel;
+    int64_t content_length;
 
     /**
      * Ticket to acquire the buffer.

include/aws/s3/private/s3_request_messages.h

@@ -78,11 +78,18 @@ struct aws_http_message *aws_s3_create_multipart_upload_message_new(
     struct aws_http_message *base_message,
     const struct checksum_config_storage *checksum_config);
 
-/* Create an HTTP request for an S3 Put Object request, using the original request as a basis.  Creates and assigns a
- * body stream using the passed in buffer.  If multipart is not needed, part number and upload_id can be 0 and NULL,
- * respectively. */
-AWS_S3_API
-struct aws_http_message *aws_s3_upload_part_message_new(
+struct aws_http_message *aws_s3_upload_part_message_new_streaming(
+    struct aws_allocator *allocator,
+    struct aws_http_message *base_message,
+    uint32_t part_number,
+    struct aws_input_stream *input_stream,
+    int64_t stream_length,
+    const struct aws_string *upload_id);
+
+/* Create an HTTP request for an S3 Put Object request, using the original request as a basis.  Creates and assigns
+ * a body stream using the passed in buffer.  If multipart is not needed, part number and upload_id can be 0 and
+ * NULL, respectively. */
+AWS_S3_API struct aws_http_message *aws_s3_upload_part_message_new(
     struct aws_allocator *allocator,
     struct aws_http_message *base_message,
     struct aws_byte_buf *buffer,

source/s3_auto_ranged_put.c

@@ -6,6 +6,7 @@
 #include "aws/s3/private/s3_auto_ranged_put.h"
 #include "aws/s3/private/s3_checksums.h"
 #include "aws/s3/private/s3_list_parts.h"
+#include "aws/s3/private/s3_parallel_input_stream.h"
 #include "aws/s3/private/s3_request_messages.h"
 #include "aws/s3/private/s3_util.h"
 #include <aws/common/clock.h>
@@ -841,7 +842,7 @@ void s_s3_auto_ranged_put_schedule_prepare_request(
      * reading. */
     bool parallel_prepare =
         (meta_request->request_body_parallel_stream && request->request_tag == AWS_S3_AUTO_RANGED_PUT_REQUEST_TAG_PART);
-
+    request->parallel = parallel_prepare;
     aws_s3_meta_request_schedule_prepare_request_default_impl(
         meta_request, request, parallel_prepare /*parallel*/, callback, user_data);
 }
@@ -974,8 +975,16 @@ struct aws_future_http_message *s_s3_prepare_upload_part(struct aws_s3_request *
     part_prep->allocator = allocator;
     part_prep->request = request;
     part_prep->on_complete = aws_future_http_message_acquire(message_future);
+    if (request->parallel) {
 
-    if (request->num_times_prepared == 0) {
+        uint64_t offset = 0;
+        size_t request_body_size = s_compute_request_body_size(meta_request, request->part_number, &offset);
+        request->request_stream = aws_input_stream_new_from_file(
+            allocator, aws_parallel_input_stream_get_file_path(meta_request->request_body_parallel_stream));
+        request->content_length = request_body_size;
+        aws_input_stream_seek(request->request_stream, offset, AWS_SSB_BEGIN);
+        s_s3_prepare_upload_part_finish(part_prep, AWS_ERROR_SUCCESS);
+    } else if (request->num_times_prepared == 0) {
         /* Preparing request for the first time.
          * Next async step: read through the body stream until we've
          * skipped over parts that were already uploaded (in case we're resuming
@@ -1155,15 +1164,26 @@ static void s_s3_prepare_upload_part_finish(struct aws_s3_prepare_upload_part_jo
     }
 
     /* Create a new put-object message to upload a part. */
-    struct aws_http_message *message = aws_s3_upload_part_message_new(
-        meta_request->allocator,
-        meta_request->initial_request_message,
-        &request->request_body,
-        request->part_number,
-        auto_ranged_put->upload_id,
-        meta_request->should_compute_content_md5,
-        &meta_request->checksum_config,
-        checksum_buf);
+    struct aws_http_message *message = NULL;
+    if (request->parallel) {
+        message = aws_s3_upload_part_message_new_streaming(
+            meta_request->allocator,
+            meta_request->initial_request_message,
+            request->part_number,
+            request->request_stream,
+            request->content_length,
+            auto_ranged_put->upload_id);
+    } else {
+        message = aws_s3_upload_part_message_new(
+            meta_request->allocator,
+            meta_request->initial_request_message,
+            &request->request_body,
+            request->part_number,
+            auto_ranged_put->upload_id,
+            meta_request->should_compute_content_md5,
+            &meta_request->checksum_config,
+            checksum_buf);
+    }
     if (message == NULL) {
         aws_future_http_message_set_error(part_prep->on_complete, aws_last_error());
         goto on_done;

source/s3_parallel_input_stream.c

@@ -138,3 +138,8 @@ struct aws_parallel_input_stream *aws_parallel_input_stream_new_from_file(
     s_para_from_file_destroy(&impl->base);
     return NULL;
 }
+
+const char *aws_parallel_input_stream_get_file_path(struct aws_parallel_input_stream *stream) {
+    struct aws_parallel_input_stream_from_file_impl *impl = stream->impl;
+    return aws_string_c_str(impl->file_path);
+}

source/s3_request_messages.c

@@ -352,6 +352,58 @@ struct aws_http_message *aws_s3_create_multipart_upload_message_new(
     return NULL;
 }
 
+/* Create a new put object request from an existing put object request.  Currently just optionally adds part information
+ * for a multipart upload. */
+struct aws_http_message *aws_s3_upload_part_message_new_streaming(
+    struct aws_allocator *allocator,
+    struct aws_http_message *base_message,
+    uint32_t part_number,
+    struct aws_input_stream *input_stream,
+    int64_t stream_length,
+    const struct aws_string *upload_id) {
+    AWS_PRECONDITION(allocator);
+    AWS_PRECONDITION(base_message);
+    AWS_PRECONDITION(part_number > 0);
+
+    struct aws_http_message *message = aws_s3_message_util_copy_http_message_no_body_filter_headers(
+        allocator,
+        base_message,
+        g_s3_upload_part_excluded_headers,
+        AWS_ARRAY_SIZE(g_s3_upload_part_excluded_headers),
+        true /*exclude_x_amz_meta*/);
+
+    if (message == NULL) {
+        return NULL;
+    }
+
+    if (aws_s3_message_util_set_multipart_request_path(allocator, upload_id, part_number, false, message)) {
+        goto error_clean_up;
+    }
+    struct aws_http_headers *headers = aws_http_message_get_headers(message);
+
+    if (headers == NULL) {
+        return NULL;
+    }
+
+    char content_length_buffer[64] = "";
+    snprintf(content_length_buffer, sizeof(content_length_buffer), "%" PRIu64, (uint64_t)stream_length);
+    struct aws_byte_cursor content_length_cursor =
+        aws_byte_cursor_from_array(content_length_buffer, strlen(content_length_buffer));
+    if (aws_http_headers_set(headers, g_content_length_header_name, content_length_cursor)) {
+        goto error_clean_up;
+    }
+
+    aws_http_message_set_body_stream(message, input_stream);
+    /* Let the message take the full ownership */
+    aws_input_stream_release(input_stream);
+
+    return message;
+
+error_clean_up:
+    aws_http_message_release(message);
+    return NULL;
+}
+
 /* Create a new put object request from an existing put object request.  Currently just optionally adds part information
  * for a multipart upload. */
 struct aws_http_message *aws_s3_upload_part_message_new(