API-2971: OutOfMemoryError: java heap space assetService.uploadFile

src/main/java/com/bynder/sdk/api/ApiFactory.java

@@ -69,9 +69,30 @@ public static BynderApi createBynderClient(final Configuration configuration) {
      * @return Implementation instance of the {@link OAuthApi} interface.
      */
     public static AmazonS3Api createAmazonS3Client(final String bucket) {
+        return createAmazonS3Client(bucket, new HttpConnectionSettings());
+    }
+
+    /**
+     * Creates an implementation of the Bynder OAuth2 endpoints defined in the {@link OAuthApi}
+     * interface.
+     *
+     * @param bucket AWS bucket URL.
+     * @param httpConnectionSettings HTTP connection settings (e.g. timeouts) to apply to the
+     * client used to upload file chunks to Amazon S3.
+     * @return Implementation instance of the {@link OAuthApi} interface.
+     */
+    public static AmazonS3Api createAmazonS3Client(
+            final String bucket,
+            final HttpConnectionSettings httpConnectionSettings
+    ) {
+        OkHttpClient.Builder httpClientBuilder = new OkHttpClient.Builder();
+        setHttpConnectionSettings(httpClientBuilder, httpConnectionSettings);
+        addUserAgentHeader(httpClientBuilder);
+
         return new Retrofit.Builder()
                 .baseUrl(bucket)
                 .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
+                .client(httpClientBuilder.build())
                 .build().create(AmazonS3Api.class);
     }
 
@@ -103,7 +124,7 @@ private static OkHttpClient createOkHttpClient(final Configuration configuration
         } else {
             setPermanentTokenInterceptor(httpClientBuilder, configuration);
         }
-        setHttpConnectionSettings(httpClientBuilder, configuration);
+        setHttpConnectionSettings(httpClientBuilder, configuration.getHttpConnectionSettings());
         addUserAgentHeader(httpClientBuilder);
         return httpClientBuilder.build();
     }
