Questions tagged [project-reactor]

Reactor is a foundational library building for reactive fast data applications on the JVM. It provides abstractions for Java, Groovy and other JVM languages to make building event and data-driven applications easier. It’s also really fast.

Filter by
Sorted by
Tagged with
0 votes
0 answers
7 views

Spring Webflux : Kafka consumer offset commit issue

For spring webflux which uses reactor kafka, I have following code for my consumer : public EventConsumer(KafkaReceiver<String, String> inputEventReceiver, MessageHelper ...
arqam's user avatar
  • 3,602
0 votes
1 answer
24 views

Reactor Flux remains blocks till getting response from server

I'v created a Flux which emits API response on subscription. The API takes few seconds to respond. I'm also switching the flow to a sceduler with 2 threads. private Scheduler sc = Schedulers....
quintin's user avatar
  • 812
0 votes
1 answer
16 views

How to get a list of messages with reactor-kafka receiver?

When we deal with synchronous approach we can receive from kafka a list of messages easily: @Bean public ConcurrentKafkaListenerContainerFactory<String, String> testContainerFactory() { ...
Mikhail Geyer's user avatar
1 vote
0 answers
16 views

Unable to cover Consumer function in code coverage for ReceiverOptions.addAssignListener()

I have been trying to increase code coverage on this particular piece of code for a while but can't seem to figure it out. This private method is invoked from within a public method. The public method ...
tripletk's user avatar
-1 votes
0 answers
25 views

subscribeOn(Schedulers.boundedElastic()) not working

Source code of this post can be found here Spring boot version: 2.3.9.RELEASE Reactor core version: 3.3.14.RELEASE I have a controller endpoint accept a binary payload and pass it to service ...
SoT's user avatar
  • 906
-1 votes
0 answers
17 views

How to save update after mao function applied

I applied a solution for modify a Flux of Product object with their Category objects. However, how do I save the modified objects from the map function. I tried it in the map function but that doesn'...
BreenDeen's user avatar
  • 643
0 votes
1 answer
17 views

Start Reactive Receiver after Spring Boot main method or after some event

I am using Reactor Kafka, but cannot solve the following problem. Is it possible to configure Receiver to start after the application is fully launched, rather than at the moment with the application? ...
Viktor Loskutov's user avatar
1 vote
1 answer
36 views

How can I combine a Mono and a Flux

I am creating a Flux of Product and a Mono Category object that needs to be applied to the Products because all they all all to be grouped into the same category. I am attempting to add the category ...
BreenDeen's user avatar
  • 643
0 votes
0 answers
14 views

Spring security @PreFilter & @PostFilter for reactive methods

Is there an example of using @PostFilter ACL's in spring-security on a domain object say X , which is reactive ? For example, the method signature would be public Flux<X> repositoryMethod(). I ...
schengalath's user avatar
-2 votes
0 answers
31 views

While loop with access to result of previous reactive call [closed]

How to write this trivial lines of code with Project Reactor? @Test public void test5(){ boolean result =true; while (result){ result = reactiveCall(result); } System.out....
Андрей Чижик's user avatar
0 votes
1 answer
22 views

How map an exception from a Mono to another exception

I have an endpoint in Spring Webflux (Java 20) that receives a UUID string and I need to validate it using UUID.fromString. If the latter throws an exception because the String is malformed, how can ...
Giustiniano's user avatar
0 votes
1 answer
19 views

Project Reactor: is it right to call subscribe within async context?

I stock encryption keys in redis database. I want to implement a fallback mecanism to generate keys unless they are already present. What perturbates me with a solution I come up with is that I call ...
VladRia's user avatar
  • 1,485
0 votes
0 answers
22 views

R2DBC Error with Spring Boot: Cannot encode value of type 'io.r2dbc.spi.Parameters$InParameter'

I'm developing a service using Spring Boot 3 and R2DBC to upload data to a MySQL 8.0 database. While doing so, I've encountered the following error when trying to bind parameters: reactor.core....
Koishi's user avatar
  • 1
0 votes
1 answer
33 views

Handle slow producer with Project Reactor/WebFlux to avoid overloading it

Backpressure is a mechanism that allows the consumer to slow down the producer when it cannot keep up with the producer rates. Can we do it in the opposite way? This is the scenario: There is an high ...
SGiux's user avatar
  • 629
0 votes
1 answer
36 views

Project Reactor - how to check if Mono<Void> was actually called?

I am a complete newbie to reactional programming (and I've read a bunch of similar SO questions). In my current pet-project, I've run into the following problem (I'm providing a pseudo code for a ...
worm359's user avatar
0 votes
0 answers
19 views

StepVerifier project reactor

I'm using springboot 3.1.3 and springwebflux I'm not understanding how to set up unit tests. Let's suppose I have a simple service like this: private Mono<Boolean> sendMessage(String uri, String ...
Angelo Immediata's user avatar
0 votes
1 answer
16 views

Is it possible to have exactly once semantics, for one producer with multiple consumer in Reactive Kafka?

I wonder if it's possible to implement exactly-once semantic with distributed consumer, in reactive manner. In the reference doc for Reactor Kafka, we have a sample code for Exactly-once delivery. The ...
Yosi Pramajaya's user avatar
0 votes
1 answer
22 views

Project Reactor's subscribeOn() doesn't affect onSubscribe() to be called in specified Scheduler as stated in JavaDoc

Project Reactor's Javadoc states about method: public final Flux<T> subscribeOn(Scheduler scheduler) that Run subscribe, onSubscribe and request on a specified Scheduler's Scheduler.Worker. As ...
Maksym Zhokha's user avatar
2 votes
0 answers
55 views

Context-Propagation does not work in @SpringBootTest

my Spring-WebMvc application which uses WebClient (reactor) loses the context in a @SpringBootTest. I have a @SpringBootApplication which offers a @RestController endpoint to the caller. When the ...
jps's user avatar
  • 21
0 votes
0 answers
21 views

The dispatcher returned from the ServletContext does not support asynchronous dispatching

I have following api returning Flux of String @GetMapping(value = "/api/getStrListFlux",produces = MediaType.TEXT_EVENT_STREAM_VALUE) public Flux<String> getListOfStr0() { ...
Pramod Sharma's user avatar
-1 votes
0 answers
43 views

reactory netty connection pool configuration details for maxLifeTime and maxIdleTime

I am using projectreactor netty connection pool and setting up a connection pool for the HttpClient. I referred a lot for the documentation from here : https://projectreactor.io/docs/netty/release/...
Venu's user avatar
  • 1,521
0 votes
0 answers
9 views

Send a message to an already established request-stream connection from the rsocket-server to all signed rscoket-clients except one

I have two rsocket-client and one rsocket-server. Both rsocket-client initially subscribe to the broadcast route listenCommand to receive commands from rsocket-server. Then one of the clients sends a ...
Tasha Bond's user avatar
0 votes
1 answer
29 views

Implement filter using project reactor

I have a list of items from Mongo and I need to filter them by remote call result, something like this: repository.findByString(String string) // Flux<SomeObject> .filter(a -> { ...
Fairy's user avatar
  • 511
-1 votes
0 answers
30 views

Unit testing nested context

I am writing a Spring WebFlux WebFilter which is meant to put some value into the Project Reactor context, to be used somewhere in the application code. The actual filter implementation is a fair bit ...
Ben's user avatar
  • 350
0 votes
0 answers
24 views

Reactor Kafka "Can't signal value due to lack of requests" when using sample() - "Concurrent Processing with Partition-Based Ordering"

We have implemented a couple of Reactor Kafka based Spring Boot WebFlux services. Within the services we use the approach described in "5.12. Concurrent Processing with Partition-Based Ordering&...
gira1's user avatar
  • 117
0 votes
1 answer
54 views

.subscribe(-) vs .map(-).subscribe() in reactive Java

I'm reading the project reactor documentation on transforming data using the transforming operators on Mono and Flux. But I can help asking the question.. What's the difference between doing logic in ...
Kaigo's user avatar
  • 1,267
0 votes
1 answer
37 views

How to synchronize rest method in Webflux?

The problem is that when the updateUser is called at the same time, an incorrect balance update occurs. 2023-09-03T18:20:53.035+03:00 INFO 12668 --- [actor-tcp-nio-1] com.example.testsync....
Dmirii Berrington's user avatar
0 votes
2 answers
47 views

Groovy Spock Test is failing when using StepVerifier for a simple Service

I am attempting to test my service ProductService, but it is failing indicating that the assert is a MonoJust instead of the Product I'm expecting. I don't know why test is failing with the issue. I ...
BreenDeen's user avatar
  • 643
0 votes
0 answers
26 views

How to delay emission between elements in Reactor Scheduler?

Now the difference in the execution time of serviceBar.externalMethod1(string) and serviceBar.externalMethod2(string) is equal to the difference in the call delay of method1and method2. I need the ...
Dmirii Berrington's user avatar
0 votes
0 answers
19 views

Retrying and dropping erroneous elements of a Flux without pausing downstream demand for data during retry backoff period

I have a Flux of elements and need to attempt to process each element while retrying the processing of each element if an error occurs. After a maximum number of retries, I need to drop erroneous ...
user22479020's user avatar
0 votes
1 answer
27 views

issue in using reactive apps with lambda

I am pretty new to use of reactive in Java. Here below is the issue that I am facing. I have developed a springboot-REST application with contoller, service and repository layer with R2DBC driver. ...
Ram's user avatar
  • 327
0 votes
0 answers
31 views

Is there utils functions that merge flatMap and flatMapIterable?

I'm using Reactor 3 in Java and I am ofen finding myself in front of this case : I have a Mono<A>, that I map to a Mono<List<B>> (flatMap), and then I want to transform this into a ...
floxcc's user avatar
  • 62
0 votes
0 answers
38 views

Flatmap() on Mono is being called twice (Java 11 and project reactor)

Here I am acting on a method called accountDatafor() which returns a Mono<Response>, the method I'm working in also returns a Mono<Response>. The aim is to avoid the block() at the end of ...
Kaigo's user avatar
  • 1,267
0 votes
0 answers
33 views

Reactor StepVerifier with expectNoEvent / then

I've implemented a queuer in Spring WebFlux as I asked about here: Queuing up requests in Spring WebFlux TL;DR is that it queues up incoming requests to an external API, and then batches them after ...
orderlyfashion's user avatar
0 votes
0 answers
50 views

Why is Blockhound detecting a blocking call after I added r2dbms to spring?

I added postgres support to my application, but when I try to run the tests, I keep getting a blocking call error from blockhound, and I'm not sure why this is happening. The application it's runs ...
Ifeanyi Onyejekwe's user avatar
1 vote
0 answers
21 views

Websockets with Spring WebFlux: properly mix authentication data with event treatment

I develop an application where users connect via WebSocket and exchange messages. I use Spring WebFlux and Reactive WebSockets. Security component provide me with a Mono containing current Principal: ...
VladRia's user avatar
  • 1,485
1 vote
1 answer
56 views

reactor iterate flux until not empty

I've reactor flux with string values, I iterate by each value and check if given value exists in database. I'd like to iterate over each value until I find the one in database. If not found (after all ...
testtt's user avatar
  • 11
0 votes
1 answer
44 views

Custom log format for HttpRequest and HttpResponse using webClient

I need log HttpRequest body and HttpResponse body in custom format. For example "Send request. Headers: {}, Body: {}, uri: {}" How can I get request body? I'm try this: log.info("Send ...
Влад Савостиков's user avatar
0 votes
1 answer
20 views

Webflux Reactor how to merge Stream.map output, which can be Mono and Flux into Flux

A is used to get source. In the following map function, source is a Flux if a owns source otherwise source is a Mono. Function returns all sources that can be queried based on List of A, list size ...
Tiina's user avatar
  • 4,295
0 votes
2 answers
49 views

Reactive Java: change window size at runtime

I have a Flux emitting items every 1 seconds, on repeat. How can I change the window size dynamically? public static void main(String[] args) throws InterruptedException { Duration interval = ...
krinklesaurus's user avatar
0 votes
1 answer
59 views

Is there a way to cache responses using Mono from a WebClient call?

Is there a way to cache the response from a call if the inputs are the same? For example, if the fileName is the same, I don't want to make the request out with my webClient again since it takes a ...
Jack Johnson's user avatar
  • 1,335
1 vote
0 answers
46 views

Project reactor limitRate operator unexplained behavior

I have a relatively simple task - I have records in a database that need to be processed sequentially and synchronously. Each record has a timestamp column which indicates the last processing time. ...
Daniel Yordanov's user avatar
0 votes
0 answers
64 views

Sinks.Many silently becomes useless after subscriber count reaches 0 the first time

If I make a Sinks.Many instance: Sinks.Many<String> sink = Sinks.many().multicast().onBackpressureBuffer(); Then in a different thread (T1), I subscribe to its Flux: new Thread(() -> { ...
kaqqao's user avatar
  • 12.9k
1 vote
1 answer
33 views

Creating a Mono from certain elements of a Flux

I am trying to create a Mono from elements of a Flux. This mono will need to contain a certain number of elements. I found a solution, which is use the Mono.create method, the takeUntil and onCancel ...
Siri's user avatar
  • 97
0 votes
2 answers
33 views

how to map flux stopping if async condition is satisfied?

consider i have a flux of integers and this method simulating an async external api data retrieval, which can return an empty response for some specific unknown input: public static Mono<String> ...
rado's user avatar
  • 5,760
0 votes
1 answer
62 views

Ordered data with Flux.groupBy, replace take(Duration.ofMillis) with e.g. takeUntil

Given the below code where I'm using Flux groupBy and .take(Duration.ofMillis(10)), I can process like 50K records per second. In tests on localhost with Flux.just, I can set the delay value to 1 ms ...
Roar S.'s user avatar
  • 8,209
2 votes
2 answers
97 views

Queuing up requests in Spring WebFlux

In my API I have a service that calls an external API. This external API has a cost per request. Hence I would like to queue up the requests to the external API. This is how it looks today: @Component ...
orderlyfashion's user avatar
1 vote
0 answers
49 views

How to have all the exceptions thrown in the Mono context when using Spring reactive WebClient?

Code I have a Spring reactive web client used to connect to a web service. @Autowired private org.springframework.web.reactive.function.client.WebClient webClient; public Mono<MyData> getMyData(...
Honza Zidek's user avatar
  • 9,274
0 votes
1 answer
43 views

Unable to use schemaRegistry in reactor kafka

I am tying to setup kafka consumer using reactor kafka . Producer is integrated with kafka schema registry @Value("${spring.kafka.schemaRegistryUrls}") private String ...
Gajukorse's user avatar
  • 147
0 votes
1 answer
31 views

Convert Flux<ObjectWithLists> to Mono<ObjectWithLists>

I have to following objects class ObjectA { List<String> success; List<String> failure; } And class ObjectB { List<String> completed; List<String> unCompleted; ...
Kaigo's user avatar
  • 1,267

1
2 3 4 5
64