@Test public void testAppendNoPartCompleted() throws Exception { final Path container = new Path("test.cyberduck.ch", EnumSet.of(Path.Type.directory, Path.Type.volume)); container.attributes().setRegion("IAD"); final Path test = new Path(container, UUID.randomUUID().toString(), EnumSet.of(Path.Type.file)); final Local local = new Local(System.getProperty("java.io.tmpdir"), UUID.randomUUID().toString()); final int length = 2 * 1024 * 1024; final byte[] content = RandomUtils.nextBytes(length); IOUtils.write(content, local.getOutputStream(false)); final TransferStatus status = new TransferStatus(); status.setLength(content.length); final AtomicBoolean interrupt = new AtomicBoolean(); try { new SwiftLargeObjectUploadFeature(session, new SwiftRegionService(session), new SwiftWriteFeature(session, new SwiftRegionService(session)), 1 * 1024L * 1024L, 1).upload(test, local, new BandwidthThrottle(BandwidthThrottle.UNLIMITED), new DisabledStreamListener() { long count; @Override public void sent(final long bytes) { count += bytes; if(count >= 32768L) { throw new RuntimeException(); } } }, status, new DisabledLoginCallback()); } catch(BackgroundException e) { interrupt.set(true); } assertTrue(interrupt.get()); assertEquals(0L, status.getOffset(), 0L); assertFalse(status.isComplete()); final TransferStatus append = new TransferStatus().append(true).length(content.length); new SwiftLargeObjectUploadFeature(session, new SwiftRegionService(session), new SwiftWriteFeature(session, new SwiftRegionService(session)), 1 * 1024L * 1024L, 1).upload(test, local, new BandwidthThrottle(BandwidthThrottle.UNLIMITED), new DisabledStreamListener(), append, new DisabledLoginCallback()); assertTrue(new SwiftFindFeature(session).find(test)); assertEquals(content.length, new SwiftAttributesFinderFeature(session).find(test).getSize()); assertEquals(content.length, append.getOffset(), 0L); assertTrue(append.isComplete()); final byte[] buffer = new byte[content.length]; final InputStream in = new SwiftReadFeature(session, new SwiftRegionService(session)).read(test, new TransferStatus(), new DisabledConnectionCallback()); IOUtils.readFully(in, buffer); in.close(); assertArrayEquals(content, buffer); new SwiftDeleteFeature(session).delete(Collections.singletonList(test), new DisabledLoginCallback(), new Delete.DisabledCallback()); local.delete(); }
@Override public StorageObject 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); final List<Path> existingSegments = new ArrayList<Path>(); if(status.isAppend()) { try { existingSegments.addAll(listService.list(segmentService.getSegmentsDirectory(file, status.getOffset() + status.getLength()), new DisabledListProgressListener()).toList()); } catch(NotfoundException e) { } } final List<StorageObject> completed = new ArrayList<StorageObject>(); final List<Future<StorageObject>> segments = new ArrayList<Future<StorageObject>>(); long remaining = status.getLength(); long offset = 0; for(int segmentNumber = 1; remaining > 0; segmentNumber++) { final Long length = Math.min(segmentSize, remaining); final Path segment = segmentService.getSegment(file, status.getOffset() + status.getLength(), segmentNumber); if(existingSegments.contains(segment)) { final Path existingSegment = existingSegments.get(existingSegments.indexOf(segment)); if(log.isDebugEnabled()) { log.debug(String.format("Skip segment %s", existingSegment)); } final StorageObject stored = new StorageObject(containerService.getKey(segment)); if(HashAlgorithm.md5.equals(existingSegment.attributes().getChecksum().algorithm)) { stored.setMd5sum(existingSegment.attributes().getChecksum().hash); } stored.setSize(existingSegment.attributes().getSize()); offset += existingSegment.attributes().getSize(); completed.add(stored); } else { segments.add(this.submit(pool, segment, local, throttle, listener, status, offset, length, callback)); if(log.isDebugEnabled()) { log.debug(String.format("Segment %s submitted with size %d and offset %d", segment, length, offset)); } remaining -= length; offset += length; } } try { for(Future<StorageObject> futureSegment : segments) { completed.add(futureSegment.get()); } } 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 DefaultExceptionMappingService().map(e.getCause()); } finally { pool.shutdown(false); } status.setComplete(); if(log.isInfoEnabled()) { log.info(String.format("Finished large file upload %s with %d parts", file, completed.size())); } try { final String manifest = segmentService.manifest(containerService.getContainer(file).getName(), completed); if(log.isDebugEnabled()) { log.debug(String.format("Creating SLO manifest %s for %s", manifest, file)); } final StorageObject stored = new StorageObject(containerService.getKey(file)); final String checksum = session.getClient().createSLOManifestObject(regionService.lookup( containerService.getContainer(file)), containerService.getContainer(file).getName(), status.getMime(), containerService.getKey(file), manifest, Collections.emptyMap()); stored.setMd5sum(checksum); return stored; } catch(GenericException e) { throw new SwiftExceptionMappingService().map("Upload {0} failed", e, file); } catch(IOException e) { throw new DefaultIOExceptionMappingService().map("Upload {0} failed", e, file); } }
SwiftLargeObjectUploadFeature extends HttpUploadFeature<StorageObject, MessageDigest> { @Override public StorageObject 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); final List<Path> existingSegments = new ArrayList<Path>(); if(status.isAppend()) { try { existingSegments.addAll(listService.list(segmentService.getSegmentsDirectory(file, status.getOffset() + status.getLength()), new DisabledListProgressListener()).toList()); } catch(NotfoundException e) { } } final List<StorageObject> completed = new ArrayList<StorageObject>(); final List<Future<StorageObject>> segments = new ArrayList<Future<StorageObject>>(); long remaining = status.getLength(); long offset = 0; for(int segmentNumber = 1; remaining > 0; segmentNumber++) { final Long length = Math.min(segmentSize, remaining); final Path segment = segmentService.getSegment(file, status.getOffset() + status.getLength(), segmentNumber); if(existingSegments.contains(segment)) { final Path existingSegment = existingSegments.get(existingSegments.indexOf(segment)); if(log.isDebugEnabled()) { log.debug(String.format("Skip segment %s", existingSegment)); } final StorageObject stored = new StorageObject(containerService.getKey(segment)); if(HashAlgorithm.md5.equals(existingSegment.attributes().getChecksum().algorithm)) { stored.setMd5sum(existingSegment.attributes().getChecksum().hash); } stored.setSize(existingSegment.attributes().getSize()); offset += existingSegment.attributes().getSize(); completed.add(stored); } else { segments.add(this.submit(pool, segment, local, throttle, listener, status, offset, length, callback)); if(log.isDebugEnabled()) { log.debug(String.format("Segment %s submitted with size %d and offset %d", segment, length, offset)); } remaining -= length; offset += length; } } try { for(Future<StorageObject> futureSegment : segments) { completed.add(futureSegment.get()); } } 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 DefaultExceptionMappingService().map(e.getCause()); } finally { pool.shutdown(false); } status.setComplete(); if(log.isInfoEnabled()) { log.info(String.format("Finished large file upload %s with %d parts", file, completed.size())); } try { final String manifest = segmentService.manifest(containerService.getContainer(file).getName(), completed); if(log.isDebugEnabled()) { log.debug(String.format("Creating SLO manifest %s for %s", manifest, file)); } final StorageObject stored = new StorageObject(containerService.getKey(file)); final String checksum = session.getClient().createSLOManifestObject(regionService.lookup( containerService.getContainer(file)), containerService.getContainer(file).getName(), status.getMime(), containerService.getKey(file), manifest, Collections.emptyMap()); stored.setMd5sum(checksum); return stored; } catch(GenericException e) { throw new SwiftExceptionMappingService().map("Upload {0} failed", e, file); } catch(IOException e) { throw new DefaultIOExceptionMappingService().map("Upload {0} failed", e, file); } } }
SwiftLargeObjectUploadFeature extends HttpUploadFeature<StorageObject, MessageDigest> { @Override public StorageObject 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); final List<Path> existingSegments = new ArrayList<Path>(); if(status.isAppend()) { try { existingSegments.addAll(listService.list(segmentService.getSegmentsDirectory(file, status.getOffset() + status.getLength()), new DisabledListProgressListener()).toList()); } catch(NotfoundException e) { } } final List<StorageObject> completed = new ArrayList<StorageObject>(); final List<Future<StorageObject>> segments = new ArrayList<Future<StorageObject>>(); long remaining = status.getLength(); long offset = 0; for(int segmentNumber = 1; remaining > 0; segmentNumber++) { final Long length = Math.min(segmentSize, remaining); final Path segment = segmentService.getSegment(file, status.getOffset() + status.getLength(), segmentNumber); if(existingSegments.contains(segment)) { final Path existingSegment = existingSegments.get(existingSegments.indexOf(segment)); if(log.isDebugEnabled()) { log.debug(String.format("Skip segment %s", existingSegment)); } final StorageObject stored = new StorageObject(containerService.getKey(segment)); if(HashAlgorithm.md5.equals(existingSegment.attributes().getChecksum().algorithm)) { stored.setMd5sum(existingSegment.attributes().getChecksum().hash); } stored.setSize(existingSegment.attributes().getSize()); offset += existingSegment.attributes().getSize(); completed.add(stored); } else { segments.add(this.submit(pool, segment, local, throttle, listener, status, offset, length, callback)); if(log.isDebugEnabled()) { log.debug(String.format("Segment %s submitted with size %d and offset %d", segment, length, offset)); } remaining -= length; offset += length; } } try { for(Future<StorageObject> futureSegment : segments) { completed.add(futureSegment.get()); } } 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 DefaultExceptionMappingService().map(e.getCause()); } finally { pool.shutdown(false); } status.setComplete(); if(log.isInfoEnabled()) { log.info(String.format("Finished large file upload %s with %d parts", file, completed.size())); } try { final String manifest = segmentService.manifest(containerService.getContainer(file).getName(), completed); if(log.isDebugEnabled()) { log.debug(String.format("Creating SLO manifest %s for %s", manifest, file)); } final StorageObject stored = new StorageObject(containerService.getKey(file)); final String checksum = session.getClient().createSLOManifestObject(regionService.lookup( containerService.getContainer(file)), containerService.getContainer(file).getName(), status.getMime(), containerService.getKey(file), manifest, Collections.emptyMap()); stored.setMd5sum(checksum); return stored; } catch(GenericException e) { throw new SwiftExceptionMappingService().map("Upload {0} failed", e, file); } catch(IOException e) { throw new DefaultIOExceptionMappingService().map("Upload {0} failed", e, file); } } SwiftLargeObjectUploadFeature(final SwiftSession session, final SwiftRegionService regionService, final Write<StorageObject> writer,
                                         final Long segmentSize, final Integer concurrency);  SwiftLargeObjectUploadFeature(final SwiftSession session,
                                         final SwiftRegionService regionService,
                                         final SwiftObjectListService listService,
                                         final SwiftSegmentService segmentService,
                                         final Write<StorageObject> writer,
                                         final Long segmentSize, final Integer concurrency); }
