@Test public void testUploadExactMultipartSize() throws Exception { final SDSNodeIdProvider nodeid = new SDSNodeIdProvider(session); final SDSDirectS3UploadFeature feature = new SDSDirectS3UploadFeature(session, nodeid, new SDSDirectS3WriteFeature(session)); final Path room = new SDSDirectoryFeature(session, nodeid).mkdir( new Path(new AlphanumericRandomStringService().random(), EnumSet.of(Path.Type.directory, Path.Type.volume, Path.Type.triplecrypt)), null, new TransferStatus()); final Path test = new Path(room, new AlphanumericRandomStringService().random(), EnumSet.of(Path.Type.file)); final Local local = new Local(System.getProperty("java.io.tmpdir"), UUID.randomUUID().toString()); final byte[] random = RandomUtils.nextBytes(10 * 1024 * 1024); final OutputStream out = local.getOutputStream(false); IOUtils.write(random, out); out.close(); final TransferStatus status = new TransferStatus(); status.setLength(random.length); feature.upload(test, local, new BandwidthThrottle(BandwidthThrottle.UNLIMITED), new DisabledStreamListener(), status, new DisabledLoginCallback()); assertTrue(new SDSFindFeature(nodeid).find(test)); final PathAttributes attributes = new SDSListService(session, nodeid).list(room, new DisabledListProgressListener()).get(test).attributes(); assertEquals(random.length, attributes.getSize()); new SDSDeleteFeature(session, nodeid).delete(Collections.singletonList(room), new DisabledLoginCallback(), new Delete.DisabledCallback()); local.delete(); }
@Override public VersionId upload(final Path file, final Local local, final BandwidthThrottle throttle, final StreamListener listener, final TransferStatus status, final ConnectionCallback callback) throws BackgroundException { final ThreadPool pool = ThreadPoolFactory.get("multipart", concurrency); try { final CreateFileUploadRequest createFileUploadRequest = new CreateFileUploadRequest() .directS3Upload(true) .size(-1 == status.getLength() ? null : status.getLength()) .parentId(Long.parseLong(nodeid.getFileid(file.getParent(), new DisabledListProgressListener()))) .name(file.getName()); final CreateFileUploadResponse createFileUploadResponse = new NodesApi(session.getClient()) .createFileUploadChannel(createFileUploadRequest, StringUtils.EMPTY); if(log.isDebugEnabled()) { log.debug(String.format("upload started for %s with response %s", file, createFileUploadResponse)); } final Map<Integer, TransferStatus> etags = new HashMap<>(); final List<PresignedUrl> presignedUrls = this.retrievePresignedUrls(createFileUploadResponse, status); final List<Future<TransferStatus>> parts = new ArrayList<>(); final Local source; final Buffer buffer; if(nodeid.isEncrypted(file)) { source = TemporaryFileServiceFactory.get().create(new AlphanumericRandomStringService().random()); buffer = new FileBuffer(source); final BufferOutputStream temporary = new BufferOutputStream(buffer); if(log.isDebugEnabled()) { log.debug(String.format("Pre-compute file key tag for upload to S3 for %s", file)); } final InputStream in = local.getInputStream(); final OutputStream out; try { final ObjectReader reader = session.getClient().getJSON().getContext(null).readerFor(FileKey.class); final FileKey fileKey = reader.readValue(status.getFilekey().array()); out = new TripleCryptOutputStream<>(session, new StatusOutputStream<TransferStatus>(temporary) { @Override public TransferStatus getStatus() { return status; } }, Crypto.createFileEncryptionCipher(TripleCryptConverter.toCryptoPlainFileKey(fileKey)), status); } catch(CryptoSystemException | InvalidFileKeyException e) { throw new TripleCryptExceptionMappingService().map("Upload {0} failed", e, file); } new StreamCopier(status, new TransferStatus()).transfer(in, out); } else { source = local; buffer = Buffer.NULL; } final long size = status.getLength() + status.getOffset(); long offset = 0; long remaining = status.getLength(); for(int partNumber = 1; remaining > 0; partNumber++) { final long length = Math.min(Math.max((size / (MAXIMUM_UPLOAD_PARTS - 1)), partsize), remaining); final PresignedUrl presignedUrl = presignedUrls.get(partNumber - 1); parts.add(this.submit(pool, file, source, throttle, listener, status, presignedUrl.getUrl(), presignedUrl.getPartNumber(), offset, length, callback)); remaining -= length; offset += length; } for(Future<TransferStatus> future : parts) { try { final TransferStatus part = future.get(); etags.put(part.getPart(), part); } catch(InterruptedException e) { log.error("Part upload failed with interrupt failure"); status.setCanceled(); throw new ConnectionCanceledException(e); } catch(ExecutionException e) { log.warn(String.format("Part upload failed with execution failure %s", e.getMessage())); if(e.getCause() instanceof BackgroundException) { throw (BackgroundException) e.getCause(); } throw new BackgroundException(e.getCause()); } } if(nodeid.isEncrypted(file)) { buffer.close(); } final CompleteS3FileUploadRequest completeS3FileUploadRequest = new CompleteS3FileUploadRequest() .keepShareLinks(status.isExists() ? PreferencesFactory.get().getBoolean("sds.upload.sharelinks.keep") : false) .resolutionStrategy(status.isExists() ? CompleteS3FileUploadRequest.ResolutionStrategyEnum.OVERWRITE : CompleteS3FileUploadRequest.ResolutionStrategyEnum.FAIL); if(status.getFilekey() != null) { final ObjectReader reader = session.getClient().getJSON().getContext(null).readerFor(FileKey.class); final FileKey fileKey = reader.readValue(status.getFilekey().array()); final EncryptedFileKey encryptFileKey = Crypto.encryptFileKey( TripleCryptConverter.toCryptoPlainFileKey(fileKey), TripleCryptConverter.toCryptoUserPublicKey(session.keyPair().getPublicKeyContainer()) ); completeS3FileUploadRequest.setFileKey(TripleCryptConverter.toSwaggerFileKey(encryptFileKey)); } etags.forEach((key, value) -> completeS3FileUploadRequest.addPartsItem( new S3FileUploadPart().partEtag(value.getChecksum().hash).partNumber(key))); if(log.isDebugEnabled()) { log.debug(String.format("Complete file upload with %s for %s", completeS3FileUploadRequest, file)); } new NodesApi(session.getClient()).completeS3FileUpload(completeS3FileUploadRequest, createFileUploadResponse.getUploadId(), StringUtils.EMPTY); final ScheduledThreadPool polling = new ScheduledThreadPool(); final CountDownLatch done = new CountDownLatch(1); final AtomicReference<BackgroundException> failure = new AtomicReference<>(); final ScheduledFuture f = polling.repeat(new Runnable() { @Override public void run() { try { final S3FileUploadStatus uploadStatus = new NodesApi(session.getClient()) .requestUploadStatusFiles(createFileUploadResponse.getUploadId(), StringUtils.EMPTY); switch(uploadStatus.getStatus()) { case "finishing": break; case "transfer": failure.set(new InteroperabilityException(uploadStatus.getStatus())); done.countDown(); break; case "error": failure.set(new InteroperabilityException(uploadStatus.getErrorDetails().getMessage())); done.countDown(); break; case "done": status.setVersion(new VersionId(String.valueOf(uploadStatus.getNode().getId()))); done.countDown(); break; } } catch(ApiException e) { done.countDown(); failure.set(new SDSExceptionMappingService().map("Upload {0} failed", e, file)); } } }, PreferencesFactory.get().getLong("sds.upload.s3.status.period"), TimeUnit.MILLISECONDS); Uninterruptibles.awaitUninterruptibly(done); polling.shutdown(); if(null != failure.get()) { throw failure.get(); } status.setComplete(); return status.getVersion(); } catch(CryptoSystemException | InvalidFileKeyException | InvalidKeyPairException e) { throw new TripleCryptExceptionMappingService().map("Upload {0} failed", e, file); } catch(ApiException e) { throw new SDSExceptionMappingService().map("Upload {0} failed", e, file); } catch(IOException e) { throw new DefaultIOExceptionMappingService().map(e); } finally { pool.shutdown(false); } }
SDSDirectS3UploadFeature extends HttpUploadFeature<VersionId, MessageDigest> { @Override public VersionId upload(final Path file, final Local local, final BandwidthThrottle throttle, final StreamListener listener, final TransferStatus status, final ConnectionCallback callback) throws BackgroundException { final ThreadPool pool = ThreadPoolFactory.get("multipart", concurrency); try { final CreateFileUploadRequest createFileUploadRequest = new CreateFileUploadRequest() .directS3Upload(true) .size(-1 == status.getLength() ? null : status.getLength()) .parentId(Long.parseLong(nodeid.getFileid(file.getParent(), new DisabledListProgressListener()))) .name(file.getName()); final CreateFileUploadResponse createFileUploadResponse = new NodesApi(session.getClient()) .createFileUploadChannel(createFileUploadRequest, StringUtils.EMPTY); if(log.isDebugEnabled()) { log.debug(String.format("upload started for %s with response %s", file, createFileUploadResponse)); } final Map<Integer, TransferStatus> etags = new HashMap<>(); final List<PresignedUrl> presignedUrls = this.retrievePresignedUrls(createFileUploadResponse, status); final List<Future<TransferStatus>> parts = new ArrayList<>(); final Local source; final Buffer buffer; if(nodeid.isEncrypted(file)) { source = TemporaryFileServiceFactory.get().create(new AlphanumericRandomStringService().random()); buffer = new FileBuffer(source); final BufferOutputStream temporary = new BufferOutputStream(buffer); if(log.isDebugEnabled()) { log.debug(String.format("Pre-compute file key tag for upload to S3 for %s", file)); } final InputStream in = local.getInputStream(); final OutputStream out; try { final ObjectReader reader = session.getClient().getJSON().getContext(null).readerFor(FileKey.class); final FileKey fileKey = reader.readValue(status.getFilekey().array()); out = new TripleCryptOutputStream<>(session, new StatusOutputStream<TransferStatus>(temporary) { @Override public TransferStatus getStatus() { return status; } }, Crypto.createFileEncryptionCipher(TripleCryptConverter.toCryptoPlainFileKey(fileKey)), status); } catch(CryptoSystemException | InvalidFileKeyException e) { throw new TripleCryptExceptionMappingService().map("Upload {0} failed", e, file); } new StreamCopier(status, new TransferStatus()).transfer(in, out); } else { source = local; buffer = Buffer.NULL; } final long size = status.getLength() + status.getOffset(); long offset = 0; long remaining = status.getLength(); for(int partNumber = 1; remaining > 0; partNumber++) { final long length = Math.min(Math.max((size / (MAXIMUM_UPLOAD_PARTS - 1)), partsize), remaining); final PresignedUrl presignedUrl = presignedUrls.get(partNumber - 1); parts.add(this.submit(pool, file, source, throttle, listener, status, presignedUrl.getUrl(), presignedUrl.getPartNumber(), offset, length, callback)); remaining -= length; offset += length; } for(Future<TransferStatus> future : parts) { try { final TransferStatus part = future.get(); etags.put(part.getPart(), part); } catch(InterruptedException e) { log.error("Part upload failed with interrupt failure"); status.setCanceled(); throw new ConnectionCanceledException(e); } catch(ExecutionException e) { log.warn(String.format("Part upload failed with execution failure %s", e.getMessage())); if(e.getCause() instanceof BackgroundException) { throw (BackgroundException) e.getCause(); } throw new BackgroundException(e.getCause()); } } if(nodeid.isEncrypted(file)) { buffer.close(); } final CompleteS3FileUploadRequest completeS3FileUploadRequest = new CompleteS3FileUploadRequest() .keepShareLinks(status.isExists() ? PreferencesFactory.get().getBoolean("sds.upload.sharelinks.keep") : false) .resolutionStrategy(status.isExists() ? CompleteS3FileUploadRequest.ResolutionStrategyEnum.OVERWRITE : CompleteS3FileUploadRequest.ResolutionStrategyEnum.FAIL); if(status.getFilekey() != null) { final ObjectReader reader = session.getClient().getJSON().getContext(null).readerFor(FileKey.class); final FileKey fileKey = reader.readValue(status.getFilekey().array()); final EncryptedFileKey encryptFileKey = Crypto.encryptFileKey( TripleCryptConverter.toCryptoPlainFileKey(fileKey), TripleCryptConverter.toCryptoUserPublicKey(session.keyPair().getPublicKeyContainer()) ); completeS3FileUploadRequest.setFileKey(TripleCryptConverter.toSwaggerFileKey(encryptFileKey)); } etags.forEach((key, value) -> completeS3FileUploadRequest.addPartsItem( new S3FileUploadPart().partEtag(value.getChecksum().hash).partNumber(key))); if(log.isDebugEnabled()) { log.debug(String.format("Complete file upload with %s for %s", completeS3FileUploadRequest, file)); } new NodesApi(session.getClient()).completeS3FileUpload(completeS3FileUploadRequest, createFileUploadResponse.getUploadId(), StringUtils.EMPTY); final ScheduledThreadPool polling = new ScheduledThreadPool(); final CountDownLatch done = new CountDownLatch(1); final AtomicReference<BackgroundException> failure = new AtomicReference<>(); final ScheduledFuture f = polling.repeat(new Runnable() { @Override public void run() { try { final S3FileUploadStatus uploadStatus = new NodesApi(session.getClient()) .requestUploadStatusFiles(createFileUploadResponse.getUploadId(), StringUtils.EMPTY); switch(uploadStatus.getStatus()) { case "finishing": break; case "transfer": failure.set(new InteroperabilityException(uploadStatus.getStatus())); done.countDown(); break; case "error": failure.set(new InteroperabilityException(uploadStatus.getErrorDetails().getMessage())); done.countDown(); break; case "done": status.setVersion(new VersionId(String.valueOf(uploadStatus.getNode().getId()))); done.countDown(); break; } } catch(ApiException e) { done.countDown(); failure.set(new SDSExceptionMappingService().map("Upload {0} failed", e, file)); } } }, PreferencesFactory.get().getLong("sds.upload.s3.status.period"), TimeUnit.MILLISECONDS); Uninterruptibles.awaitUninterruptibly(done); polling.shutdown(); if(null != failure.get()) { throw failure.get(); } status.setComplete(); return status.getVersion(); } catch(CryptoSystemException | InvalidFileKeyException | InvalidKeyPairException e) { throw new TripleCryptExceptionMappingService().map("Upload {0} failed", e, file); } catch(ApiException e) { throw new SDSExceptionMappingService().map("Upload {0} failed", e, file); } catch(IOException e) { throw new DefaultIOExceptionMappingService().map(e); } finally { pool.shutdown(false); } } }
SDSDirectS3UploadFeature extends HttpUploadFeature<VersionId, MessageDigest> { @Override public VersionId upload(final Path file, final Local local, final BandwidthThrottle throttle, final StreamListener listener, final TransferStatus status, final ConnectionCallback callback) throws BackgroundException { final ThreadPool pool = ThreadPoolFactory.get("multipart", concurrency); try { final CreateFileUploadRequest createFileUploadRequest = new CreateFileUploadRequest() .directS3Upload(true) .size(-1 == status.getLength() ? null : status.getLength()) .parentId(Long.parseLong(nodeid.getFileid(file.getParent(), new DisabledListProgressListener()))) .name(file.getName()); final CreateFileUploadResponse createFileUploadResponse = new NodesApi(session.getClient()) .createFileUploadChannel(createFileUploadRequest, StringUtils.EMPTY); if(log.isDebugEnabled()) { log.debug(String.format("upload started for %s with response %s", file, createFileUploadResponse)); } final Map<Integer, TransferStatus> etags = new HashMap<>(); final List<PresignedUrl> presignedUrls = this.retrievePresignedUrls(createFileUploadResponse, status); final List<Future<TransferStatus>> parts = new ArrayList<>(); final Local source; final Buffer buffer; if(nodeid.isEncrypted(file)) { source = TemporaryFileServiceFactory.get().create(new AlphanumericRandomStringService().random()); buffer = new FileBuffer(source); final BufferOutputStream temporary = new BufferOutputStream(buffer); if(log.isDebugEnabled()) { log.debug(String.format("Pre-compute file key tag for upload to S3 for %s", file)); } final InputStream in = local.getInputStream(); final OutputStream out; try { final ObjectReader reader = session.getClient().getJSON().getContext(null).readerFor(FileKey.class); final FileKey fileKey = reader.readValue(status.getFilekey().array()); out = new TripleCryptOutputStream<>(session, new StatusOutputStream<TransferStatus>(temporary) { @Override public TransferStatus getStatus() { return status; } }, Crypto.createFileEncryptionCipher(TripleCryptConverter.toCryptoPlainFileKey(fileKey)), status); } catch(CryptoSystemException | InvalidFileKeyException e) { throw new TripleCryptExceptionMappingService().map("Upload {0} failed", e, file); } new StreamCopier(status, new TransferStatus()).transfer(in, out); } else { source = local; buffer = Buffer.NULL; } final long size = status.getLength() + status.getOffset(); long offset = 0; long remaining = status.getLength(); for(int partNumber = 1; remaining > 0; partNumber++) { final long length = Math.min(Math.max((size / (MAXIMUM_UPLOAD_PARTS - 1)), partsize), remaining); final PresignedUrl presignedUrl = presignedUrls.get(partNumber - 1); parts.add(this.submit(pool, file, source, throttle, listener, status, presignedUrl.getUrl(), presignedUrl.getPartNumber(), offset, length, callback)); remaining -= length; offset += length; } for(Future<TransferStatus> future : parts) { try { final TransferStatus part = future.get(); etags.put(part.getPart(), part); } catch(InterruptedException e) { log.error("Part upload failed with interrupt failure"); status.setCanceled(); throw new ConnectionCanceledException(e); } catch(ExecutionException e) { log.warn(String.format("Part upload failed with execution failure %s", e.getMessage())); if(e.getCause() instanceof BackgroundException) { throw (BackgroundException) e.getCause(); } throw new BackgroundException(e.getCause()); } } if(nodeid.isEncrypted(file)) { buffer.close(); } final CompleteS3FileUploadRequest completeS3FileUploadRequest = new CompleteS3FileUploadRequest() .keepShareLinks(status.isExists() ? PreferencesFactory.get().getBoolean("sds.upload.sharelinks.keep") : false) .resolutionStrategy(status.isExists() ? CompleteS3FileUploadRequest.ResolutionStrategyEnum.OVERWRITE : CompleteS3FileUploadRequest.ResolutionStrategyEnum.FAIL); if(status.getFilekey() != null) { final ObjectReader reader = session.getClient().getJSON().getContext(null).readerFor(FileKey.class); final FileKey fileKey = reader.readValue(status.getFilekey().array()); final EncryptedFileKey encryptFileKey = Crypto.encryptFileKey( TripleCryptConverter.toCryptoPlainFileKey(fileKey), TripleCryptConverter.toCryptoUserPublicKey(session.keyPair().getPublicKeyContainer()) ); completeS3FileUploadRequest.setFileKey(TripleCryptConverter.toSwaggerFileKey(encryptFileKey)); } etags.forEach((key, value) -> completeS3FileUploadRequest.addPartsItem( new S3FileUploadPart().partEtag(value.getChecksum().hash).partNumber(key))); if(log.isDebugEnabled()) { log.debug(String.format("Complete file upload with %s for %s", completeS3FileUploadRequest, file)); } new NodesApi(session.getClient()).completeS3FileUpload(completeS3FileUploadRequest, createFileUploadResponse.getUploadId(), StringUtils.EMPTY); final ScheduledThreadPool polling = new ScheduledThreadPool(); final CountDownLatch done = new CountDownLatch(1); final AtomicReference<BackgroundException> failure = new AtomicReference<>(); final ScheduledFuture f = polling.repeat(new Runnable() { @Override public void run() { try { final S3FileUploadStatus uploadStatus = new NodesApi(session.getClient()) .requestUploadStatusFiles(createFileUploadResponse.getUploadId(), StringUtils.EMPTY); switch(uploadStatus.getStatus()) { case "finishing": break; case "transfer": failure.set(new InteroperabilityException(uploadStatus.getStatus())); done.countDown(); break; case "error": failure.set(new InteroperabilityException(uploadStatus.getErrorDetails().getMessage())); done.countDown(); break; case "done": status.setVersion(new VersionId(String.valueOf(uploadStatus.getNode().getId()))); done.countDown(); break; } } catch(ApiException e) { done.countDown(); failure.set(new SDSExceptionMappingService().map("Upload {0} failed", e, file)); } } }, PreferencesFactory.get().getLong("sds.upload.s3.status.period"), TimeUnit.MILLISECONDS); Uninterruptibles.awaitUninterruptibly(done); polling.shutdown(); if(null != failure.get()) { throw failure.get(); } status.setComplete(); return status.getVersion(); } catch(CryptoSystemException | InvalidFileKeyException | InvalidKeyPairException e) { throw new TripleCryptExceptionMappingService().map("Upload {0} failed", e, file); } catch(ApiException e) { throw new SDSExceptionMappingService().map("Upload {0} failed", e, file); } catch(IOException e) { throw new DefaultIOExceptionMappingService().map(e); } finally { pool.shutdown(false); } } SDSDirectS3UploadFeature(final SDSSession session, final SDSNodeIdProvider nodeid, final Write<VersionId> writer);  SDSDirectS3UploadFeature(final SDSSession session, final SDSNodeIdProvider nodeid, final Write<VersionId> writer, final Long partsize, final Integer concurrency); }
SDSDirectS3UploadFeature extends HttpUploadFeature<VersionId, MessageDigest> { @Override public VersionId upload(final Path file, final Local local, final BandwidthThrottle throttle, final StreamListener listener, final TransferStatus status, final ConnectionCallback callback) throws BackgroundException { final ThreadPool pool = ThreadPoolFactory.get("multipart", concurrency); try { final CreateFileUploadRequest createFileUploadRequest = new CreateFileUploadRequest() .directS3Upload(true) .size(-1 == status.getLength() ? null : status.getLength()) .parentId(Long.parseLong(nodeid.getFileid(file.getParent(), new DisabledListProgressListener()))) .name(file.getName()); final CreateFileUploadResponse createFileUploadResponse = new NodesApi(session.getClient()) .createFileUploadChannel(createFileUploadRequest, StringUtils.EMPTY); if(log.isDebugEnabled()) { log.debug(String.format("upload started for %s with response %s", file, createFileUploadResponse)); } final Map<Integer, TransferStatus> etags = new HashMap<>(); final List<PresignedUrl> presignedUrls = this.retrievePresignedUrls(createFileUploadResponse, status); final List<Future<TransferStatus>> parts = new ArrayList<>(); final Local source; final Buffer buffer; if(nodeid.isEncrypted(file)) { source = TemporaryFileServiceFactory.get().create(new AlphanumericRandomStringService().random()); buffer = new FileBuffer(source); final BufferOutputStream temporary = new BufferOutputStream(buffer); if(log.isDebugEnabled()) { log.debug(String.format("Pre-compute file key tag for upload to S3 for %s", file)); } final InputStream in = local.getInputStream(); final OutputStream out; try { final ObjectReader reader = session.getClient().getJSON().getContext(null).readerFor(FileKey.class); final FileKey fileKey = reader.readValue(status.getFilekey().array()); out = new TripleCryptOutputStream<>(session, new StatusOutputStream<TransferStatus>(temporary) { @Override public TransferStatus getStatus() { return status; } }, Crypto.createFileEncryptionCipher(TripleCryptConverter.toCryptoPlainFileKey(fileKey)), status); } catch(CryptoSystemException | InvalidFileKeyException e) { throw new TripleCryptExceptionMappingService().map("Upload {0} failed", e, file); } new StreamCopier(status, new TransferStatus()).transfer(in, out); } else { source = local; buffer = Buffer.NULL; } final long size = status.getLength() + status.getOffset(); long offset = 0; long remaining = status.getLength(); for(int partNumber = 1; remaining > 0; partNumber++) { final long length = Math.min(Math.max((size / (MAXIMUM_UPLOAD_PARTS - 1)), partsize), remaining); final PresignedUrl presignedUrl = presignedUrls.get(partNumber - 1); parts.add(this.submit(pool, file, source, throttle, listener, status, presignedUrl.getUrl(), presignedUrl.getPartNumber(), offset, length, callback)); remaining -= length; offset += length; } for(Future<TransferStatus> future : parts) { try { final TransferStatus part = future.get(); etags.put(part.getPart(), part); } catch(InterruptedException e) { log.error("Part upload failed with interrupt failure"); status.setCanceled(); throw new ConnectionCanceledException(e); } catch(ExecutionException e) { log.warn(String.format("Part upload failed with execution failure %s", e.getMessage())); if(e.getCause() instanceof BackgroundException) { throw (BackgroundException) e.getCause(); } throw new BackgroundException(e.getCause()); } } if(nodeid.isEncrypted(file)) { buffer.close(); } final CompleteS3FileUploadRequest completeS3FileUploadRequest = new CompleteS3FileUploadRequest() .keepShareLinks(status.isExists() ? PreferencesFactory.get().getBoolean("sds.upload.sharelinks.keep") : false) .resolutionStrategy(status.isExists() ? CompleteS3FileUploadRequest.ResolutionStrategyEnum.OVERWRITE : CompleteS3FileUploadRequest.ResolutionStrategyEnum.FAIL); if(status.getFilekey() != null) { final ObjectReader reader = session.getClient().getJSON().getContext(null).readerFor(FileKey.class); final FileKey fileKey = reader.readValue(status.getFilekey().array()); final EncryptedFileKey encryptFileKey = Crypto.encryptFileKey( TripleCryptConverter.toCryptoPlainFileKey(fileKey), TripleCryptConverter.toCryptoUserPublicKey(session.keyPair().getPublicKeyContainer()) ); completeS3FileUploadRequest.setFileKey(TripleCryptConverter.toSwaggerFileKey(encryptFileKey)); } etags.forEach((key, value) -> completeS3FileUploadRequest.addPartsItem( new S3FileUploadPart().partEtag(value.getChecksum().hash).partNumber(key))); if(log.isDebugEnabled()) { log.debug(String.format("Complete file upload with %s for %s", completeS3FileUploadRequest, file)); } new NodesApi(session.getClient()).completeS3FileUpload(completeS3FileUploadRequest, createFileUploadResponse.getUploadId(), StringUtils.EMPTY); final ScheduledThreadPool polling = new ScheduledThreadPool(); final CountDownLatch done = new CountDownLatch(1); final AtomicReference<BackgroundException> failure = new AtomicReference<>(); final ScheduledFuture f = polling.repeat(new Runnable() { @Override public void run() { try { final S3FileUploadStatus uploadStatus = new NodesApi(session.getClient()) .requestUploadStatusFiles(createFileUploadResponse.getUploadId(), StringUtils.EMPTY); switch(uploadStatus.getStatus()) { case "finishing": break; case "transfer": failure.set(new InteroperabilityException(uploadStatus.getStatus())); done.countDown(); break; case "error": failure.set(new InteroperabilityException(uploadStatus.getErrorDetails().getMessage())); done.countDown(); break; case "done": status.setVersion(new VersionId(String.valueOf(uploadStatus.getNode().getId()))); done.countDown(); break; } } catch(ApiException e) { done.countDown(); failure.set(new SDSExceptionMappingService().map("Upload {0} failed", e, file)); } } }, PreferencesFactory.get().getLong("sds.upload.s3.status.period"), TimeUnit.MILLISECONDS); Uninterruptibles.awaitUninterruptibly(done); polling.shutdown(); if(null != failure.get()) { throw failure.get(); } status.setComplete(); return status.getVersion(); } catch(CryptoSystemException | InvalidFileKeyException | InvalidKeyPairException e) { throw new TripleCryptExceptionMappingService().map("Upload {0} failed", e, file); } catch(ApiException e) { throw new SDSExceptionMappingService().map("Upload {0} failed", e, file); } catch(IOException e) { throw new DefaultIOExceptionMappingService().map(e); } finally { pool.shutdown(false); } } SDSDirectS3UploadFeature(final SDSSession session, final SDSNodeIdProvider nodeid, final Write<VersionId> writer);  SDSDirectS3UploadFeature(final SDSSession session, final SDSNodeIdProvider nodeid, final Write<VersionId> writer, final Long partsize, final Integer concurrency); @Override VersionId upload(final Path file, final Local local, final BandwidthThrottle throttle, final StreamListener listener,
                            final TransferStatus status, final ConnectionCallback callback); }
