Как заставить Mono ждать, пока не будет запущен метод fetch - PullRequest
Новая
       7

Как заставить Mono ждать, пока не будет запущен метод fetch

0 голосов
/ 13 января 2019

Я пытаюсь реализовать функцию экспорта в excel через веб-сервис, который использует webflux, так как другие API и контроллеры работают хорошо. Моя проблема заключается в том, что вызов функции, которая генерирует файл Excel, доступен после извлечения данных из репозитория в виде потока (без проблем). Я отсортировал результаты и пытаюсь вызвать другой заполненный метид через flatMap. У меня возникло несколько проблем, пытаясь заставить это работать и убедиться, что код в flatMap запускается до того, как код в веб-сервисе вернет файл .

Ниже приведен код для веб-сервиса: 


    @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;
        }

И код для класса exelExport: 


    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++;
                }
            }
        }

Рабочая книга никогда не обновляется, поскольку populateEntry никогда не выполняется. Как я уже сказал, я пробовал несколько разных методов, включая Mono.just и Mono.when, но я не могу найти правильную комбинацию для обработки до того, как метод webservice попытается вернуть файл.

Любая помощь будет отличной.

Edit1: показывает идеальный метод crateDocument. 

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();
    }

Ответы [ 2 ]

0 голосов
/ 16 января 2019

Благодаря @SimonBasie за указатели мой рабочий код теперь выглядит следующим образом. 

@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;
    }
0 голосов
/ 14 января 2019

Есть несколько проблем в реализации вашего веб-сервиса.

Когда subscribe

Прежде всего, в реактивном программировании вы, как правило, должны пытаться построить одиночный конвейер обработки (вызывая операторы Mono и Flux и возвращая конечный результат как Mono и Flux ). В любом случае вы должны либо позволить платформе выполнить subscribe, либо, по крайней мере, подписаться только один раз в конце этого конвейера.

Здесь вместо этого вы смешиваете два подхода: ваш метод createDocument правильно возвращает Mono, но он также делает subscribe. Хуже того, подписка выполняется на промежуточном этапе, и ничто не подписывается на весь конвейер в методе веб-сервиса.

Таким образом, фактически никто не видит вторую половину конвейера (начиная с groupBy), и поэтому он никогда не выполняется (это ленивый Flux, также называемый «холодным» потоком).

Смешивание синхронного и асинхронного

Другая проблема - это проблема смешения двух подходов: ваш Flux ленив и асинхронен, но ваш веб-сервис написан в императивном и синхронном стиле.

Таким образом, код запускает асинхронный Flux из БД, немедленно возвращается к контроллеру и пытается загрузить данные файла с диска.

Вариант 1: сделать контроллер более Flux -ориентированным

Если вы используете Spring MVC, вы все равно можете написать эти обязательные контроллеры стилей, добавив при этом некоторый WebFlux. В этом случае вы можете вернуть Mono или Flux, и Spring MVC переведет его в правильную асинхронную конструкцию сервлета. Но это будет означать, что вы должны превратить обработку OutputStream и bytes в Mono, чтобы связать ее с документом Mono, используя что-то вроде then / flatMap / etc ... It немного сложнее.

Вариант 2: Превращение Flux в код обязательной блокировки

Другой вариант - вернуться к императиву и стилю блокировки, вызвав block() на createDocument() Mono. Это подпишется на него и будет ждать его завершения. После этого весь ваш императивный код должен работать нормально.

Примечание на стороне

groupBy имеет ограничение, при котором, если оно приводит к более чем 256 открытым группам, оно может зависнуть. Здесь группы не могут закрываться до тех пор, пока не будет достигнут конец файла, но, к счастью, поскольку вы обрабатываете данные только в течение одного месяца, поток не превысит 31 групп.

...