Reduce a flux to mono using data from flux

I have this scenario. I have one paginated API which gives me the data for last 12 months. The response of the API is like: public class PagedTransfersDto { private List content; …

Is Reactor chain still running

I have an application based on Spring WebFlux and Reactor. Application starts everyday on a schedule and loads statistics. But there is also available a manual launch option through controller. So if …

Java reactor how to properly start async cancellable sideeffect

I’m trying to write something using reactor which I know how to write using completable futures. I’m getting “Calling subscribe in non-blocking scope” warning in it. My goal is to call …

How to create a generic wrapper for just any method call?

I want to create a helper method that can wrap/convert just any sync method call into an async Mono. The following is close, but shows an error: Required type: Mono Provided: Mono<? …

Continue consuming subsequent records in reactor kafka after deserialization exception

I am using reactor kafka and have a custom AvroDeserializer class for deserialization of messages. Now I have a case where for certain payloads the deserialization class throws an exception. My Kafka …

How to handle file access in a reactive environment

I’m refactoring some blocking code to reactive (using Reactor). I think most of the methods in the java.nio.file.Files class are blocking. Is it right if I replace a method like: public boolean exists(…

Combine two Stream into one Flux

How can I combine two streams Stream into Flux? What I understand is that I might need to use Flux create method to create this but I am not really sure about it: flux1.create(sink ->…

Obtaining a nested objects using Spring Data R2DBC

I’m new to Project Reactor and R2DBC. How to receive and merge Flux with Mono properly using Spring Data R2DBC reactive repositories? Parent: @Table(“parent”) public class …