How do I get a Mono to wait till dependenat fetch method has run

Question

I am trying to implement an export to excel function via a web service which uses webflux as the other api and controllers work well. My problem is that calling the function that generates the excel file is accessed after retrieving data from repository as a Flux (no problem there). I have sorted the results and am trying to call another populate methid via flatMap, I am having a number of issues trying to get this to work and to make sure that the code in the flatMap runs before the code in the webservice to return the file.

Below is the code for the webservice:

    @GetMapping(API_BASE_PATH + "/download")
        public ResponseEntity<byte[]> download() {
            Mono<Void> createExcel = excelExport.createDocument(false);

            Mono.when(createExcel).log("Excel Created").then();

            Workbook workbook = excelExport.getWb();

            OutputStream outputStream = new ByteArrayOutputStream();
            try {
                workbook.write(outputStream);
            } catch (IOException e) {
                e.printStackTrace();
            }

            byte[] media = ((ByteArrayOutputStream) outputStream).toByteArray();
            HttpHeaders headers = new HttpHeaders();
            headers.setCacheControl(CacheControl.noCache().getHeaderValue());
            headers.setContentType(MediaType.valueOf("text/html"));
            headers.set("Content-disposition", "attachment; filename=filename.xlsx");
            ResponseEntity<byte[]> responseEntity = new ResponseEntity<>(media, headers, HttpStatus.OK);
            return responseEntity;
        }

And the code for the exelExport class:

    public Mono<Void> createDocument(boolean all) {
            InputStream inputStream = new ClassPathResource("Timesheet Template.xlsx").getInputStream();
            try {
                wb = WorkbookFactory.create(inputStream);
                Sheet sheet = wb.getSheetAt(0);
                Row row = sheet.getRow(1);
                Cell cell = row.getCell(3);
                if (cell == null)
                    cell = row.createCell(3);
                cell.setCellType(CellType.STRING);
                cell.setCellValue("a test");

                log.info("Created document");

                Flux<TimeKeepingEntry> entries = service.findByMonth(LocalDate.now().getMonth().getDisplayName(TextStyle.FULL, Locale.ENGLISH)).log("Excel Export - retrievedMonths");
                entries.subscribe();

                return entries.groupBy(TimeKeepingEntry::getDateOfMonth).flatMap(Flux::collectList).flatMap(timeKeepingEntries -> this.populateEntry(sheet, timeKeepingEntries)).then();
            } catch (IOException e) {
                log.error("Error Creating Document", e);
            }

            //should never get here
            return Mono.empty();
        }

    private void populateEntry(Sheet sheet, List<TimeKeepingEntry> timeKeepingEntries) {
            int rowNum = 0;
            for (int i = 0; i < timeKeepingEntries.size(); i++) {
                TimeKeepingEntry timeKeepingEntry = timeKeepingEntries.get(i);
                if (i == 0) {
                    rowNum = calculateFirstRow(timeKeepingEntry.getDay());
                }
                LocalDate date = timeKeepingEntry.getFullDate();
                Row row2 = sheet.getRow(rowNum);
                Cell cell2 = row2.getCell(1);
                cell2.setCellValue(date.toString());
                if (timeKeepingEntry.getDay().equals(DayOfWeek.FRIDAY.getDisplayName(TextStyle.FULL, Locale.ENGLISH))) {
                    rowNum = +2;
                } else {
                    rowNum++;
                }
            }
        }

The workbook is never update because the populateEntry is never executed. As I said I have tried a number of differnt methods including Mono.just and Mono.when, but I cant seem to get the correct combination to get it to process before the webservice method tries to return the file.

Any help would be great.

Edit1: Shows the ideal crateDocument Method.

public Mono<Void> createDocument(boolean all) {
        try {
            InputStream inputStream = new ClassPathResource("Timesheet Template.xlsx").getInputStream();
            wb = WorkbookFactory.create(inputStream);
            Sheet sheet = wb.getSheetAt(0);

            log.info("Created document");

            if (all) {
                //all entries
            } else {
                service.findByMonth(currentMonthName).log("Excel Export - retrievedMonths").collectSortedList(Comparator.comparing(TimeKeepingEntry::getDateOfMonth)).doOnNext(timeKeepingEntries -> {
                    this.populateEntry(sheet, timeKeepingEntries);
                });
            }
        } catch (IOException e) {
            log.error("Error Importing File", e);
        }
        return Mono.empty();
    }

Answer 1

Thanks to @SimonBasie for the pointers, my working code is now as follows.

@GetMapping(value = API_BASE_PATH + "/download", produces = "application/vnd.ms-excel")
    public Mono<Resource> download() throws IOException {
        Flux<TimeKeepingEntry> createExcel = excelExport.createDocument(false);

        return createExcel.then(Mono.fromCallable(() -> {
            ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
            excelExport.getWb().write(outputStream);
            return new ByteArrayResource(outputStream.toByteArray());
        }));
}

public Flux<TimeKeepingEntry> createDocument(boolean all) {
        Flux<TimeKeepingEntry> entries = null;
        try {
            InputStream inputStream = new ClassPathResource("Timesheet Template.xlsx").getInputStream();
            wb = WorkbookFactory.create(inputStream);
            Sheet sheet = wb.getSheetAt(0);

            log.info("Created document");

            if (all) {
                //all entries
            } else {
                entries = service.findByMonth(currentMonthName).log("Excel Export - retrievedMonths").sort(Comparator.comparing(TimeKeepingEntry::getDateOfMonth)).doOnNext(timeKeepingEntry-> {
                    this.populateEntry(sheet, timeKeepingEntry);
                });
            }
        } catch (IOException e) {
            log.error("Error Importing File", e);
        }
        return entries;
    }

Answer 2

There are several problems in the implementation of your webservice.

When to subscribe

First off, in reactive programming you must generally try to build a single processing pipeline (by calling Mono and Flux operators and returning the end result as a Mono and Flux). In any case, you should either let the framework do the subscribe or at least only subscribe once, at the end of that pipeline.

Here instead you are mixing two approaches: your createDocument method correctly returns a Mono, but it also does the subscribe. Even worse, the subscription is done on an intermediate step, and nothing subscribes to the whole pipeline in the webservice method.

So in effect, nobody sees the second half of the pipeline (starting with groupBy) and thus it never gets executed (this is a lazy Flux, also called a "cold" Flux).

Mixing synchronous and asynchronous

The other problem is again an issue of mixing two approaches: your Flux are lazy and asynchronous, but your webservice is written in an imperative and synchronous style.

So the code starts an asynchronous Flux from the DB, immediately return to the controller and tries to load the file data from disk.

Option 1: Making the controller more Flux-oriented

If you use Spring MVC, you can still write these imperative style controllers yet sprinkle in some WebFlux. In that case, you can return a Mono or Flux and Spring MVC will translate that to the correct asynchronous Servlet construct. But that would mean that you must turn the OutputStream and bytes handling into a Mono, to chain it to the document-writing Mono using something like then/flatMap/etc... It is a bit more involved.

Option 2: Turning the Flux into imperative blocking code

The other option is to go back to imperative and blocking style by calling block() on the createDocument() Mono. This will subscribe to it and wait for it to complete. After that, the rest of your imperative code should work fine.

Side Note

groupBy has a limitation where if it results in more than 256 open groups it can hang. Here the groups cannot close until the end of the file has been reached, but fortunately since you only process data for a single month, the Flux wouldn't exceed 31 groups.