FileTransporter { Single<FileMetadata> upload(UUID id, String name, File file, MimeType mimeType) { return rxJavaVertx.fileSystem().openObservable(file.getAbsolutePath(), new OpenOptions().setRead(true)) .flatMap(AsyncFile::toObservable) .reduce(Buffer.buffer(), (reduced, buffer) -> { return reduced.appendBuffer(buffer); }) .toSingle() .flatMap(reducedBuffer -> { return observableFileSystem.store(id, reducedBuffer); }).flatMap(v -> { FileMetadata metadata = new FileMetadata(id, name, mimeType, file.length(), Instant.now()); return observableFileMetadataRegistry.store(metadata); }); }    }
@Test public void testUpload(TestContext context) throws IOException { Async async = context.async(); SharedFileSystem fileSystem = Mockito.mock(SharedFileSystem.class); FileMetadataRegistry fileMetadataRegistry = Mockito.mock(FileMetadataRegistry.class); FileTransporter transporter = new FileTransporter(vertx, fileSystem, fileMetadataRegistry); UUID id = UUID.randomUUID(); File file = folder.newFile(); doReturn(Single.just((Void) null)).when(fileSystem).store(eq(id), any()); doReturn(Single.just((Void) null)).when(fileMetadataRegistry).store(any()); transporter.upload(id, "name", file, MimeType.valueOf("text/plain")).subscribe(v -> { Mockito.verify(fileSystem).store(eq(id), any()); Mockito.verify(fileMetadataRegistry).store(any()); async.complete(); }); }