Почему асинхронный WebClient не обрабатывает до конца?

Question

[федор unknown]

Добрый день!
у меня есть 2 api endpoint, по которым стучится Webclient.
Дело в том, что если я выкармливаю данные из 200 объектов, то он обрабатывает не до конца.
в итоге в базе должно быть 200 записей, но количество записей всегда по разному.
код:

public void getAdditionalInfo(List<ExcelModel> excelList) {
        log.warn("Beginning send request to: ST-helper and Sybase-helper API`s");
        for (ExcelModel excel : excelList) {
            Mono<Response> postgresResponse = webClient.get()
                    .uri(postgres + excel.getSuip())
                    .retrieve()
                    .bodyToMono(Response.class)
                    .timeout(Duration.ofSeconds(3000000))
                    .filter(Response::getResult);

            // When request completed
            postgresResponse.subscribe(response -> {
                excel.setAmount(response.getAmount());
                excel.setRequisite(response.getRequisite());
                excel.setIsValidated(true);
                excelModelRepository.save(excel);
                log.info("postgres: {}", excel);
            });

//**********************************************************************************************

           //  Sending async request to sybase (old base):
            Mono<Response> sybaseResponse = webClient.get()
                    .uri(sybase + excel.getTransactionID())
                    .retrieve()
                    .bodyToMono(Response.class)
                    .timeout(Duration.ofSeconds(3000000))
                    .filter(Response::getResult);

            // When request completed
            sybaseResponse.subscribe(response -> {
                excel.setAmount(response.getAmount());
                excel.setRequisite(response.getRequisite());
                excel.setIsValidated(true);
                excelModelRepository.save(excel);
                log.info("sybase: {}", excel);
            });
        }
    }

Comments

[Dmitry Roo]

Залогируйте это место: .filter(Response::getResult); Возможно, не всегда ответ просто приходит.
И еще: у вас куча паралельных запросов. Пора подумать о выделенном ExecutorService-е.

[федор unknown]

Dmitry Roo,

место: .filter(Response::getResult); Возможно, не всегда ответ просто приходит.

если первый запрос не прошел в фильтре (potgres), то он пойдет на второй (sybase)

[Михаил]

если первый запрос не прошел в фильтре (potgres), то он пойдет на второй (sybase)

Нет. У вас запросы в постгрес и сибейз выполняются параллельно. Не может проблемы быть в этом - каждая модель сохраняется дважды?

[федор unknown]

Михаил, идея заключается в том что есть объект ExcelModel:

public class ExcelModel {
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;
    private String date;
    private String state;
    private String suip;
    private String transactionID;
    private String requisite;
    private Long amount;
    private Boolean isValidated = false;
}

сперва идет запрос в postgres (URL + ExcelModel.getSuip)
если он вернул:

{
    "result": false,
    "requisite": "",
    "amount": 0,
    "created": null,
    "suip": null
}

то стучимся по: sybase (URL + ExcelModel.getTransactionID)
P.s хотелось бы на sybase отправлял запросы в том случае, если postgres вернул объект с пустым значениями

[Михаил]

федор unknown, ну так и поставьте обращение к sybase (flatMap) в цепочку к постгрес.

Mono<Response> getDataFromPosgres();
Mono<Response> getDataFromSybase();
boolean ifEmpty(Response response);
void saveToDatabase(Response response);

getDataFromPostgres()
  .flatMap(response -> if (isEmpty(response)) return getDataFromSybase(); else return Mono.just(response))
  .subscribe(response -> saveToDatabase(response));