Digamos que tengo una matriz de ID: [9, 8, 7, 6] .
Hago un poco de procesamiento y un elemento hace que se produzca una excepción. Quiero manejar esta situación a mi manera (digamos registrarla) y dejar que los otros elementos sigan la corriente.
¿Cómo puedo saber cuál era? Necesito tener este elemento en mi procesamiento onError .
Flux.fromArray(myArray) .flatMap(element -> { var foo = processMyEl(element); return anotherProcess(foo); // this returns Mono }) .onErrorOperator(element -> handleMyError(element)) // this line is what I need Entonces, lo que vi, es este casi agradable .onErrorContinue((error, obj) -> que emite un error y un objeto.
Pero este obj no es el element que causó la excepción sino el objeto que la hizo. Ocurre dentro de mis métodos de procesamiento y no tiene que ser el mismo tipo de objeto cada vez.
.onErrorReturn(...) - no es realmente lo que quiero
.doOnError(error -> - no hay información de mi elemento
.onErrorResume(error -> - igual que arriba
hubo sugerencias de que puedo crear mi propia excepción y pasar allí el elemento y luego recuperarlo de la excepción. Pero, ¿cómo debo lanzar la excepción?
¿Debería ir con una forma antigua de intentar atrapar:
Flux.fromArray(myArray) .flatMap(el -> { try { var foo = processMyEl(el); return anotherProcess(foo); // this returns Mono } catch (Exception e) { return Mono.error(new MyException(el)); } }) .onErrorOperator(error -> handleMyError(error.getElement()))no se ve bien
Editar:
No solo se ve mal, sino que tampoco funciona. La excepción no se detecta en absoluto y activa directamente doOnTerminate() y detiene todo el flujo
Actualizar:
Gracias a @JEY usé .onErrorResume() dentro flatMap .
También transformé el primer método para que fuera un flujo reactivo por Mono.defer(() -> Mono.just(processMyEl(el))) .
Solo como nota: usar Mono.defer() me permite usar onErrorResume ya que Mono.just() no puede señalar errores.
El código final se ve así:
Flux.fromArray(myArray) .flatMap(element -> Mono.defer(() -> Mono.just(processMyEl(element))) .onErrorResume(th -> handleMyError(element, th)) ) .flatMap(foo -> anotherProcess(foo) .onErrorResume(th -> handleMyError(foo, th) )Donde:
private Mono<> handleMyError(el, th) { // handling code return Mono.empty() }Según lo solicitado por @Kamil, agregaré mis comentarios como respuesta:
Simplemente debe manejar el error en flatMap y devolver un Mono.empty () para descartarlo, haga algo como:
Flux.fromArray(myArray) .flatMap(el -> anotherProcess(processMyEl(el)).onErrorResume(th -> handleError(th, el))Con error de manejo como:
Mono<Void> handleError(Throwable th, Object element) { LOG.error("An error occurred on {}", element, th); return Mono.empty() }O si desea hacer algo más complejo que requiera async:
Mono<Void> handleError(Throwable th, Object element) { return doSomethingThaReturnFluxOrMono(element).then(); }} catch (Exception e) { throw new MyException(el, e); }