@Override
    public AttributedList<Path> list(final Path directory, final ListProgressListener listener) throws BackgroundException {
        final ThreadPool pool = ThreadPoolFactory.get("list", concurrency);
        try {
            final String prefix = this.createPrefix(directory);
            if(log.isDebugEnabled()) {
                log.debug(String.format("List with prefix %s", prefix));
            }
            final Path bucket = containerService.getContainer(directory);
            final AttributedList<Path> objects = new AttributedList<>();
            String priorLastKey = null;
            String priorLastVersionId = null;
            long revision = 0L;
            String lastKey = null;
            boolean hasDirectoryPlaceholder = bucket.isRoot() || containerService.isContainer(directory);
            do {
                final VersionOrDeleteMarkersChunk chunk = session.getClient().listVersionedObjectsChunked(
                        bucket.isRoot() ? StringUtils.EMPTY : bucket.getName(), prefix, String.valueOf(Path.DELIMITER),
                        new HostPreferences(session.getHost()).getInteger("s3.listing.chunksize"),
                        priorLastKey, priorLastVersionId, false);
                // Amazon S3 returns object versions in the order in which they were stored, with the most recently stored returned first.
                for(BaseVersionOrDeleteMarker marker : chunk.getItems()) {
                    final String key = URIEncoder.decode(marker.getKey());
                    if(new SimplePathPredicate(PathNormalizer.compose(bucket, key)).test(directory)) {
                        if(log.isDebugEnabled()) {
                            log.debug(String.format("Skip placeholder key %s", key));
                        }
                        hasDirectoryPlaceholder = true;
                        continue;
                    }
                    final PathAttributes attr = new PathAttributes();
                    attr.setVersionId(marker.getVersionId());
                    if(!StringUtils.equals(lastKey, key)) {
                        // Reset revision for next file
                        revision = 0L;
                    }
                    attr.setRevision(++revision);
                    attr.setDuplicate(marker.isDeleteMarker() && marker.isLatest() || !marker.isLatest());
                    if(marker.isDeleteMarker()) {
                        attr.setCustom(Collections.singletonMap(KEY_DELETE_MARKER, String.valueOf(true)));
                    }
                    attr.setModificationDate(marker.getLastModified().getTime());
                    attr.setRegion(bucket.attributes().getRegion());
                    if(marker instanceof S3Version) {
                        final S3Version object = (S3Version) marker;
                        attr.setSize(object.getSize());
                        if(StringUtils.isNotBlank(object.getEtag())) {
                            attr.setETag(StringUtils.remove(object.getEtag(), "\""));
                            // The ETag will only be the MD5 of the object data when the object is stored as plaintext or encrypted
                            // using SSE-S3. If the object is encrypted using another method (such as SSE-C or SSE-KMS) the ETag is
                            // not the MD5 of the object data.
                            attr.setChecksum(Checksum.parse(StringUtils.remove(object.getEtag(), "\"")));
                        }
                        if(StringUtils.isNotBlank(object.getStorageClass())) {
                            attr.setStorageClass(object.getStorageClass());
                        }
                    }
                    final Path f = new Path(directory.isDirectory() ? directory : directory.getParent(),
                            PathNormalizer.name(key), EnumSet.of(Path.Type.file), attr);
                    if(metadata) {
                        f.withAttributes(attributes.find(f));
                    }
                    objects.add(f);
                    lastKey = key;
                }
                final String[] prefixes = chunk.getCommonPrefixes();
                final List<Future<Path>> folders = new ArrayList<>();
                for(String common : prefixes) {
                    if(new SimplePathPredicate(PathNormalizer.compose(bucket, URIEncoder.decode(common))).test(directory)) {
                        continue;
                    }
                    folders.add(this.submit(pool, bucket, directory, URIEncoder.decode(common)));
                }
                for(Future<Path> f : folders) {
                    try {
                        objects.add(Uninterruptibles.getUninterruptibly(f));
                    }
                    catch(ExecutionException e) {
                        log.warn(String.format("Listing versioned objects failed with execution failure %s", e.getMessage()));
                        for(Throwable cause : ExceptionUtils.getThrowableList(e)) {
                            Throwables.throwIfInstanceOf(cause, BackgroundException.class);
                        }
                        throw new DefaultExceptionMappingService().map(Throwables.getRootCause(e));
                    }
                }
                priorLastKey = null != chunk.getNextKeyMarker() ? URIEncoder.decode(chunk.getNextKeyMarker()) : null;
                priorLastVersionId = chunk.getNextVersionIdMarker();
                listener.chunk(directory, objects);
            }
            while(priorLastKey != null);
            if(!hasDirectoryPlaceholder && objects.isEmpty()) {
                // Only for AWS
                if(S3Session.isAwsHostname(session.getHost().getHostname())) {
                    if(StringUtils.isEmpty(RequestEntityRestStorageService.findBucketInHostname(session.getHost()))) {
                        if(log.isWarnEnabled()) {
                            log.warn(String.format("No placeholder found for directory %s", directory));
                        }
                        throw new NotfoundException(directory.getAbsolute());
                    }
                }
                else {
                    // Handle missing prefix for directory placeholders in Minio
                    final VersionOrDeleteMarkersChunk chunk = session.getClient().listVersionedObjectsChunked(
                            bucket.isRoot() ? StringUtils.EMPTY : bucket.getName(),
                            String.format("%s%s", this.createPrefix(directory.getParent()), directory.getName()),
                            String.valueOf(Path.DELIMITER), 1, null, null, false);
                    if(Arrays.stream(chunk.getCommonPrefixes()).map(URIEncoder::decode).noneMatch(common -> common.equals(prefix))) {
                        throw new NotfoundException(directory.getAbsolute());
                    }
                }
            }
            return objects;
        }
        catch(ServiceException e) {
            throw new S3ExceptionMappingService().map("Listing directory {0} failed", e, directory);
        }
        finally {
            // Cancel future tasks
            pool.shutdown(false);
        }
    }
@Test
    public void testDirectory() throws Exception {
        final Path bucket = new Path("test-eu-central-1-cyberduck", EnumSet.of(Path.Type.directory, Path.Type.volume));
        final S3AccessControlListFeature acl = new S3AccessControlListFeature(session);
        final Path directory = new S3DirectoryFeature(session, new S3WriteFeature(session, acl), acl).mkdir(new Path(bucket, new AlphanumericRandomStringService().random(), EnumSet.of(Path.Type.directory)), new TransferStatus());
        final S3VersionedObjectListService feature = new S3VersionedObjectListService(session, acl);
        assertTrue(feature.list(bucket, new DisabledListProgressListener()).contains(directory));
        assertTrue(feature.list(directory, new DisabledListProgressListener()).isEmpty());
        new S3DefaultDeleteFeature(session).delete(Collections.singletonList(directory), new DisabledLoginCallback(), new Delete.DisabledCallback());
        assertFalse(feature.list(bucket, new DisabledListProgressListener()).contains(directory));
        try {
            feature.list(directory, new DisabledListProgressListener());
            fail();
        }
        catch(NotfoundException e) {

        }
    }