S3Storage implements Storage { @Override public String put(String key, long contentLength, UploadStreamProvider payload) throws IOException { checkArgument(key != null, "key is null"); ObjectMetadata meta = new ObjectMetadata(); meta.setContentLength(contentLength); try { return uploadRetryExecutor() .onRetry((exception, retryCount, retryLimit, retryWait) -> { logger.warn("Retrying uploading file bucket "+bucket+" key "+key+" error: "+exception); }) .retryIf((exception) -> { if (exception instanceof IOException || exception instanceof InterruptedException) { return false; } return true; }) .runInterruptible(() -> { try (InputStream in = payload.open()) { PutObjectRequest req = new PutObjectRequest(bucket, key, in, meta); UploadResult result = transferManager.upload(req).waitForUploadResult(); return result.getETag(); } }); } catch (InterruptedException ex) { throw Throwables.propagate(ex); } catch (RetryGiveupException ex) { Throwable cause = ex.getCause(); Throwables.propagateIfInstanceOf(cause, IOException.class); throw Throwables.propagate(cause); } } S3Storage(final Config config, AmazonS3Client client, String bucket); @Override StorageObject open(String key); @Override String put(String key, long contentLength,
            UploadStreamProvider payload); @Override void list(String keyPrefix, FileListing callback); @Override Optional<DirectDownloadHandle> getDirectDownloadHandle(String key); @Override Optional<DirectUploadHandle> getDirectUploadHandle(String key);  }
@Test public void putReturnsMd5() throws Exception { String checksum1 = storage.put("key1", 10, contents("0123456789")); String checksum2 = storage.put("key2", 5, contents("01234")); assertThat(checksum1, is(md5hex("0123456789"))); assertThat(checksum2, is(md5hex("01234"))); }