@@ -162,15 +183,13 @@ private static void setPermanentTokenInterceptor(
      * Sets the HTTP connection settings for the HTTP client.
      *
      * @param httpClientBuilder Builder instance of the HTTP client.
-     * @param configuration HTTP connection settings for the HTTP communication with
+     * @param httpConnectionSettings HTTP connection settings for the HTTP communication with
      * Bynder.
      */
     private static void setHttpConnectionSettings(
             final Builder httpClientBuilder,
-            final Configuration configuration
+            final HttpConnectionSettings httpConnectionSettings
     ) {
-        HttpConnectionSettings httpConnectionSettings = configuration.getHttpConnectionSettings();
-
         if (httpConnectionSettings.isLoggingInterceptorEnabled()) {
             httpClientBuilder.addInterceptor(
                     new HttpLoggingInterceptor().setLevel(HttpLoggingInterceptor.Level.BODY)

src/main/java/com/bynder/sdk/configuration/HttpConnectionSettings.java

@@ -17,7 +17,7 @@
  */
 public class HttpConnectionSettings {
 
-    public static final int DEFAULT_TIMEOUT_SECONDS = 30;
+    public static final int DEFAULT_TIMEOUT_SECONDS = 60;
 
     /**
      * Read timeout in seconds.

src/main/java/com/bynder/sdk/service/BynderClientImpl.java

@@ -90,7 +90,11 @@ public OAuthService getOAuthService() {
     @Override
     public AssetService getAssetService() {
         if (assetService == null) {
-            assetService = AssetService.Builder.create(bynderApi, queryDecoder);
+            assetService = AssetService.Builder.create(
+                    bynderApi,
+                    queryDecoder,
+                    configuration.getHttpConnectionSettings()
+            );
         }
 
         return assetService;

src/main/java/com/bynder/sdk/service/amazons3/AmazonS3Service.java

@@ -6,6 +6,7 @@
  */
 package com.bynder.sdk.service.amazons3;
 
+import com.bynder.sdk.configuration.HttpConnectionSettings;
 import com.bynder.sdk.model.upload.MultipartParameters;
 import com.bynder.sdk.service.asset.AssetService;
 import com.bynder.sdk.util.Indexed;
@@ -43,7 +44,14 @@ private Builder() {
         }
 
         public static AmazonS3Service create(final String bucket) {
-            return new AmazonS3ServiceImpl(bucket);
+            return new AmazonS3ServiceImpl(bucket, new HttpConnectionSettings());
+        }
+
+        public static AmazonS3Service create(
+                final String bucket,
+                final HttpConnectionSettings httpConnectionSettings
+        ) {
+            return new AmazonS3ServiceImpl(bucket, httpConnectionSettings);
         }
     }
 }

src/main/java/com/bynder/sdk/service/amazons3/AmazonS3ServiceImpl.java

@@ -8,6 +8,7 @@
 
 import com.bynder.sdk.api.AmazonS3Api;
 import com.bynder.sdk.api.ApiFactory;
+import com.bynder.sdk.configuration.HttpConnectionSettings;
 import com.bynder.sdk.model.upload.MultipartParameters;
 import com.bynder.sdk.util.Indexed;
 import com.bynder.sdk.util.RXUtils;
@@ -36,7 +37,18 @@ public class AmazonS3ServiceImpl implements AmazonS3Service {
      * @param bucket AWS bucket with the URL to upload the part to.
      */
     public AmazonS3ServiceImpl(final String bucket) {
-        amazonS3Api = ApiFactory.createAmazonS3Client(bucket);
+        this(bucket, new HttpConnectionSettings());
+    }
+
+    /**
+     * Initialises a new instance of the class.
+     *
+     * @param bucket AWS bucket with the URL to upload the part to.
+     * @param httpConnectionSettings HTTP connection settings (e.g. timeouts) to apply to the
+     * client used to upload file chunks to Amazon S3.
+     */
+    public AmazonS3ServiceImpl(final String bucket, final HttpConnectionSettings httpConnectionSettings) {
+        amazonS3Api = ApiFactory.createAmazonS3Client(bucket, httpConnectionSettings);
     }
 
     /**

src/main/java/com/bynder/sdk/service/asset/AssetService.java

@@ -7,6 +7,7 @@
 package com.bynder.sdk.service.asset;
 
 import com.bynder.sdk.api.BynderApi;
+import com.bynder.sdk.configuration.HttpConnectionSettings;
 import com.bynder.sdk.model.*;
 import com.bynder.sdk.model.upload.SaveMediaResponse;
 import com.bynder.sdk.model.upload.UploadAdditionalMediaResponse;
@@ -170,5 +171,13 @@ private Builder() {
         public static AssetService create(final BynderApi bynderApi, final QueryDecoder queryDecoder) {
             return new AssetServiceImpl(bynderApi, queryDecoder);
         }
+
+        public static AssetService create(
+                final BynderApi bynderApi,
+                final QueryDecoder queryDecoder,
+                final HttpConnectionSettings httpConnectionSettings
+        ) {
+            return new AssetServiceImpl(bynderApi, queryDecoder, httpConnectionSettings);
+        }
     }
 }

src/main/java/com/bynder/sdk/service/asset/AssetServiceImpl.java

@@ -7,6 +7,7 @@
 package com.bynder.sdk.service.asset;
 
 import com.bynder.sdk.api.BynderApi;
+import com.bynder.sdk.configuration.HttpConnectionSettings;
 import com.bynder.sdk.model.*;
 import com.bynder.sdk.model.upload.SaveMediaResponse;
 import com.bynder.sdk.model.upload.UploadAdditionalMediaResponse;
@@ -46,9 +47,25 @@ public class AssetServiceImpl implements AssetService {
      * @param queryDecoder Query decoder.
      */
     public AssetServiceImpl(final BynderApi bynderApi, final QueryDecoder queryDecoder) {
+        this(bynderApi, queryDecoder, new HttpConnectionSettings());
+    }
+
+    /**
+     * Initialises a new instance of the class.
+     *
+     * @param bynderApi Instance to handle the HTTP communication with the Bynder API.
+     * @param queryDecoder Query decoder.
+     * @param httpConnectionSettings HTTP connection settings (e.g. timeouts) to apply to the
+     * client used to upload file chunks to Amazon S3.
+     */
+    public AssetServiceImpl(
+            final BynderApi bynderApi,
+            final QueryDecoder queryDecoder,
+            final HttpConnectionSettings httpConnectionSettings
+    ) {
         this.bynderApi = bynderApi;
         this.queryDecoder = queryDecoder;
-        this.fileUploader = new FileUploader(bynderApi, queryDecoder);
+        this.fileUploader = new FileUploader(bynderApi, queryDecoder, httpConnectionSettings);
     }
 
     /**

src/main/java/com/bynder/sdk/service/upload/FileUploader.java

@@ -7,6 +7,7 @@
 package com.bynder.sdk.service.upload;
 
 import com.bynder.sdk.api.BynderApi;
+import com.bynder.sdk.configuration.HttpConnectionSettings;
 import com.bynder.sdk.exception.BynderUploadException;
 import com.bynder.sdk.model.upload.*;
 import com.bynder.sdk.query.decoder.QueryDecoder;
@@ -55,15 +56,38 @@ public class FileUploader {
      */
     private final QueryDecoder queryDecoder;
 
+    /**
+     * HTTP connection settings (e.g. timeouts) applied to the client used to upload file chunks
+     * to Amazon S3.
+     */
+    private final HttpConnectionSettings httpConnectionSettings;
+
     /**
      * Creates a new instance of the class.
      *
      * @param bynderApi    Instance to handle the HTTP communication with the Bynder API.
      * @param queryDecoder Query decoder.
      */
     public FileUploader(final BynderApi bynderApi, final QueryDecoder queryDecoder) {
+        this(bynderApi, queryDecoder, new HttpConnectionSettings());
+    }
+
+    /**
+     * Creates a new instance of the class.
+     *
+     * @param bynderApi    Instance to handle the HTTP communication with the Bynder API.
+     * @param queryDecoder Query decoder.
+     * @param httpConnectionSettings HTTP connection settings (e.g. timeouts) to apply to the
+     * client used to upload file chunks to Amazon S3.
+     */
+    public FileUploader(
+            final BynderApi bynderApi,
+            final QueryDecoder queryDecoder,
+            final HttpConnectionSettings httpConnectionSettings
+    ) {
         this.bynderApi = bynderApi;
         this.queryDecoder = queryDecoder;
+        this.httpConnectionSettings = httpConnectionSettings;
     }
 
     /**
@@ -75,7 +99,7 @@ public FileUploader(final BynderApi bynderApi, final QueryDecoder queryDecoder)
     public Single<SaveMediaResponse> uploadFile(final UploadQuery uploadQuery) {
         return getClosestS3Endpoint().flatMap(awsBucket -> {
             String filename = uploadQuery.getFilename();
-            AmazonS3Service amazonS3Service = AmazonS3Service.Builder.create(awsBucket);
+            AmazonS3Service amazonS3Service = AmazonS3Service.Builder.create(awsBucket, httpConnectionSettings);
             return getUploadInformation(new RequestUploadQuery(filename))
                     .flatMap(uploadRequest -> uploadChunk(
                             amazonS3Service,
@@ -96,7 +120,7 @@ public Single<SaveMediaResponse> uploadFile(final UploadQuery uploadQuery) {
     public Single<UploadAdditionalMediaResponse> uploadAdditionalFile(final UploadQuery uploadQuery) {
         return getClosestS3Endpoint().flatMap(awsBucket -> {
             String filename = uploadQuery.getFilename();
-            AmazonS3Service amazonS3Service = AmazonS3Service.Builder.create(awsBucket);
+            AmazonS3Service amazonS3Service = AmazonS3Service.Builder.create(awsBucket, httpConnectionSettings);
             return getUploadInformation(new RequestUploadQuery(filename))
                     .flatMap(uploadRequest -> uploadChunk(
                             amazonS3Service,
@@ -122,7 +146,7 @@ public Single<UploadAdditionalMediaResponse> uploadAdditionalFile(final UploadQu
     public Observable<UploadProgress> uploadFileWithProgress(final UploadQuery uploadQuery) {
         String filename = uploadQuery.getFilename();
         return getClosestS3Endpoint().flatMapObservable(awsBucket -> {
-            AmazonS3Service amazonS3Service = AmazonS3Service.Builder.create(awsBucket);
+            AmazonS3Service amazonS3Service = AmazonS3Service.Builder.create(awsBucket, httpConnectionSettings);
             return getUploadInformation(new RequestUploadQuery(filename))
                     .flatMapObservable(uploadRequest -> uploadChunk(
                             amazonS3Service,
@@ -144,7 +168,7 @@ private Observable<UploadProgress> uploadChunk(
         return RXUtils.mapWithIndex(
                 RXUtils.readFileChunks(uploadQuery.getFilepath(), MAX_CHUNK_SIZE),
                 1
-        ).flatMapSingle(chunk -> amazonS3Service.uploadPartToAmazon(
+        ).concatMapSingle(chunk -> amazonS3Service.uploadPartToAmazon(
                 chunk,
                 filename,
                 (int) ((fileSize - 1) / MAX_CHUNK_SIZE + 1),

src/main/java/com/bynder/sdk/util/RXUtils.java

@@ -9,6 +9,7 @@
 import java.io.File;
 import java.io.FileInputStream;
 import java.util.Arrays;
+import java.util.concurrent.atomic.AtomicInteger;
 
 /**
  * Utilities to make working with ReactiveX more convenient.
@@ -24,10 +25,8 @@ public class RXUtils {
      * @return observable that emits indexed items
      */
     public static <T> Observable<Indexed<T>> mapWithIndex(Observable<T> observable, int startFrom) {
-        return observable.zipWith(
-                Observable.range(startFrom, Integer.MAX_VALUE - startFrom),
-                Indexed::new
-        );
+        AtomicInteger index = new AtomicInteger(startFrom);
+        return observable.map(item -> new Indexed<>(item, index.getAndIncrement()));
     }
 
     /**

src/test/java/com/bynder/sdk/api/ApiFactoryTest.java

@@ -52,6 +52,15 @@ public void createAmazonS3Client() {
         assertNotNull(amazonS3Api);
     }
 
+    @Test
+    public void createAmazonS3ClientWithHttpConnectionSettings() {
+        AmazonS3Api amazonS3Api = ApiFactory.createAmazonS3Client(
+            BASE_URL,
+            new HttpConnectionSettings(null, null, null, 120, 120, true)
+        );
+        assertNotNull(amazonS3Api);
+    }
+
     @Test
     public void createOAuthClient() {
         OAuthApi oAuthApi = ApiFactory.createOAuthClient(BASE_URL);

src/test/java/com/bynder/sdk/configuration/HttpConnectionSettingsTest.java

@@ -27,7 +27,7 @@
  */
 public class HttpConnectionSettingsTest {
 
-    public static final int EXPECTED_TIMEOUT_SECONDS = 30;
+    public static final int EXPECTED_TIMEOUT_SECONDS = 60;
     public static final boolean EXPECTED_RETRY_ON_CONNECTION_FAILURE = true;
     public static final boolean EXPECTED_LOGGING_INTERCEPTOR_ENABLED = false;
 

src/test/java/com/bynder/sdk/service/asset/AssetServiceImplTest.java

@@ -11,13 +11,15 @@
 package com.bynder.sdk.service.asset;
 
 import com.bynder.sdk.api.BynderApi;
+import com.bynder.sdk.configuration.HttpConnectionSettings;
 import com.bynder.sdk.query.*;
 import com.bynder.sdk.query.decoder.QueryDecoder;
 import org.junit.Before;
 import org.junit.Test;
 import org.mockito.Mock;
 import org.mockito.MockitoAnnotations;
 
+import static org.junit.Assert.assertNotNull;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.*;
 
@@ -41,6 +43,18 @@ public void setUp() {
         assetService = AssetService.Builder.create(bynderApi, queryDecoder);
     }
 
+    @Test
+    public void createWithHttpConnectionSettings() {
+        AssetService serviceWithSettings = AssetService.Builder.create(
+            bynderApi, queryDecoder, new HttpConnectionSettings()
+        );
+
+        assertNotNull(serviceWithSettings);
+
+        serviceWithSettings.getBrands();
+        verify(bynderApi, times(1)).getBrands();
+    }
+
     @Test
     public void getBrands() {
         assetService.getBrands();

src/test/java/com/bynder/sdk/util/RXUtilsTest.java

@@ -12,6 +12,7 @@
 import java.util.Arrays;
 import java.util.Iterator;
 import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import org.junit.Before;
 import org.junit.Test;
@@ -62,6 +63,28 @@ public void mapWithIndex() {
         assertEquals(expected, actual);
     }
 
+    @Test
+    public void mapWithIndexDoesNotEagerlyDrainSource() {
+        // Regression test: mapWithIndex must pair each item with its index as it is emitted,
+        // without buffering the whole source upfront (which previously caused an
+        // OutOfMemoryError when indexing large file chunk streams).
+        AtomicInteger generatedCount = new AtomicInteger();
+        Observable<Integer> source = Observable.generate(emitter -> {
+            int count = generatedCount.incrementAndGet();
+            if (count > 5) {
+                emitter.onComplete();
+            } else {
+                emitter.onNext(count);
+            }
+        });
+
+        Indexed<Integer> first = RXUtils.mapWithIndex(source).blockingFirst();
+
+        assertEquals(1, generatedCount.get());
+        assertEquals(1, first.getValue().intValue());
+        assertEquals(0, first.getIndex());
+    }
+
     @Test
     public void handleSuccessResponse()
             throws HttpResponseException {