@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<>();
        if(status.isAppend()) {
            // Get a lexicographically ordered list of the existing file segments
            try {
                existingSegments.addAll(listService.list(segmentService.getSegmentsDirectory(file), new DisabledListProgressListener()).toList());
            }
            catch(NotfoundException e) {
                // Ignore
            }
        }
        else {
            if(status.isExists()) {
                // Delete existing segments
                if(new HostPreferences(session.getHost()).getBoolean("openstack.upload.largeobject.cleanup")) {
                    // Clean up any old segments
                    new SwiftMultipleDeleteFeature(session).delete(
                            new SwiftSegmentService(session, regionService).list(file), callback, new Delete.DisabledCallback());
                }
            }
        }
        // Get the results of the uploads in the order they were submitted
        // this is important for building the manifest, and is not a problem in terms of performance
        // because we should only continue when all segments have uploaded successfully
        final List<StorageObject> completed = new ArrayList<>();
        // Submit file segments for concurrent upload
        final List<Future<StorageObject>> segments = new ArrayList<>();
        long remaining = status.getLength();
        long offset = 0;
        for(int segmentNumber = 1; remaining > 0; segmentNumber++) {
            final long length = Math.min(segmentSize, remaining);
            // Segment name with left padded segment number
            final Path segment = segmentService.getSegment(file, 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 {
                // Submit to queue
                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 {
            completed.addAll(Interruptibles.awaitAll(segments));
        }
        finally {
            pool.shutdown(false);
        }
        if(log.isInfoEnabled()) {
            log.info(String.format("Finished large file upload %s with %d parts", file, completed.size()));
        }
        // Create and upload the large object manifest. It is best to upload all the segments first and
        // then create or update the manifest.
        try {
            // Static Large Object.
            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));
            stored.setSize(status.getOffset() + status.getLength());
            final String checksum = session.getClient().createSLOManifestObject(regionService.lookup(
                            containerService.getContainer(file)),
                    containerService.getContainer(file).getName(),
                    status.getMime(),
                    containerService.getKey(file), manifest, Collections.emptyMap());
            // The value of the Content-Length header is the total size of all segment objects, and the value of the ETag header is calculated by taking
            // the ETag value of each segment, concatenating them together, and then returning the MD5 checksum of the result.
            stored.setMd5sum(checksum);
            // Mark parent status as complete
            status.withResponse(new SwiftAttributesFinderFeature(session).toAttributes(stored)).setComplete();
            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);
        }
    }
@Test
    public void testUploadOverwrite() 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, new AlphanumericRandomStringService().random() + ".txt", EnumSet.of(Path.Type.file));
        final Local local = new Local(System.getProperty("java.io.tmpdir"), new AlphanumericRandomStringService().random());

        // Each segment, except the last, must be larger than 1048576 bytes.
        // 2MB + 1
        {
            final int length = 1048576 + 1048576 + 1;
            final byte[] content = RandomUtils.nextBytes(length);

            final OutputStream out = local.getOutputStream(false);
            IOUtils.write(content, out);
            out.close();
            final TransferStatus status = new TransferStatus();
            status.setLength(content.length);

            final SwiftRegionService regionService = new SwiftRegionService(session);
            final SwiftLargeObjectUploadFeature upload = new SwiftLargeObjectUploadFeature(session,
                    regionService,
                    new SwiftObjectListService(session, regionService),
                    new SwiftSegmentService(session, ".segments-test/"),
                    new SwiftWriteFeature(session, regionService), (long) (content.length / 2), 4);

            final BytecountStreamListener count = new BytecountStreamListener();
            final StorageObject object = upload.upload(test, local, new BandwidthThrottle(BandwidthThrottle.UNLIMITED), count,
                    status, new DisabledConnectionCallback());
            assertEquals(Checksum.NONE, Checksum.parse(object.getMd5sum()));
            assertNotEquals(Checksum.NONE, new SwiftAttributesFinderFeature(session).find(test).getChecksum());
            assertNotNull(new DefaultAttributesFinderFeature(session).find(test).getChecksum().hash);

            assertTrue(status.isComplete());
            assertNotSame(PathAttributes.EMPTY, status.getResponse());
            assertEquals(content.length, status.getResponse().getSize());

            // Verify not canceled
            status.validate();
            assertEquals(content.length, count.getSent());

            assertTrue(new SwiftFindFeature(session).find(test));
            final InputStream in = new SwiftReadFeature(session, regionService).read(test, new TransferStatus(), new DisabledConnectionCallback());
            final ByteArrayOutputStream buffer = new ByteArrayOutputStream(content.length);
            new StreamCopier(status, status).transfer(in, buffer);
            in.close();
            buffer.close();
            assertArrayEquals(content, buffer.toByteArray());

            final Map<String, String> metadata = new SwiftMetadataFeature(session).getMetadata(test);
            assertFalse(metadata.isEmpty());
            assertEquals("text/plain", metadata.get("Content-Type"));

            final List<Path> segments = new SwiftSegmentService(session).list(test);
            assertFalse(segments.isEmpty());
            assertEquals(3, segments.size());
            assertEquals(1048576L, segments.get(0).attributes().getSize());
            assertEquals(1048576L, segments.get(1).attributes().getSize());
            assertEquals(1L, segments.get(2).attributes().getSize());
        }
        // Overwrite
        {
            final int length = 1048576 + 1;
            final byte[] content = RandomUtils.nextBytes(length);

            final OutputStream out = local.getOutputStream(false);
            IOUtils.write(content, out);
            out.close();
            final TransferStatus status = new TransferStatus();
            status.setExists(true);
            status.setLength(content.length);

            final SwiftRegionService regionService = new SwiftRegionService(session);
            final SwiftLargeObjectUploadFeature upload = new SwiftLargeObjectUploadFeature(session,
                    regionService,
                    new SwiftObjectListService(session, regionService),
                    new SwiftSegmentService(session, ".segments-test/"),
                    new SwiftWriteFeature(session, regionService), 1048576L, 4);

            final BytecountStreamListener count = new BytecountStreamListener();
            upload.upload(test, local, new BandwidthThrottle(BandwidthThrottle.UNLIMITED), count,
                    status, new DisabledConnectionCallback());

            assertTrue(status.isComplete());
            assertEquals(content.length, status.getResponse().getSize());

            // Verify not canceled
            status.validate();
            assertEquals(content.length, count.getSent());

            final List<Path> segments = new SwiftSegmentService(session).list(test);
            assertFalse(segments.isEmpty());
            assertEquals(2, segments.size());
            assertEquals(1048576L, segments.get(0).attributes().getSize());
            assertEquals(1L, segments.get(1).attributes().getSize());

            assertTrue(new SwiftFindFeature(session).find(test));
            final InputStream in = new SwiftReadFeature(session, regionService).read(test, new TransferStatus(), new DisabledConnectionCallback());
            final ByteArrayOutputStream buffer = new ByteArrayOutputStream(content.length);
            new StreamCopier(status, status).transfer(in, buffer);
            in.close();
            buffer.close();
            assertArrayEquals(content, buffer.toByteArray());
        }

        new SwiftDeleteFeature(session).delete(Collections.singletonList(test), new DisabledLoginCallback(), new Delete.DisabledCallback());
        assertEquals(0, new SwiftSegmentService(session).list(test).size());
        local.delete();
    }