Turns a `Flux<DataBuffer>` into an `InputStream`

DataBuffersInputStream.java

class DataBuffersInputStream extends InputStream {
    volatile DataBuffer currentBuf;
    volatile boolean complete;

    BaseSubscriber<DataBuffer> subscriber = new BaseSubscriber<>() {
        @Override
        protected void hookOnNext(DataBuffer buf) {
            releaseCurrentBuf();
            currentBuf = buf;
            ((NettyDataBuffer) buf).retain();
            synchronized (DataBuffersInputStream.this) {
                DataBuffersInputStream.this.notify();
            }
        }

        @Override
        protected void hookFinally(SignalType type) {
            releaseCurrentBuf();
            complete = true;
            synchronized (DataBuffersInputStream.this) {
                DataBuffersInputStream.this.notify();
            }
        }
    };

    DataBuffersInputStream(Flux<DataBuffer> dataBuffers) {
        dataBuffers.subscribe(subscriber);
    }

    @Override
    public int read() throws IOException {
        while (!complete) {
            if (currentBuf == null || currentBuf.readableByteCount() == 0) {
                try {
                    synchronized (this) {
                        subscriber.request(1);
                        wait();
                    }
                } catch (InterruptedException e) {
                    throw new IOException(e);
                }
            } else {
                return currentBuf.read();
            }
        }
        return -1;
    }

    @Override
    public void close() {
        releaseCurrentBuf();
        subscriber.cancel();
    }

    private void releaseCurrentBuf() {
        if (currentBuf != null) {
            DataBufferUtils.release(currentBuf);
            currentBuf = null;
        }
    }
}