UploadManager { public UploadResponse upload(UploadRequest uploadDetails) { if (MultipartUtils.shouldUseMultipart( uploadConfiguration, uploadDetails.putObjectRequest.getContentLength())) { return multipartUpload(uploadDetails); } return singleUpload(uploadDetails, uploadDetails.putObjectRequest.getContentLength()); }  UploadResponse upload(UploadRequest uploadDetails);  }
@Test public void upload_singleUpload() throws IOException { UploadConfiguration uploadConfiguration = UploadConfiguration.builder().allowMultipartUploads(false).build(); UploadManager uploadManager = new UploadManager(objectStorage, uploadConfiguration); UploadRequest request = createUploadRequest(); ArgumentCaptor<PutObjectRequest> putRequestCaptor = ArgumentCaptor.forClass(PutObjectRequest.class); PutObjectResponse putResponse = PutObjectResponse.builder() .eTag("etag") .opcContentMd5("md5") .opcRequestId(REQ_ID) .opcClientRequestId(CLIENT_REQ_ID) .build(); when(objectStorage.putObject(putRequestCaptor.capture())).thenReturn(putResponse); UploadResponse uploadResponse = uploadManager.upload(request); assertNotNull(uploadResponse); assertEquals("etag", uploadResponse.getETag()); assertEquals("md5", uploadResponse.getContentMd5()); assertNull(uploadResponse.getMultipartMd5()); assertEquals(REQ_ID, uploadResponse.getOpcRequestId()); assertEquals(CLIENT_REQ_ID, uploadResponse.getOpcClientRequestId()); byte[] buffer = new byte[(int) CONTENT_LENGTH]; putRequestCaptor.getValue().getPutObjectBody().read(buffer); assertEquals(CONTENT, new String(buffer)); assertSame( UploadManager.RETRY_CONFIGURATION, putRequestCaptor.getValue().getRetryConfiguration()); assertEquals(CONTENT_LENGTH, putRequestCaptor.getValue().getContentLength().longValue()); assertEquals(CLIENT_REQ_ID, putRequestCaptor.getValue().getOpcClientRequestId()); assertSame(METADATA, putRequestCaptor.getValue().getOpcMeta()); }
@Test public void upload_singleUpload_enforceMd5() throws Exception { UploadConfiguration uploadConfiguration = UploadConfiguration.builder().allowMultipartUploads(false).enforceMd5(true).build(); UploadManager uploadManager = new UploadManager(objectStorage, uploadConfiguration); UploadRequest request = createUploadRequest(); ArgumentCaptor<PutObjectRequest> putRequestCaptor = ArgumentCaptor.forClass(PutObjectRequest.class); PutObjectResponse putResponse = PutObjectResponse.builder().build(); when(objectStorage.putObject(putRequestCaptor.capture())).thenReturn(putResponse); UploadResponse uploadResponse = uploadManager.upload(request); assertNotNull(uploadResponse); byte[] buffer = new byte[(int) CONTENT_LENGTH]; putRequestCaptor.getValue().getPutObjectBody().read(buffer); assertEquals(CONTENT, new String(buffer)); assertEquals(CONTENT_LENGTH, putRequestCaptor.getValue().getContentLength().longValue()); assertEquals( "U2yw5mJhFHg/U4cBMrrFyw==", putRequestCaptor.getValue().getContentMD5()); }
@Test public void upload_singleUpload_uploadRequestRetryConfiguration() throws Exception { RetryConfiguration retryConfiguration = RetryConfiguration.builder() .delayStrategy(new ExponentialBackoffDelayStrategy(120000)) .retryCondition(new DefaultRetryCondition()) .build(); UploadConfiguration uploadConfiguration = UploadConfiguration.builder().allowMultipartUploads(false).build(); UploadManager uploadManager = new UploadManager(objectStorage, uploadConfiguration); UploadRequest request = createUploadRequest(retryConfiguration); ArgumentCaptor<PutObjectRequest> putRequestCaptor = ArgumentCaptor.forClass(PutObjectRequest.class); PutObjectResponse putResponse = PutObjectResponse.builder() .eTag("etag") .opcContentMd5("md5") .opcRequestId(REQ_ID) .opcClientRequestId(CLIENT_REQ_ID) .build(); when(objectStorage.putObject(putRequestCaptor.capture())).thenReturn(putResponse); UploadResponse uploadResponse = uploadManager.upload(request); assertNotNull(uploadResponse); assertEquals("etag", uploadResponse.getETag()); assertEquals("md5", uploadResponse.getContentMd5()); assertNull(uploadResponse.getMultipartMd5()); assertEquals(REQ_ID, uploadResponse.getOpcRequestId()); assertEquals(CLIENT_REQ_ID, uploadResponse.getOpcClientRequestId()); byte[] buffer = new byte[(int) CONTENT_LENGTH]; putRequestCaptor.getValue().getPutObjectBody().read(buffer); assertSame(retryConfiguration, putRequestCaptor.getValue().getRetryConfiguration()); assertEquals(CONTENT, new String(buffer)); assertEquals(CONTENT_LENGTH, putRequestCaptor.getValue().getContentLength().longValue()); assertEquals(CLIENT_REQ_ID, putRequestCaptor.getValue().getOpcClientRequestId()); assertSame(METADATA, putRequestCaptor.getValue().getOpcMeta()); }
@Test(expected = BmcException.class) public void upload_singleUpload_enforceMd5_streamTooLargeToBuffer() { UploadConfiguration uploadConfiguration = UploadConfiguration.builder().allowMultipartUploads(false).enforceMd5(true).build(); UploadManager uploadManager = new UploadManager(objectStorage, uploadConfiguration); UploadRequest request = UploadRequest.builder(new ByteArrayInputStream(new byte[0]), Long.MAX_VALUE) .build(PutObjectRequest.builder().build()); uploadManager.upload(request); }
@Test public void upload_multipartUpload() { UploadConfiguration uploadConfiguration = UploadConfiguration.builder() .minimumLengthForMultipartUpload(10) .lengthPerUploadPart(10) .build(); UploadManager uploadManager = new UploadManager(objectStorage, uploadConfiguration) { @Override protected MultipartObjectAssembler createAssembler( PutObjectRequest request, UploadRequest uploadRequest, ExecutorService executorService) { return assembler; } }; UploadRequest request = createUploadRequest(); when(assembler.commit()) .thenReturn( CommitMultipartUploadResponse.builder() .eTag("finalEtag") .opcRequestId(REQ_ID) .opcClientRequestId(CLIENT_REQ_ID) .opcMultipartMd5("multipartMd5") .build()); UploadResponse uploadResponse = uploadManager.upload(request); validateUploadResponseForMultipart(uploadResponse); verify(assembler).newRequest(CONTENT_TYPE, CONTENT_LANG, CONTENT_ENCODING, METADATA); verify(assembler, times(2)) .addPart(any(InputStream.class), eq(CONTENT_LENGTH / 2), eq(null)); }
@Test public void upload_multipartUpload_enforceMd5() { UploadConfiguration uploadConfiguration = UploadConfiguration.builder() .minimumLengthForMultipartUpload(10) .lengthPerUploadPart(10) .enforceMd5(true) .build(); UploadManager uploadManager = new UploadManager(objectStorage, uploadConfiguration) { @Override protected MultipartObjectAssembler createAssembler( PutObjectRequest request, UploadRequest uploadRequest, ExecutorService executorService) { return assembler; } }; UploadRequest request = createUploadRequest(); when(assembler.commit()) .thenReturn( CommitMultipartUploadResponse.builder() .eTag("finalEtag") .opcRequestId(REQ_ID) .opcClientRequestId(CLIENT_REQ_ID) .build()); UploadResponse uploadResponse = uploadManager.upload(request); assertNotNull(uploadResponse); verify(assembler).newRequest(CONTENT_TYPE, CONTENT_LANG, CONTENT_ENCODING, METADATA); verify(assembler, times(2)) .addPart( any(InputStream.class), eq(CONTENT_LENGTH / 2), eq("5W4QR5ShjfX0H20th7TMZw==")); }
@Test public void upload_multipartUpload_duplicatableStream_canNotParallelize() { UploadConfiguration uploadConfiguration = UploadConfiguration.builder() .minimumLengthForMultipartUpload(10) .lengthPerUploadPart(10) .allowParallelUploads(false) .build(); UploadManager uploadManager = new UploadManager(objectStorage, uploadConfiguration) { @Override protected MultipartObjectAssembler createAssembler( PutObjectRequest request, UploadRequest uploadRequest, ExecutorService executorService) { return assembler; } }; UploadRequest request = createUploadRequest(); when(assembler.commit()) .thenReturn( CommitMultipartUploadResponse.builder() .eTag("finalEtag") .opcRequestId(REQ_ID) .opcClientRequestId(CLIENT_REQ_ID) .build()); UploadResponse uploadResponse = uploadManager.upload(request); assertNotNull(uploadResponse); verify(assembler).newRequest(CONTENT_TYPE, CONTENT_LANG, CONTENT_ENCODING, METADATA); verify(assembler, times(2)) .addPart(any(InputStream.class), eq(CONTENT_LENGTH / 2), eq(null)); }
@Test(expected = BmcException.class) public void upload_multipartUpload_exception() { UploadConfiguration uploadConfiguration = UploadConfiguration.builder() .minimumLengthForMultipartUpload(10) .lengthPerUploadPart(10) .allowParallelUploads(false) .build(); UploadManager uploadManager = new UploadManager(objectStorage, uploadConfiguration) { @Override protected MultipartObjectAssembler createAssembler( PutObjectRequest request, UploadRequest uploadRequest, ExecutorService executorService) { return assembler; } }; UploadRequest request = createUploadRequest(); when(assembler.newRequest(CONTENT_TYPE, CONTENT_LANG, CONTENT_ENCODING, METADATA)) .thenReturn(new MultipartManifestImpl("uploadId")); when(assembler.commit()).thenThrow(new BmcException(false, "test", null, null)); try { uploadManager.upload(request); } catch (Exception e) { verify(assembler).abort(); throw e; } }
@Test(expected = BmcException.class) public void upload_multipartUpload_exception_noAbort() { UploadConfiguration uploadConfiguration = UploadConfiguration.builder() .minimumLengthForMultipartUpload(10) .lengthPerUploadPart(10) .allowParallelUploads(false) .disableAutoAbort(true) .build(); UploadManager uploadManager = new UploadManager(objectStorage, uploadConfiguration) { @Override protected MultipartObjectAssembler createAssembler( PutObjectRequest request, UploadRequest uploadRequest, ExecutorService executorService) { return assembler; } }; UploadRequest request = createUploadRequest(); when(assembler.newRequest(CONTENT_TYPE, CONTENT_LANG, CONTENT_ENCODING, METADATA)) .thenReturn(new MultipartManifestImpl("uploadId")); when(assembler.commit()).thenThrow(new BmcException(false, "test", null, null)); try { uploadManager.upload(request); } catch (Exception e) { verify(assembler, times(0)).abort(); throw e; } }
@Test public void singleUpload_progressReporter() { final UploadConfiguration uploadConfiguration = UploadConfiguration.builder().allowMultipartUploads(false).build(); final UploadManager uploadManager = new UploadManager(objectStorage, uploadConfiguration); final AtomicInteger expectedProgressNotificationCount = new AtomicInteger(); when(objectStorage.putObject(any(PutObjectRequest.class))) .then( new Answer<PutObjectResponse>() { @Override public PutObjectResponse answer(InvocationOnMock invocationOnMock) throws Throwable { final PutObjectRequest putObjectRequest = invocationOnMock.getArgumentAt(0, null); final InputStream inputStream = putObjectRequest.getPutObjectBody(); byte[] buffer = new byte[READ_BLOCK_SIZE]; while (inputStream.read(buffer) != -1) { expectedProgressNotificationCount.getAndIncrement(); } return PutObjectResponse.builder().build(); } }); final ProgressReporter progressReporter = mock(ProgressReporter.class); final UploadResponse uploadResponse = uploadManager.upload(createUploadRequest(progressReporter)); assertNotNull(uploadResponse); verify(progressReporter, times(expectedProgressNotificationCount.get())) .onProgress(and(gt(0L), leq(CONTENT_LENGTH)), eq(CONTENT_LENGTH)); }
@Test public void multipartUpload_progressReporter() { final UploadManager uploadManager = new UploadManager(objectStorage, getMultipartUploadConfiguration()) { @Override protected MultipartObjectAssembler createAssembler( PutObjectRequest request, UploadRequest uploadRequest, ExecutorService executorService) { return assembler; } }; final AtomicInteger onProgressCallbackCount = new AtomicInteger(); when(assembler.addPart(any(InputStream.class), anyLong(), anyString())) .thenAnswer( new Answer<Integer>() { @Override public Integer answer(InvocationOnMock invocationOnMock) throws Throwable { final InputStream inputStream = invocationOnMock.getArgumentAt(0, null); final byte[] buffer = new byte[READ_BLOCK_SIZE]; while (inputStream.read(buffer) != -1) { onProgressCallbackCount.incrementAndGet(); } return ThreadLocalRandom.current().nextInt(); } }); when(assembler.commit()).thenReturn(CommitMultipartUploadResponse.builder().build()); final ProgressReporter progressReporter = mock(ProgressReporter.class); final UploadResponse uploadResponse = uploadManager.upload(createUploadRequest(progressReporter)); assertNotNull(uploadResponse); verify(progressReporter, times(onProgressCallbackCount.get())) .onProgress(and(gt(0L), leq(CONTENT_LENGTH)), eq(CONTENT_LENGTH)); }
@Test public void upload_multipartUpload_with_minConfiguration() { final UploadConfiguration uploadConfiguration = UploadConfiguration.builder() .minimumLengthForMultipartUpload(0) .lengthPerUploadPart(1) .build(); final UploadManager uploadManager = new UploadManager(objectStorage, uploadConfiguration) { @Override protected MultipartObjectAssembler createAssembler( PutObjectRequest request, UploadRequest uploadRequest, ExecutorService executorService) { return assembler; } }; final UploadRequest request = createUploadRequest(); when(assembler.commit()) .thenReturn( CommitMultipartUploadResponse.builder() .eTag("finalEtag") .opcRequestId(REQ_ID) .opcClientRequestId(CLIENT_REQ_ID) .opcMultipartMd5("multipartMd5") .build()); final UploadResponse uploadResponse = uploadManager.upload(request); validateUploadResponseForMultipart(uploadResponse); verify(assembler).newRequest(CONTENT_TYPE, CONTENT_LANG, CONTENT_ENCODING, METADATA); verify(assembler, times(20)) .addPart(any(InputStream.class), eq(CONTENT_LENGTH / 20), eq(null)); }