@Override
    public HttpResponseOutputStream<Node> write(final Path file, final TransferStatus status, final ConnectionCallback callback) throws BackgroundException {
        try {
            final CreateFileUploadRequest createFileUploadRequest = new CreateFileUploadRequest()
                    .directS3Upload(true)
                    .timestampModification(status.getModified() != null ? new DateTime(status.getModified()) : null)
                    .timestampCreation(status.getCreated() != null ? new DateTime(status.getCreated()) : null)
                    .parentId(Long.parseLong(nodeid.getVersionId(file.getParent())))
                    .name(file.getName());
            final CreateFileUploadResponse createFileUploadResponse = new NodesApi(session.getClient())
                    .createFileUploadChannel(createFileUploadRequest, StringUtils.EMPTY);
            if(log.isDebugEnabled()) {
                log.debug(String.format("upload started for %s with response %s", file, createFileUploadResponse));
            }
            final MultipartOutputStream proxy = new MultipartOutputStream(createFileUploadResponse, file, status);
            return new HttpResponseOutputStream<Node>(new MemorySegementingOutputStream(proxy, partsize),
                    new SDSAttributesAdapter(session), status) {
                @Override
                public Node getStatus() {
                    return proxy.getResult();
                }
            };
        }
        catch(ApiException e) {
            throw new SDSExceptionMappingService(nodeid).map("Upload {0} failed", e, file);
        }

    }
@Test
    public void testWriteZeroLength() throws Exception {
        final SDSNodeIdProvider nodeid = new SDSNodeIdProvider(session);
        final Path room = new SDSDirectoryFeature(session, nodeid).mkdir(
                new Path(new AlphanumericRandomStringService().random(), EnumSet.of(Path.Type.directory, Path.Type.volume)), new TransferStatus());
        final TransferStatus status = new TransferStatus();
        final Path test = new Path(room, new AlphanumericRandomStringService().random(), EnumSet.of(Path.Type.file));
        final SDSDirectS3MultipartWriteFeature writer = new SDSDirectS3MultipartWriteFeature(session, nodeid);
        final StatusOutputStream<Node> out = writer.write(test, status, new DisabledConnectionCallback());
        assertNotNull(out);
        new StreamCopier(status, status).transfer(new NullInputStream(0L), out);
        assertEquals(0L, out.getStatus().getSize(), 0L);
        assertNotNull(test.attributes().getVersionId());
        assertTrue(new DefaultFindFeature(session).find(test));
        new SDSDeleteFeature(session, nodeid).delete(Collections.singletonList(room), new DisabledLoginCallback(), new Delete.DisabledCallback());
    }