He escrito una lógica usando la biblioteca Spring Reactor para obtener todos los operadores y luego todos los dispositivos para cada operador (paginados) en modo asíncrono.
Creó un flujo para obtener todos los operadores y luego suscribirse a él.
final Flux<List<OperatorDetails>> operatorDetailsFlux = reactiveResourceProvider.getOperators(); operatorDetailsFlux .subscribe(operatorDetailsList -> { for (final OperatorDetails operatorDetails : operatorDetailsList) { getAndCacheDevicesForOperator(operatorDetails.getId()); } });Ahora, para cada operador, busco los dispositivos que requieren múltiples suscripciones para obtener el dispositivo mono que obtiene todas las páginas asíncronas al suscribirse al MONO.
private void getAndCacheDevicesForOperator(final int operatorId) { Mono<DeviceListResponseEntity> deviceListResponseEntityMono = reactiveResourceProvider.getConnectedDeviceMonoWithRetryAndErrorSpec( operatorId, 0); deviceListResponseEntityMono.subscribe(deviceListResponseEntity -> { final PaginatedResponseEntity PaginatedResponseEntity = deviceListResponseEntity.getData(); final long totalDevicesInOperator = PaginatedResponseEntity.getTotalCount(); int deviceCount = PaginatedResponseEntity.getCount(); while (deviceCount < totalDevicesInOperator) { final Mono<DeviceListResponseEntity> deviceListResponseEntityPageMono = reactiveResourceProvider.getConnectedDeviceMonoWithRetryAndErrorSpec( operatorId, deviceCount); deviceListResponseEntityPageMono.subscribe(deviceListResponseEntityPage -> { final List<DeviceDetails> deviceDetailsList = deviceListResponseEntityPage.getData() .getItems(); // work on devices }); deviceCount += DEVICE_PAGE_SIZE; } }); }Este código funciona bien. Pero mi pregunta, ¿es una buena idea suscribirse a mono desde adentro?
Lo dividí en dos flujos: primero obtuve todos los operadores y luego obtuve todos los dispositivos para cada operador.
Para la paginación estoy usando Flux.expand para extraer todas las páginas.
public Flux<OperatorDetails> getAllOperators() { return getOperatorsMonoWithRetryAndErrorSpec(0) .expand(paginatedResponse -> { final PaginatedEntity operatorDetailsPage = paginatedResponse.getData(); if (morePagesAvailable(operatorDetailsPage) { return getOperatorsMonoWithRetryAndErrorSpec(operatorDetailsPage.getOffset() + operatorDetailsPage.getCount()); } return Mono.empty(); }) .flatMap(responseEntity -> fromIterable(responseEntity.getData().getItems())) .subscribeOn(apiScheduler); } public Flux<Device> getAllDevices(final int opId, final int offset) { return getConnectedDeviceMonoWithRetryAndErrorSpec(opId, offset) .expand(paginatedResponse -> { final PaginatedEntity deviceDetailsPage = paginatedResponse.getData(); if (morePagesAvailabile(deviceDetailsPage)) { return getConnectedDeviceMonoWithRetryAndErrorSpec(opId, deviceDetailsPage.getOffset() + deviceDetailsPage.getCount()); } return Mono.empty(); }) .flatMap(responseEntity -> fromIterable(responseEntity.getData().getItems())) .subscribeOn(apiScheduler); }Finalmente, estoy creando una canalización y me suscribo a ella para activar la canalización.
operatorDetailsFlux .flatMap(operatorDetails -> { return reactiveResourceProvider.getAllDevices(operatorDetails.getId(), 0); }) .subscribe(deviceDetails -> { // act on devices });