Merge branch 'part_streaming_input_stream' into parallel_input_stream_update

Author: TingDaoK

include/aws/s3/private/s3_parallel_input_stream.h

@@ -47,6 +47,15 @@ struct aws_parallel_input_stream_vtable {
      */
     struct aws_future_bool *(
         *read)(struct aws_parallel_input_stream *stream, uint64_t offset, size_t max_length, struct aws_byte_buf *dest);
+
+    /**
+     * Get the length of the stream.
+     *
+     * @param stream            The stream to get length from
+     * @param out_length        The output length
+     * @return                  AWS_OP_SUCCESS if success, otherwise AWS_OP_ERR
+     */
+    int (*get_length)(struct aws_parallel_input_stream *stream, int64_t *out_length);
 };
 
 AWS_EXTERN_C_BEGIN
@@ -95,6 +104,16 @@ struct aws_future_bool *aws_parallel_input_stream_read(
     size_t max_length,
     struct aws_byte_buf *dest);
 
+/**
+ * Get the total length of the parallel input stream.
+ *
+ * @param stream
+ * @param out_length
+ * @return AWS_S3_API
+ */
+AWS_S3_API
+int aws_parallel_input_stream_get_length(struct aws_parallel_input_stream *stream, int64_t *out_length);
+
 /**
  * Creates a new parallel input stream that reads from a file.
  * This stream uses an event loop group to perform file I/O operations asynchronously.

include/aws/s3/private/s3_part_streaming_input_stream.h

@@ -0,0 +1,41 @@
+/**
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0.
+ */
+
+#ifndef AWS_S3_PART_STREAMING_INPUT_STREAM_H
+#define AWS_S3_PART_STREAMING_INPUT_STREAM_H
+
+#include <aws/s3/s3.h>
+
+struct aws_input_stream;
+struct aws_s3_buffer_ticket;
+struct aws_parallel_input_stream;
+
+AWS_PUSH_SANE_WARNING_LEVEL
+AWS_EXTERN_C_BEGIN
+
+/**
+ * Creates a new streaming input stream that reads from a parallel input stream.
+ * This adapter allows using a parallel input stream with APIs that expect a standard input stream.
+ * The adapter uses double-buffering to read ahead and provide efficient streaming.
+ *
+ * @param allocator The allocator to use for memory allocation
+ * @param para_stream The parallel input stream to read from
+ * @param buffer_ticket The buffer pool ticket to use for buffering
+ * @param offset The starting offset in the stream
+ * @param request_body_size The total size to read
+ * @return A new input stream that reads from the parallel input stream
+ */
+AWS_S3_API
+struct aws_input_stream *aws_part_streaming_input_stream_new(
+    struct aws_allocator *allocator,
+    struct aws_parallel_input_stream *para_stream,
+    struct aws_s3_buffer_ticket *buffer_ticket,
+    uint64_t offset,
+    size_t request_body_size);
+
+AWS_EXTERN_C_END
+AWS_POP_SANE_WARNING_LEVEL
+
+#endif /* AWS_S3_PART_STREAMING_INPUT_STREAM_H */

source/s3_parallel_input_stream.c

@@ -44,6 +44,14 @@ struct aws_parallel_input_stream *aws_parallel_input_stream_release(struct aws_p
     return NULL;
 }
 
+int aws_parallel_input_stream_get_length(struct aws_parallel_input_stream *stream, int64_t *out_length) {
+    if (stream->vtable->get_length) {
+        return stream->vtable->get_length(stream, out_length);
+    } else {
+        return aws_raise_error(AWS_ERROR_UNSUPPORTED_OPERATION);
+    }
+}
+
 struct aws_future_bool *aws_parallel_input_stream_read(
     struct aws_parallel_input_stream *stream,
     uint64_t offset,
@@ -68,7 +76,7 @@ struct aws_parallel_input_stream_from_file_impl {
     struct aws_event_loop_group *reading_elg;
 };
 
-static void s_para_from_file_destroy(struct aws_parallel_input_stream *stream) {
+static void s_parallel_from_file_destroy(struct aws_parallel_input_stream *stream) {
     struct aws_parallel_input_stream_from_file_impl *impl =
         AWS_CONTAINER_OF(stream, struct aws_parallel_input_stream_from_file_impl, base);
 
@@ -102,6 +110,7 @@ static int s_read_from_file_impl(
     int rt_code = AWS_OP_ERR;
     FILE *file_stream = aws_fopen_safe(file_path, s_readonly_bytes_mode);
     if (file_stream == NULL) {
+        AWS_LOGF_ERROR(AWS_LS_S3_GENERAL, "Failed to open file %s", aws_string_c_str(file_path));
         return AWS_OP_ERR;
     }
 
@@ -158,7 +167,7 @@ static void s_s3_parallel_from_file_read_task(struct aws_task *task, void *arg,
     aws_mem_release(impl->base.alloc, read_task);
 }
 
-struct aws_future_bool *s_para_from_file_read(
+struct aws_future_bool *s_parallel_from_file_read(
     struct aws_parallel_input_stream *stream,
     uint64_t offset,
     size_t max_length,
@@ -199,9 +208,23 @@ struct aws_future_bool *s_para_from_file_read(
     return future;
 }
 
+int s_parallel_from_file_get_length(struct aws_parallel_input_stream *stream, int64_t *length) {
+    struct aws_parallel_input_stream_from_file_impl *impl =
+        AWS_CONTAINER_OF(stream, struct aws_parallel_input_stream_from_file_impl, base);
+    FILE *file = aws_fopen_safe(impl->file_path, s_readonly_bytes_mode);
+    if (!file) {
+        return AWS_OP_ERR;
+    }
+
+    int ret_val = aws_file_get_length(file, length);
+    fclose(file);
+    return ret_val;
+}
+
 static struct aws_parallel_input_stream_vtable s_parallel_input_stream_from_file_vtable = {
-    .destroy = s_para_from_file_destroy,
-    .read = s_para_from_file_read,
+    .destroy = s_parallel_from_file_destroy,
+    .read = s_parallel_from_file_read,
+    .get_length = s_parallel_from_file_get_length,
 };
 
 struct aws_parallel_input_stream *aws_parallel_input_stream_new_from_file(
@@ -219,7 +242,7 @@ struct aws_parallel_input_stream *aws_parallel_input_stream_new_from_file(
     if (!aws_path_exists(impl->file_path)) {
         /* If file path not exists, raise error from errno. */
         aws_translate_and_raise_io_error(errno);
-        s_para_from_file_destroy(&impl->base);
+        s_parallel_from_file_destroy(&impl->base);
         return NULL;
     }