@Override
    public StorageObject upload(final Path file, final Local local, final BandwidthThrottle throttle,
                                final StreamListener listener, final TransferStatus status, final ConnectionCallback callback) throws BackgroundException {
        if(Checksum.NONE == status.getChecksum()) {
            // The client-side checksum is passed to the BlackPearl gateway by supplying the applicable CRC HTTP header.
            // If this is done, the BlackPearl gateway verifies that the data received matches the checksum provided.
            // End-to-end data protection requires that the client provide the CRC when uploading the object and then
            // verify the CRC after downloading the object at a later time (see Get Object). The BlackPearl gateway also
            // verifies the CRC when reading from physical data stores so the gateway can identify problems before
            // transmitting data to the client.
            status.setChecksum(writer.checksum(file, status).compute(local.getInputStream(), status));
        }
        // Make sure file is available in cache
        final List<TransferStatus> chunks = bulk.query(Transfer.Type.upload, file, status);
        StorageObject stored = null;
        for(TransferStatus chunk : chunks) {
            chunk.setChecksum(ChecksumComputeFactory.get(HashAlgorithm.md5).compute(local.getInputStream(), chunk));
            stored = super.upload(file, local, throttle, listener, chunk, callback);
        }
        return stored;
    }
@Test
    public void testUpload() throws Exception {
        final Local local = new Local(System.getProperty("java.io.tmpdir"), new AlphanumericRandomStringService().random());
        final int length = 32770;
        final byte[] content = RandomUtils.nextBytes(length);
        final OutputStream out = local.getOutputStream(false);
        IOUtils.write(content, out);
        out.close();
        final Path container = new SpectraDirectoryFeature(session, new SpectraWriteFeature(session)).mkdir(
                new Path(new AlphanumericRandomStringService().random(), EnumSet.of(Path.Type.directory, Path.Type.volume)), new TransferStatus());
        final Path test = new Path(container, new AlphanumericRandomStringService().random(), EnumSet.of(Path.Type.file));
        final TransferStatus writeStatus = new TransferStatus().withLength(content.length);
        final SpectraBulkService bulk = new SpectraBulkService(session);
        bulk.pre(Transfer.Type.upload, Collections.singletonMap(new TransferItem(test), writeStatus), new DisabledConnectionCallback());
        final SpectraUploadFeature upload = new SpectraUploadFeature(session, new SpectraWriteFeature(session), new SpectraBulkService(session));
        upload.upload(test, local, new BandwidthThrottle(BandwidthThrottle.UNLIMITED), new DisabledStreamListener(),
                writeStatus, new DisabledConnectionCallback());
        final byte[] buffer = new byte[content.length];
        final TransferStatus readStatus = new TransferStatus().withLength(content.length);
        bulk.pre(Transfer.Type.download, Collections.singletonMap(new TransferItem(test), readStatus), new DisabledConnectionCallback());
        final InputStream in = new SpectraReadFeature(session).read(test, readStatus, new DisabledConnectionCallback());
        IOUtils.readFully(in, buffer);
        in.close();
        assertArrayEquals(content, buffer);
        new SpectraDeleteFeature(session).delete(Collections.singletonList(container), new DisabledLoginCallback(), new Delete.DisabledCallback());
        local.delete();
    }