Implementation of reactive file write from InputStream

Question

I want to implement reactive write to file. In low level, it uses NIO and Futures. After reading this section I created this code but maybe there's more clean or less code solution.

Thanks.

public class LocalImageService implements IImageService {
 private String imageDir = "/tmp";
@Override
public Mono<String> storeImage(@NonNull InputStream imageStream, @NonNull String ext) {
 val processor = new ImageProcessor();
 processor.fileName = UUID.randomUUID() + "." + ext;
 val targetFile = new File(imageDir, processor.fileName);
 final Flux<CompletableFuture<Integer>> bridge = Flux.push(sink -> {
 try {
 processor.sink = sink;
 processor.inputChannel = Channels.newChannel(imageStream);
 processor.fileChannel = AsynchronousFileChannel.open(targetFile.toPath(),
 StandardOpenOption.CREATE, StandardOpenOption.WRITE);
 processor.processNextChunk();
 } catch (IOException e) {
 processor.onError(new RuntimeException("Can not write image file: " + processor.fileName, e));
 }
 });
 return bridge.flatMap(Mono::fromFuture)
 .reduce(0, (acc, num) -> {
 processor.buffer.compact();
 processor.processNextChunk();
 return acc + num;
 })
 .then(Mono.just(targetFile.getPath()));
}
private static final class ImageProcessor {
 int position = 0;
 ReadableByteChannel inputChannel;
 AsynchronousFileChannel fileChannel;
 FluxSink<CompletableFuture<Integer>> sink;
 final ByteBuffer buffer = ByteBuffer.allocate(4096);
 String fileName;
 void onComplete() {
 closeStreams();
 if (null != sink) {
 sink.complete();
 }
 }
 private void closeStreams() {
 IOUtils.closeQuietly(inputChannel);
 IOUtils.closeQuietly(fileChannel);
 }
 void onError(Throwable e) {
 closeStreams();
 if (null != sink) {
 sink.error(e);
 }
 }
 void processNextChunk() {
 try {
 int receivedBytes;
 if ((receivedBytes = inputChannel.read(buffer)) >= 0 || buffer.position() > 0) {
 buffer.flip();
 val lastWrite = fileChannel.write(buffer, position);
 position += receivedBytes;
 if (lastWrite != null) {
 sink.next(CompletableFuture.supplyAsync(() -> {
 try {
 return lastWrite.get();
 } catch (InterruptedException | ExecutionException e) {
 onError(new RuntimeException("Can not write image file: " + fileName, e));
 return -1;
 }
 }));
 }
 } else {
 onComplete();
 }
 } catch (IOException e) {
 onError(new RuntimeException("Can not write image file: " + fileName, e));
 }
 }
}
}

• Archmede: would this help? projectreactor.io/docs/core/release/reference/…