SwiftLargeObjectUploadFeature extends HttpUploadFeature<StorageObject, MessageDigest> { @Override public StorageObject 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); final List<Path> existingSegments = new ArrayList<Path>(); if(status.isAppend()) { try { existingSegments.addAll(listService.list(segmentService.getSegmentsDirectory(file, status.getOffset() + status.getLength()), new DisabledListProgressListener()).toList()); } catch(NotfoundException e) { } } final List<StorageObject> completed = new ArrayList<StorageObject>(); final List<Future<StorageObject>> segments = new ArrayList<Future<StorageObject>>(); long remaining = status.getLength(); long offset = 0; for(int segmentNumber = 1; remaining > 0; segmentNumber++) { final Long length = Math.min(segmentSize, remaining); final Path segment = segmentService.getSegment(file, status.getOffset() + status.getLength(), segmentNumber); if(existingSegments.contains(segment)) { final Path existingSegment = existingSegments.get(existingSegments.indexOf(segment)); if(log.isDebugEnabled()) { log.debug(String.format("Skip segment %s", existingSegment)); } final StorageObject stored = new StorageObject(containerService.getKey(segment)); if(HashAlgorithm.md5.equals(existingSegment.attributes().getChecksum().algorithm)) { stored.setMd5sum(existingSegment.attributes().getChecksum().hash); } stored.setSize(existingSegment.attributes().getSize()); offset += existingSegment.attributes().getSize(); completed.add(stored); } else { segments.add(this.submit(pool, segment, local, throttle, listener, status, offset, length, callback)); if(log.isDebugEnabled()) { log.debug(String.format("Segment %s submitted with size %d and offset %d", segment, length, offset)); } remaining -= length; offset += length; } } try { for(Future<StorageObject> futureSegment : segments) { completed.add(futureSegment.get()); } } 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 DefaultExceptionMappingService().map(e.getCause()); } finally { pool.shutdown(false); } status.setComplete(); if(log.isInfoEnabled()) { log.info(String.format("Finished large file upload %s with %d parts", file, completed.size())); } try { final String manifest = segmentService.manifest(containerService.getContainer(file).getName(), completed); if(log.isDebugEnabled()) { log.debug(String.format("Creating SLO manifest %s for %s", manifest, file)); } final StorageObject stored = new StorageObject(containerService.getKey(file)); final String checksum = session.getClient().createSLOManifestObject(regionService.lookup( containerService.getContainer(file)), containerService.getContainer(file).getName(), status.getMime(), containerService.getKey(file), manifest, Collections.emptyMap()); stored.setMd5sum(checksum); return stored; } catch(GenericException e) { throw new SwiftExceptionMappingService().map("Upload {0} failed", e, file); } catch(IOException e) { throw new DefaultIOExceptionMappingService().map("Upload {0} failed", e, file); } } SwiftLargeObjectUploadFeature(final SwiftSession session, final SwiftRegionService regionService, final Write<StorageObject> writer,
                                         final Long segmentSize, final Integer concurrency);  SwiftLargeObjectUploadFeature(final SwiftSession session,
                                         final SwiftRegionService regionService,
                                         final SwiftObjectListService listService,
                                         final SwiftSegmentService segmentService,
                                         final Write<StorageObject> writer,
                                         final Long segmentSize, final Integer concurrency); @Override StorageObject upload(final Path file, final Local local,
                                final BandwidthThrottle throttle,
                                final StreamListener listener,
                                final TransferStatus status,
                                final ConnectionCallback callback); @Override Upload<StorageObject> withWriter(final Write<StorageObject> writer); }
