eclipse-edc · ndr-brt · Aug 10, 2022 · Jun 21, 2022 · Jun 27, 2022 · Jun 27, 2022
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -56,6 +56,7 @@ in the detailed section referring to by linking pull requests or issues.
 * Fix race condition in `ContractNegotiationIntegrationTest` (#1505)
 * Fix for change in Cosmos DB behavior on missing sort fields (#1514)
 * Effectively removed default LIMIT in SQL Contract Def Store (#1515)
+* Sending data to an AWS S3 presigned upload url does not work (#1643)
 
 ## [milestone-4] - 2022-06-07
 

@@ -26,11 +26,11 @@
 /**
  * Streams content into an OK HTTP buffered sink.
  */
-public class StreamingRequestBody extends RequestBody {
+public class ChunkedTransferRequestBody extends RequestBody {
     private final Supplier<InputStream> bodySupplier;
     private final String contentType;
 
-    public StreamingRequestBody(Supplier<InputStream> contentSupplier, String contentType) {
+    public ChunkedTransferRequestBody(Supplier<InputStream> contentSupplier, String contentType) {
         this.bodySupplier = contentSupplier;
         this.contentType = contentType;
     }

@@ -15,6 +15,7 @@
 package org.eclipse.dataspaceconnector.dataplane.http.pipeline;
 
 import okhttp3.Request;
+import okhttp3.RequestBody;
 import org.jetbrains.annotations.Nullable;
 
 import java.io.InputStream;
@@ -26,13 +27,15 @@
 public class HttpRequestParams {
 
     private static final String DEFAULT_CONTENT_TYPE = "application/octet-stream";
+    private static final Boolean DEFAULT_TRANSFER_CHUNKED = Boolean.TRUE;
 
     private String method;
     private String baseUrl;
     private String path;
     private String queryParams;
     private String contentType = DEFAULT_CONTENT_TYPE;
     private String body;
+    private Boolean transferChuncked = DEFAULT_TRANSFER_CHUNKED;
     private final Map<String, String> headers = new HashMap<>();
 
     /**
@@ -54,9 +57,7 @@ public Request toRequest() {
      * @return HTTP request.
      */
     public Request toRequest(@Nullable Supplier<InputStream> bodySupplier) {
-        var requestBody = (bodySupplier != null && contentType != null) ?
-                new StreamingRequestBody(bodySupplier, contentType) :
-                null;
+        var requestBody = createRequestBody(bodySupplier);
 
         var requestBuilder = new Request.Builder()
                 .url(toUrl())
@@ -65,6 +66,16 @@ public Request toRequest(@Nullable Supplier<InputStream> bodySupplier) {
         return requestBuilder.build();
     }
 
+    private RequestBody createRequestBody(@Nullable Supplier<InputStream> bodySupplier) {
+        if (bodySupplier == null || contentType == null) {
+            return null;
+        }
+
+        return transferChuncked ?
+                new ChunkedTransferRequestBody(bodySupplier, contentType) : new NonChunkedTransferRequestBody(bodySupplier, contentType);
+    }
+
+
     /**
      * Creates a URL from the base url, path and query parameters provided in input.
      *
@@ -128,6 +139,11 @@ public HttpRequestParams.Builder path(String path) {
             return this;
         }
 
+        public HttpRequestParams.Builder transferChuncked(Boolean transferChuncked) {
+            params.transferChuncked = transferChuncked;
+            return this;
+        }
+
         public HttpRequestParams build() {
             params.headers.forEach((s, s2) -> Objects.requireNonNull(s2, "value for header: " + s));
             Objects.requireNonNull(params.baseUrl, "baseUrl");

@@ -66,10 +66,13 @@ public HttpRequestParams apply(DataFlowRequest request) {
                     params.contentType(ct);
                     params.body(extractBody(address, request));
                 });
+        Optional.ofNullable(extractTransferChunked(address)).ifPresent(params::transferChuncked);
 
         return params.build();
     }
 
+    protected abstract Boolean extractTransferChunked(HttpDataAddress address);
+
     @NotNull
     protected abstract DataAddress selectAddress(DataFlowRequest request);
 

@@ -29,11 +29,17 @@
 public class HttpSinkRequestParamsSupplier extends HttpRequestParamsSupplier {
 
     private static final String DEFAULT_METHOD = "POST";
+    private static final Boolean DEFAULT_TRANSFER_CHUNKED = Boolean.TRUE;
 
     public HttpSinkRequestParamsSupplier(Vault vault) {
         super(vault);
     }
 
+    @Override
+    protected Boolean extractTransferChunked(HttpDataAddress address) {
+        return Optional.ofNullable(address.getTransferChunked()).orElse(DEFAULT_TRANSFER_CHUNKED);
+    }
+
     @Override
     protected @NotNull DataAddress selectAddress(DataFlowRequest request) {
         return request.getDestinationDataAddress();

@@ -42,6 +42,11 @@ public HttpSourceRequestParamsSupplier(Vault vault) {
         super(vault);
     }
 
+    @Override
+    protected Boolean extractTransferChunked(HttpDataAddress address) {
+        return null;
+    }
+
     @Override
     protected @NotNull DataAddress selectAddress(DataFlowRequest request) {
         return request.getSourceDataAddress();

@@ -0,0 +1,64 @@
+/*
+ *  Copyright (c) 2021 Microsoft Corporation
+ *
+ *  This program and the accompanying materials are made available under the
+ *  terms of the Apache License, Version 2.0 which is available at
+ *  https://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  SPDX-License-Identifier: Apache-2.0
+ *
+ *  Contributors:
+ *       Microsoft Corporation - initial API and implementation
+ *
+ */
+
+package org.eclipse.dataspaceconnector.dataplane.http.pipeline;
+
+import okhttp3.MediaType;
+import okhttp3.RequestBody;
+import okio.BufferedSink;
+import org.jetbrains.annotations.NotNull;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Optional;
+import java.util.function.Supplier;
+
+/**
+ * Streams content into an OK HTTP buffered sink.
+ */
+public class NonChunkedTransferRequestBody extends RequestBody {
+    private Optional<byte[]> bytes;
+    private final String contentType;
+
+    public NonChunkedTransferRequestBody(Supplier<InputStream> contentSupplier, String contentType) {
+        try {
+            this.bytes = Optional.of(contentSupplier.get().readAllBytes());
+        } catch (IOException e) {
+            this.bytes = Optional.empty();
+        }
+        this.contentType = contentType;
+    }
+
+    @Override
+    public long contentLength() {
+        return bytes.map(value -> value.length).orElse(0);
+    }
+
+    @Override
+    public MediaType contentType() {
+        return MediaType.parse(contentType);
+    }
+
+    @Override
+    public void writeTo(@NotNull BufferedSink sink) throws IOException {
+        if (!bytes.isPresent()) {
+            return;
+        }
+
+        try (var os = sink.outputStream()) {
+            os.write(bytes.get());
+        }
+    }
+}
+
@@ -27,7 +27,7 @@
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
-class StreamingRequestBodyTest {
+class ChunkedTransferRequestBodyTest {
     private static final Faker FAKER = new Faker();
 
     @Test
@@ -38,7 +38,7 @@ void verifyStreamingTransfer() throws IOException {
 
         when(sink.outputStream()).thenReturn(outputStream);
 
-        var body = new StreamingRequestBody(() -> new ByteArrayInputStream(content.getBytes()), HttpDataAddress.OCTET_STREAM);
+        var body = new ChunkedTransferRequestBody(() -> new ByteArrayInputStream(content.getBytes()), HttpDataAddress.OCTET_STREAM);
         body.writeTo(sink);
 
         assertThat(outputStream).hasToString(content);

@@ -9,6 +9,7 @@
  *
  *  Contributors:
  *       Amadeus - initial API and implementation
+ *       Siemens - add chunked parameter
  *
  */
 
@@ -139,6 +140,26 @@ void verifyAbstractMethodsInvokation() throws IOException {
         assertThat(body.contentType()).isEqualTo(MediaType.get(supplier.contentType));
         assertThat(HttpTestFixtures.formatRequestBodyAsString(body)).isEqualTo(supplier.body);
         assertThat(httpRequest.method()).isEqualTo(supplier.method);
+        assertThat(httpRequest.body().contentLength()).isEqualTo(-1L);
+    }
+
+    @Test
+    void verifyChunkedCall() throws IOException {
+        var dataAddress = HttpDataAddress.Builder.newInstance()
+                .baseUrl("http://" + FAKER.internet().url())
+                .build();
+        var request = createRequest(dataAddress);
+
+        var supplier = new TestHttpRequestParamsSupplier(vaultMock, false);
+        var httpRequest = supplier.apply(request).toRequest();
+
+        assertThat(httpRequest.url().url()).hasToString(dataAddress.getBaseUrl() + "/" + supplier.path + "?" + supplier.queryParams);
+        var body = httpRequest.body();
+        assertThat(body).isNotNull();
+        assertThat(body.contentType()).isEqualTo(MediaType.get(supplier.contentType));
+        assertThat(HttpTestFixtures.formatRequestBodyAsString(body)).isEqualTo(supplier.body);
+        assertThat(httpRequest.method()).isEqualTo(supplier.method);
+        assertThat(httpRequest.body().contentLength()).isEqualTo(supplier.body.getBytes().length);
     }
 
     private static DataFlowRequest createRequest(DataAddress source) {
@@ -156,16 +177,33 @@ public static final class TestHttpRequestParamsSupplier extends HttpRequestParam
         private final String queryParams;
         private final String contentType;
         private final String body;
+        private final Boolean chunked;
 
         private TestHttpRequestParamsSupplier(Vault vault) {
             super(vault);
             this.method = new Random().nextBoolean() ? HttpMethod.PUT.name() : HttpMethod.POST.name();
+            this.chunked = new Random().nextBoolean() ? true : null;
             this.path = FAKER.lorem().word();
             this.queryParams = FAKER.lorem().word();
             this.contentType = new Random().nextBoolean() ? APPLICATION_JSON : APPLICATION_X_WWW_FORM_URLENCODED;
             this.body = FAKER.lorem().word();
         }
 
+        private TestHttpRequestParamsSupplier(Vault vault, boolean chunked) {
+            super(vault);
+            this.method = new Random().nextBoolean() ? HttpMethod.PUT.name() : HttpMethod.POST.name();
+            this.chunked = chunked;
+            this.path = FAKER.lorem().word();
+            this.queryParams = FAKER.lorem().word();
+            this.contentType = new Random().nextBoolean() ? APPLICATION_JSON : APPLICATION_X_WWW_FORM_URLENCODED;
+            this.body = FAKER.lorem().word();
+        }
+
+        @Override
+        protected Boolean extractTransferChunked(HttpDataAddress address) {
+            return chunked;
+        }
+
         @Override
         protected @NotNull DataAddress selectAddress(DataFlowRequest request) {
             return request.getSourceDataAddress();

@@ -22,6 +22,8 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
+import java.util.Random;
+
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.Mockito.mock;
 
@@ -106,4 +108,16 @@ void extractContentType() {
     void extractBody() {
         assertThat(supplier.extractPath(mock(HttpDataAddress.class), null)).isNull();
     }
+
+    @Test
+    void extractTransferChunked() {
+        var chunked = new Random().nextBoolean();
+        var address = HttpDataAddress.Builder.newInstance()
+                .transferChunked(String.valueOf(chunked))
+                .build();
+
+        var result = supplier.extractTransferChunked(address);
+
+        assertThat(result).isEqualTo(chunked);
+    }
 }
@@ -25,6 +25,7 @@
 import org.junit.jupiter.api.Test;
 
 import java.util.Map;
+import java.util.Random;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
@@ -230,6 +231,18 @@ void extractBodyFilteredByProxy() {
         assertThat(result).isNull();
     }
 
+    @Test
+    void extractTransferChunked() {
+        var chunked = new Random().nextBoolean();
+        var address = HttpDataAddress.Builder.newInstance()
+                .transferChunked(String.valueOf(chunked))
+                .build();
+
+        var result = supplier.extractTransferChunked(address);
+
+        assertThat(result).isEqualTo(null);
+    }
+
     private static DataFlowRequest createRequest(DataAddress source) {
         return createRequest(source, Map.of());
     }