SDSDirectS3UploadFeature extends HttpUploadFeature<VersionId, MessageDigest> { @Override public VersionId upload(final Path file, final Local local, final BandwidthThrottle throttle, final StreamListener listener, final TransferStatus status, final ConnectionCallback callback) throws BackgroundException { final ThreadPool pool = ThreadPoolFactory.get("multipart", concurrency); try { final CreateFileUploadRequest createFileUploadRequest = new CreateFileUploadRequest() .directS3Upload(true) .size(-1 == status.getLength() ? null : status.getLength()) .parentId(Long.parseLong(nodeid.getFileid(file.getParent(), new DisabledListProgressListener()))) .name(file.getName()); final CreateFileUploadResponse createFileUploadResponse = new NodesApi(session.getClient()) .createFileUploadChannel(createFileUploadRequest, StringUtils.EMPTY); if(log.isDebugEnabled()) { log.debug(String.format("upload started for %s with response %s", file, createFileUploadResponse)); } final Map<Integer, TransferStatus> etags = new HashMap<>(); final List<PresignedUrl> presignedUrls = this.retrievePresignedUrls(createFileUploadResponse, status); final List<Future<TransferStatus>> parts = new ArrayList<>(); final Local source; final Buffer buffer; if(nodeid.isEncrypted(file)) { source = TemporaryFileServiceFactory.get().create(new AlphanumericRandomStringService().random()); buffer = new FileBuffer(source); final BufferOutputStream temporary = new BufferOutputStream(buffer); if(log.isDebugEnabled()) { log.debug(String.format("Pre-compute file key tag for upload to S3 for %s", file)); } final InputStream in = local.getInputStream(); final OutputStream out; try { final ObjectReader reader = session.getClient().getJSON().getContext(null).readerFor(FileKey.class); final FileKey fileKey = reader.readValue(status.getFilekey().array()); out = new TripleCryptOutputStream<>(session, new StatusOutputStream<TransferStatus>(temporary) { @Override public TransferStatus getStatus() { return status; } }, Crypto.createFileEncryptionCipher(TripleCryptConverter.toCryptoPlainFileKey(fileKey)), status); } catch(CryptoSystemException | InvalidFileKeyException e) { throw new TripleCryptExceptionMappingService().map("Upload {0} failed", e, file); } new StreamCopier(status, new TransferStatus()).transfer(in, out); } else { source = local; buffer = Buffer.NULL; } final long size = status.getLength() + status.getOffset(); long offset = 0; long remaining = status.getLength(); for(int partNumber = 1; remaining > 0; partNumber++) { final long length = Math.min(Math.max((size / (MAXIMUM_UPLOAD_PARTS - 1)), partsize), remaining); final PresignedUrl presignedUrl = presignedUrls.get(partNumber - 1); parts.add(this.submit(pool, file, source, throttle, listener, status, presignedUrl.getUrl(), presignedUrl.getPartNumber(), offset, length, callback)); remaining -= length; offset += length; } for(Future<TransferStatus> future : parts) { try { final TransferStatus part = future.get(); etags.put(part.getPart(), part); } catch(InterruptedException e) { log.error("Part upload failed with interrupt failure"); status.setCanceled(); throw new ConnectionCanceledException(e); } catch(ExecutionException e) { log.warn(String.format("Part upload failed with execution failure %s", e.getMessage())); if(e.getCause() instanceof BackgroundException) { throw (BackgroundException) e.getCause(); } throw new BackgroundException(e.getCause()); } } if(nodeid.isEncrypted(file)) { buffer.close(); } final CompleteS3FileUploadRequest completeS3FileUploadRequest = new CompleteS3FileUploadRequest() .keepShareLinks(status.isExists() ? PreferencesFactory.get().getBoolean("sds.upload.sharelinks.keep") : false) .resolutionStrategy(status.isExists() ? CompleteS3FileUploadRequest.ResolutionStrategyEnum.OVERWRITE : CompleteS3FileUploadRequest.ResolutionStrategyEnum.FAIL); if(status.getFilekey() != null) { final ObjectReader reader = session.getClient().getJSON().getContext(null).readerFor(FileKey.class); final FileKey fileKey = reader.readValue(status.getFilekey().array()); final EncryptedFileKey encryptFileKey = Crypto.encryptFileKey( TripleCryptConverter.toCryptoPlainFileKey(fileKey), TripleCryptConverter.toCryptoUserPublicKey(session.keyPair().getPublicKeyContainer()) ); completeS3FileUploadRequest.setFileKey(TripleCryptConverter.toSwaggerFileKey(encryptFileKey)); } etags.forEach((key, value) -> completeS3FileUploadRequest.addPartsItem( new S3FileUploadPart().partEtag(value.getChecksum().hash).partNumber(key))); if(log.isDebugEnabled()) { log.debug(String.format("Complete file upload with %s for %s", completeS3FileUploadRequest, file)); } new NodesApi(session.getClient()).completeS3FileUpload(completeS3FileUploadRequest, createFileUploadResponse.getUploadId(), StringUtils.EMPTY); final ScheduledThreadPool polling = new ScheduledThreadPool(); final CountDownLatch done = new CountDownLatch(1); final AtomicReference<BackgroundException> failure = new AtomicReference<>(); final ScheduledFuture f = polling.repeat(new Runnable() { @Override public void run() { try { final S3FileUploadStatus uploadStatus = new NodesApi(session.getClient()) .requestUploadStatusFiles(createFileUploadResponse.getUploadId(), StringUtils.EMPTY); switch(uploadStatus.getStatus()) { case "finishing": break; case "transfer": failure.set(new InteroperabilityException(uploadStatus.getStatus())); done.countDown(); break; case "error": failure.set(new InteroperabilityException(uploadStatus.getErrorDetails().getMessage())); done.countDown(); break; case "done": status.setVersion(new VersionId(String.valueOf(uploadStatus.getNode().getId()))); done.countDown(); break; } } catch(ApiException e) { done.countDown(); failure.set(new SDSExceptionMappingService().map("Upload {0} failed", e, file)); } } }, PreferencesFactory.get().getLong("sds.upload.s3.status.period"), TimeUnit.MILLISECONDS); Uninterruptibles.awaitUninterruptibly(done); polling.shutdown(); if(null != failure.get()) { throw failure.get(); } status.setComplete(); return status.getVersion(); } catch(CryptoSystemException | InvalidFileKeyException | InvalidKeyPairException e) { throw new TripleCryptExceptionMappingService().map("Upload {0} failed", e, file); } catch(ApiException e) { throw new SDSExceptionMappingService().map("Upload {0} failed", e, file); } catch(IOException e) { throw new DefaultIOExceptionMappingService().map(e); } finally { pool.shutdown(false); } } SDSDirectS3UploadFeature(final SDSSession session, final SDSNodeIdProvider nodeid, final Write<VersionId> writer);  SDSDirectS3UploadFeature(final SDSSession session, final SDSNodeIdProvider nodeid, final Write<VersionId> writer, final Long partsize, final Integer concurrency); @Override VersionId upload(final Path file, final Local local, final BandwidthThrottle throttle, final StreamListener listener,
                            final TransferStatus status, final ConnectionCallback callback); static final int MAXIMUM_UPLOAD_PARTS; }