SwiftLargeObjectUploadFeature extends HttpUploadFeature<StorageObject, MessageDigest> { @Override public StorageObject 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); final List<Path> existingSegments = new ArrayList<Path>(); if(status.isAppend()) { try { existingSegments.addAll(listService.list(segmentService.getSegmentsDirectory(file, status.getOffset() + status.getLength()), new DisabledListProgressListener()).toList()); } catch(NotfoundException e) { } } final List<StorageObject> completed = new ArrayList<StorageObject>(); final List<Future<StorageObject>> segments = new ArrayList<Future<StorageObject>>(); long remaining = status.getLength(); long offset = 0; for(int segmentNumber = 1; remaining > 0; segmentNumber++) { final Long length = Math.min(segmentSize, remaining); final Path segment = segmentService.getSegment(file, status.getOffset() + status.getLength(), segmentNumber); if(existingSegments.contains(segment)) { final Path existingSegment = existingSegments.get(existingSegments.indexOf(segment)); if(log.isDebugEnabled()) { log.debug(String.format("Skip segment %s", existingSegment)); } final StorageObject stored = new StorageObject(containerService.getKey(segment)); if(HashAlgorithm.md5.equals(existingSegment.attributes().getChecksum().algorithm)) { stored.setMd5sum(existingSegment.attributes().getChecksum().hash); } stored.setSize(existingSegment.attributes().getSize()); offset += existingSegment.attributes().getSize(); completed.add(stored); } else { segments.add(this.submit(pool, segment, local, throttle, listener, status, offset, length, callback)); if(log.isDebugEnabled()) { log.debug(String.format("Segment %s submitted with size %d and offset %d", segment, length, offset)); } remaining -= length; offset += length; } } try { for(Future<StorageObject> futureSegment : segments) { completed.add(futureSegment.get()); } } 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 DefaultExceptionMappingService().map(e.getCause()); } finally { pool.shutdown(false); } status.setComplete(); if(log.isInfoEnabled()) { log.info(String.format("Finished large file upload %s with %d parts", file, completed.size())); } try { final String manifest = segmentService.manifest(containerService.getContainer(file).getName(), completed); if(log.isDebugEnabled()) { log.debug(String.format("Creating SLO manifest %s for %s", manifest, file)); } final StorageObject stored = new StorageObject(containerService.getKey(file)); final String checksum = session.getClient().createSLOManifestObject(regionService.lookup( containerService.getContainer(file)), containerService.getContainer(file).getName(), status.getMime(), containerService.getKey(file), manifest, Collections.emptyMap()); stored.setMd5sum(checksum); return stored; } catch(GenericException e) { throw new SwiftExceptionMappingService().map("Upload {0} failed", e, file); } catch(IOException e) { throw new DefaultIOExceptionMappingService().map("Upload {0} failed", e, file); } } SwiftLargeObjectUploadFeature(final SwiftSession session, final SwiftRegionService regionService, final Write<StorageObject> writer,
                                         final Long segmentSize, final Integer concurrency);  SwiftLargeObjectUploadFeature(final SwiftSession session,
                                         final SwiftRegionService regionService,
                                         final SwiftObjectListService listService,
                                         final SwiftSegmentService segmentService,
                                         final Write<StorageObject> writer,
                                         final Long segmentSize, final Integer concurrency); @Override StorageObject upload(final Path file, final Local local,
                                final BandwidthThrottle throttle,
                                final StreamListener listener,
                                final TransferStatus status,
                                final ConnectionCallback callback); @Override Upload<StorageObject> withWriter(final Write<StorageObject> writer);  }