class B2LargeUploadService extends HttpUploadFeature<BaseB2Response, MessageDigest> {

    @Override
    public BaseB2Response 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("largeupload", concurrency);
        try {
            // 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<B2UploadPartResponse> completed = new ArrayList<B2UploadPartResponse>();
            final Map<String, String> fileinfo = new HashMap<>(status.getMetadata());
            final Checksum checksum = status.getChecksum();
            if(Checksum.NONE != checksum) {
                switch(checksum.algorithm) {
                    case sha1:
                        fileinfo.put(X_BZ_INFO_LARGE_FILE_SHA1, status.getChecksum().hash);
                        break;
                }
            }
            if(null != status.getTimestamp()) {
                fileinfo.put(X_BZ_INFO_SRC_LAST_MODIFIED_MILLIS, String.valueOf(status.getTimestamp()));
            }
            if(status.isAppend()) {
                // Add already completed parts
                final B2LargeUploadPartService partService = new B2LargeUploadPartService(session, fileid);
                final List<B2FileInfoResponse> uploads = partService.find(file);
                if(uploads.isEmpty()) {
                    status.setVersion(new VersionId(session.getClient().startLargeFileUpload(fileid.getFileid(containerService.getContainer(file), new DisabledListProgressListener()),
                        containerService.getKey(file), status.getMime(), fileinfo).getFileId()));
                }
                else {
                    status.setVersion(new VersionId(uploads.iterator().next().getFileId()));
                    completed.addAll(partService.list(status.getVersion().id));
                }
            }
            else {
                status.setVersion(new VersionId(session.getClient().startLargeFileUpload(fileid.getFileid(containerService.getContainer(file), new DisabledListProgressListener()),
                    containerService.getKey(file), status.getMime(), fileinfo).getFileId()));
            }
            // Full size of file
            final long size = status.getLength() + status.getOffset();
            // Submit file segments for concurrent upload
            final List<Future<B2UploadPartResponse>> parts = new ArrayList<Future<B2UploadPartResponse>>();
            long remaining = status.getLength();
            long offset = 0;
            for(int partNumber = 1; remaining > 0; partNumber++) {
                boolean skip = false;
                if(status.isAppend()) {
                    if(log.isInfoEnabled()) {
                        log.info(String.format("Determine if part number %d can be skipped", partNumber));
                    }
                    for(B2UploadPartResponse c : completed) {
                        if(c.getPartNumber().equals(partNumber)) {
                            if(log.isInfoEnabled()) {
                                log.info(String.format("Skip completed part number %d", partNumber));
                            }
                            skip = true;
                            offset += c.getContentLength();
                            break;
                        }
                    }
                }
                if(!skip) {
                    final Long length = Math.min(Math.max((size / B2LargeUploadService.MAXIMUM_UPLOAD_PARTS), partSize), remaining);
                    // Submit to queue
                    parts.add(this.submit(pool, file, local, throttle, listener, status, partNumber, offset, length, callback));
                    if(log.isDebugEnabled()) {
                        log.debug(String.format("Part %s submitted with size %d and offset %d", partNumber, length, offset));
                    }
                    remaining -= length;
                    offset += length;
                }
            }
            try {
                for(Future<B2UploadPartResponse> f : parts) {
                    completed.add(f.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());
            }
            completed.sort(new Comparator<B2UploadPartResponse>() {
                @Override
                public int compare(final B2UploadPartResponse o1, final B2UploadPartResponse o2) {
                    return o1.getPartNumber().compareTo(o2.getPartNumber());
                }
            });
            final List<String> checksums = new ArrayList<String>();
            for(B2UploadPartResponse part : completed) {
                checksums.add(part.getContentSha1());
            }
            final B2FinishLargeFileResponse response = session.getClient().finishLargeFileUpload(status.getVersion().id, checksums.toArray(new String[checksums.size()]));
            if(log.isInfoEnabled()) {
                log.info(String.format("Finished large file upload %s with %d parts", file, completed.size()));
            }
            // Mark parent status as complete
            status.setComplete();
            return response;
        }
        catch(B2ApiException e) {
            throw new B2ExceptionMappingService().map("Upload {0} failed", e, file);
        }
        catch(IOException e) {
            throw new DefaultIOExceptionMappingService().map("Upload {0} failed", e, file);
        }
        finally {
            pool.shutdown(false);
        }
    }

    public  B2LargeUploadService(final B2Session session, final B2FileidProvider fileid, final Write<BaseB2Response> writer);
    public  B2LargeUploadService(final B2Session session, final B2FileidProvider fileid, final Write<BaseB2Response> writer, final Long partSize, final Integer concurrency);

}