Add StreamingFileUploadRequest (#424)

* avoid calling request.getInputStream() when the InputStream is already
in BlobParallelUploadOptions

gradle/libs.versions.toml

@@ -2,7 +2,7 @@
 micronaut = "4.7.4"
 micronaut-platform = "4.6.3"
 micronaut-docs = "2.0.0"
-micronaut-test = "4.5.0"
+micronaut-test = "4.6.0"
 micronaut-aws = "4.8.0"
 micronaut-azure = "5.8.0"
 micronaut-gcp = "5.8.0"
@@ -11,7 +11,7 @@ micronaut-test-resources = "2.6.2"
 micronaut-validation = "4.8.0"
 micronaut-logging = "1.5.0"
 
-groovy = '4.0.15'
+groovy = '4.0.23'
 spock = '2.3-groovy-4.0'
 gcp-libraries = '26.50.0'
 bytebuddy = '1.15.10'

object-storage-azure/src/main/java/io/micronaut/objectstorage/azure/AzureBlobStorageOperations.java

@@ -155,7 +155,8 @@ private UploadResponse<BlockBlobItem> doUpload(@NonNull UploadRequest request,
         BlockBlobItem item;
         if (request.getContentSize().isPresent()) {
             long length = request.getContentSize().get();
-            BlockBlobSimpleUploadOptions simpleUploadOptions = toBlockBlobSimpleUploadOptions(options, request.getInputStream(), length);
+            InputStream inputStream = Optional.ofNullable(options.getDataStream()).orElse(request.getInputStream());
+            BlockBlobSimpleUploadOptions simpleUploadOptions = toBlockBlobSimpleUploadOptions(options, inputStream, length);
             item = client
                 .getBlockBlobClient()
                 .uploadWithResponse(simpleUploadOptions, null, Context.NONE)

object-storage-core/build.gradle.kts

@@ -13,4 +13,7 @@ dependencies {
 
     testImplementation(projects.micronautObjectStorageTck)
     testImplementation(mn.micronaut.http.server.netty)
+
+    testImplementation(mn.reactor)
+
 }

object-storage-core/src/main/java/io/micronaut/objectstorage/request/StreamingFileUploadRequest.java

@@ -0,0 +1,91 @@
+/*
+ * Copyright 2017-2024 original authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.micronaut.objectstorage.request;
+
+import java.io.InputStream;
+import java.util.Collections;
+import java.util.Map;
+import java.util.Optional;
+
+import io.micronaut.core.annotation.NonNull;
+import io.micronaut.http.MediaType;
+import io.micronaut.http.multipart.StreamingFileUpload;
+
+/**
+ * 
+ * An {@link UploadRequest} backed by a {@link StreamingFileUpload}.
+ * @since 2.7.0
+ */
+public class StreamingFileUploadRequest implements UploadRequest {
+
+    @NonNull
+    private final StreamingFileUpload streamingFileUpload;
+    @NonNull
+    private final String key;
+
+    @NonNull
+    private Map<String, String> metadata;
+
+    public StreamingFileUploadRequest(@NonNull StreamingFileUpload streamingFileUpload) {
+        this(streamingFileUpload, streamingFileUpload.getName(), Collections.emptyMap());
+    }
+
+    public StreamingFileUploadRequest(@NonNull StreamingFileUpload streamingFileUpload, @NonNull String key) {
+        this(streamingFileUpload, key, Collections.emptyMap());
+    }
+
+    public StreamingFileUploadRequest(@NonNull StreamingFileUpload streamingFileUpload, @NonNull String key, @NonNull Map<String, String> metadata) {
+        this.streamingFileUpload = streamingFileUpload;
+        this.key = key;
+        this.metadata = metadata;
+    }
+
+    @NonNull
+    @Override
+    public Optional<String> getContentType() {
+        return streamingFileUpload.getContentType()
+            .map(MediaType::getName);
+    }
+
+    @NonNull
+    @Override
+    public String getKey() {
+        return key;
+    }
+
+    @NonNull
+    @Override
+    public Optional<Long> getContentSize() {
+        return Optional.of(streamingFileUpload.getSize());
+    }
+
+    @NonNull
+    @Override
+    public InputStream getInputStream() {
+    	return streamingFileUpload.asInputStream();
+    }
+
+    @Override
+    @NonNull
+    public Map<String, String> getMetadata() {
+        return this.metadata;
+    }
+
+    @Override
+    public void setMetadata(@NonNull Map<String, String> metadata) {
+        this.metadata = metadata;
+    }
+}

object-storage-core/src/main/java/io/micronaut/objectstorage/request/UploadRequest.java

@@ -17,6 +17,7 @@
 
 import io.micronaut.core.annotation.NonNull;
 import io.micronaut.http.multipart.CompletedFileUpload;
+import io.micronaut.http.multipart.StreamingFileUpload;
 
 import java.io.InputStream;
 import java.nio.file.Path;
@@ -91,6 +92,28 @@ static UploadRequest fromCompletedFileUpload(@NonNull CompletedFileUpload comple
                                                  @NonNull String key) {
         return new CompletedFileUploadRequest(completedFileUpload, key);
     }
+
+    /**
+     * @param streamingFileUpload the {@link StreamingFileUpload}.
+     * @return An {@link UploadRequest} from the given {@link StreamingFileUpload}.
+     * @since 2.7.0
+     */
+    @NonNull
+    static UploadRequest fromStreamingFileUpload(@NonNull StreamingFileUpload streamingFileUpload) {
+        return new StreamingFileUploadRequest(streamingFileUpload);
+    }
+
+    /**
+     * @param streamingFileUpload the {@link StreamingFileUpload}.
+     * @param key the key under which the file will be stored (<code>path/to/file</code>)
+     * @return An {@link UploadRequest} from the given {@link StreamingFileUpload} and key.
+     * @since 2.7.0
+     */
+    @NonNull
+    static UploadRequest fromStreamingFileUpload(@NonNull StreamingFileUpload streamingFileUpload,
+                                                 @NonNull String key) {
+        return new StreamingFileUploadRequest(streamingFileUpload, key);
+    }
 
 
     /**

object-storage-core/src/test/groovy/io/micronaut/objectstorage/request/StreamingFileUploadSpec.groovy

@@ -0,0 +1,64 @@
+package io.micronaut.objectstorage.request
+
+import io.micronaut.http.server.HttpServerConfiguration
+import io.micronaut.http.server.netty.MicronautHttpData
+import io.micronaut.http.server.netty.multipart.NettyStreamingFileUpload
+import io.micronaut.objectstorage.ObjectStorageOperationsSpecification
+import io.netty.buffer.Unpooled
+import io.netty.handler.codec.http.multipart.FileUpload
+import spock.lang.Specification
+import spock.lang.Subject
+
+import java.nio.charset.Charset
+import java.nio.charset.StandardCharsets
+import java.nio.file.Path
+
+import static io.micronaut.objectstorage.ObjectStorageOperationsSpecification.TEXT
+
+@Subject(StreamingFileUploadRequest)
+class StreamingFileUploadSpec extends Specification {
+
+    void "it can be created from a streaming file upload"() {
+        given:
+        NettyStreamingFileUpload streamingFileUpload = createStreamingFileUpload()
+
+        when:
+        StreamingFileUploadRequest request =
+                UploadRequest.fromStreamingFileUpload(streamingFileUpload) as StreamingFileUploadRequest
+
+        then:
+        assertThatRequestIsValid(request, streamingFileUpload.name)
+    }
+
+    void "it can be created from a streaming file upload and key"() {
+        given:
+        NettyStreamingFileUpload streamingFileUpload = createStreamingFileUpload()
+        String key = "path/tp=o/file.txt"
+
+        when:
+        StreamingFileUploadRequest request =
+                UploadRequest.fromStreamingFileUpload(streamingFileUpload, key) as StreamingFileUploadRequest
+
+        then:
+        assertThatRequestIsValid(request, key)
+
+    }
+
+    private NettyStreamingFileUpload createStreamingFileUpload() {
+        HttpServerConfiguration.MultipartConfiguration cfg = new HttpServerConfiguration.MultipartConfiguration()
+        cfg.mixed = true
+        FileUpload fileUpload = new MicronautHttpData.Factory(cfg, StandardCharsets.UTF_8).createFileUpload(null, "test-file.txt", "test-file.txt", "text/plain", "chunked", Charset.defaultCharset(), TEXT.length())
+
+        return new NettyStreamingFileUpload.Factory(cfg, null).create(fileUpload, null)
+
+    }
+
+    private void assertThatRequestIsValid(StreamingFileUploadRequest request, String expectedKey) {
+        request.with {
+            assert key == expectedKey
+            assert contentType.present
+            assert contentType.get() == "text/plain"
+
+        }
+    }
+}