Webflux rest api that returns a file using data from DB

My goal is to read data from db in chunks and right that data to input stream so it can be sent to client as file. The reason we want to read in chunks is to prevent reading a lot of data at once and choking server memory. What I have tried without success is below and I think the reason why it isn't working is that in the CustomFileInputStream I am fetching data into a flux which being asynchronous return immediately while the read method is being invoked by the framework synchronously/imperatively. Can someone suggest a way to get around this?

Controller Code

@GetMapping(path = "files/xyz", produces = MediaType.APPLICATION_OCTET_STREAM_VALUE)
    public Mono<ResponseEntity<Resource>> getXYZFile(@PathVariable("clientId") long clientId) throws IOException {

        CustomFileInputStream is = new CustomFileInputStream (type, clientId, startDate, endDate, props, mediaStatusRepository);
        InputStreamResource resource = new InputStreamResource(is); 
        ResponseEntity<Resource> responseEntity = ResponseEntity.ok().header("Content-Disposition", "attachment; filename=dummyFileName").body(file);
        return Mono.just(responseEntity);
    }

Code for CustomFileInputStream

public class CustomFileInputStream extends InputStream {
     private PipedInputStream is = null;
     private PipedOutputStream os = null;
@Override
    public int read() throws IOException {
        Flux<DataFromDB> dataFlux = dao.getData(offSet); //offset fetchjed data in chunk
        dataFlux.doOnNext(dataFromDB -> os.write(dataFromDB.toString().getBytes())) //toString is for simplicity in reality we generate a string using some logic.
        .doOnTerminate(() -> dataFlux .count()
                        .subscribe(count -> {
                            if(count.longValue() == 0){
                                noMoreDataFromDB = true;
                            }
                            offSet= offSet+ count.intValue();
                            try {
                                os.flush();
                            } catch (IOException e) {
                                throw new RuntimeException(e);
                            }
                        })).subscribe(); 
        if(is.available() == 0 && noMoreDataFromDB ){
            IOUtils.closeQuietly(os);
        }
        return is.read();
    }
}