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.
2,674
questions
0
votes
0
answers
31
views
How to loop through a list of Mono objects in kotlin? [duplicate]
I am new to Kotlin, sorry if it is simple but couldn't find the answer on the internet. I have a list of Mono objects. The objects have user information including email, like
data class User(var ...
0
votes
1
answer
37
views
Which scheduler should I use for a "penalty box" when I do delayElement in WebFlux?
I am trying to implement a "penalty box" for unsuccessful attempts to use our API when it is dealing with security tokens. This extends my previous question Would delayElement be ...
0
votes
1
answer
57
views
Spring Kafka error after upgrading spring boot to 2.6.6
I am facing runtime error after I have upgraded spring boot version from 2.4.0 to 2.6.6 (and spring cloud version 2021.0.2). My application consumes messages from a kafka topic and produces it to ...
0
votes
1
answer
13
views
How to determine downstream demand from multicast sink
Is there any way to determine the minimum downstream request among subscribers to a multicast sink? With a regular FluxSink there is requestedFromDownstream() which I can use to implement backpressure,...
1
vote
0
answers
24
views
How do I create a hot flux that will provide data and replenish itself as needed
Since SecureRandom is slow, I was wondering how to make it so that I have a Flux that I can request the next random number, but it is prepopulated by say 100 items initially and is periodically ...
0
votes
1
answer
25
views
Flux delayElements without initial delay
I would like to emit elements where elements are delayed each other by some time without initial delay.
I can not use
Flux.delayElements(Duration)
because it delays initial delay before first element ...
0
votes
1
answer
27
views
A difference between Mono.mapNotNull(....) and Mono.flatMap(Mono.justOrEmpty(....))
Being a novice to Project Reactor I'm trying to wrap my head around the documentation. However I still can't figure out if there any practical difference between the following (pseudo-)code:
...
1
vote
1
answer
28
views
Converting a mono object into two different mono object and the root mono is executing two times
Suppose I have create a Student entity and storing it into database using reactive.
Student student{
id;
name;
institutionId;
courseId;
}
I am storing this student entity using a ...
0
votes
1
answer
47
views
Spring WebClient aggregates multiple 401 UNAUTHORIZED error to throw Exceptions.CompositeException
When WebClient calls an external api (with different query params) asynchronously with a wrong token, the first few returns with 401 UNAUTHORIZED and the control flow stops with Exceptions....
0
votes
1
answer
27
views
How do you take an initial sample from a flux?
The sample operation takes the first entry from flux after the duration. How can I make it such that I take the first entry on subscribe and then do the rest?
My thinking is that it is something ...
0
votes
1
answer
44
views
Measure execution time in Project Reactor
I'm trying to load test one of my projects and I think somewhere in my changes I may have introduced some point where the performance is getting impacted.
Is there a way to measure execution time and ...
0
votes
0
answers
42
views
Unable to start Netty in spring boot 2.5.13
I have a spring boot application based on reactor but it can't be launched on Window 10 although it works fine on Linux. It failed to start Netty. This application uses spring boot 2.5.13 now. This ...
-1
votes
0
answers
17
views
reactor Mono<Flux> to Flux [closed]
I have a type Mono<Flux<T>> and I need to convert it to type Flux<T>
this can be done by (in kotlin)
Mono.just(listOf(1,2,3,4).toFlux())
-1
votes
1
answer
22
views
Project Reactor Schedulers on field level or chain level
I saw that when schedulers are extracted to field or bean in Spring then Intellij suggest that is a blocking call in non-blocking scope. I know that but when I put Scheduler like below then Intellij ...
1
vote
1
answer
31
views
Does @Scheduled support Project Reactor for fixedDelay with initial delay of zero?
Similar to Spring webflux/reactor using @Scheduled to read database and perform some tasks but I am asking specifically for fixedDelay
I know I can't just use a Mono.subscribe() as the delay semantics ...
0
votes
1
answer
24
views
Is there an equivalent to doOnNext but ignoring any reactive results (except errors)?
Presently I am doing this
final ReactiveHashOperations<String, String, String> ops = redisTemplate.opsForHash();
var theMonoIWant =
Mono.fromCallable(this::...
0
votes
0
answers
18
views
Does ReactiveDiscoveryClient.probe actually do anything?
I'm looking through the code for ReactiveDiscoveryClient and I noticed the following
Flux<String> getServices();
/**
* Can be used to verify the client is still valid and able to ...
0
votes
1
answer
23
views
Would delayElement be susceptible to a DoS attack?
I have a service that authenticates a user and I put delayElement for 2 seconds when it is about to return a 401.
When I run through artillery, it seems to start timing out some of the requests and I ...
0
votes
1
answer
54
views
How to buffer and group elements in Reactor Flux in Java
Given an infinite flux of objects, where each object has an ID, how can I use flux to create a buffered list of updates grouped by ID property (keeping the last emitted value)?
Thanks
Example
Obj(...
0
votes
1
answer
44
views
Reactor Java unit test what is the order of return values from a mocked dependency
I am a beginner at Reactive Java. I am trying to understand why a simple unit test fails when a mock dependency is setup to return two different values. This dependency is called twice in the main ...
0
votes
1
answer
39
views
doOnFirst element in Reactor Flux?
I would like to perform some action (side effect) only on the first emission of the element in the Flux.
Is there a way to do that?
Some context: I want to call .elapsed() on Flux and log only the ...
0
votes
1
answer
33
views
How to make order-independent assertions on Flux output?
I have a test case for a Flux from Project Reactor roughly like this:
testMultipleChunks(StepVerifier.FirstStep<Chunk> verifier, Chunk chunk1, Chunk chunk2) {
verifier.then(() -> {
...
0
votes
1
answer
26
views
Project Reactor onErrorResume get stuck
I'm using project reactor and I have a very long flow in which I get an exception (when parsing a string to json with Jackson). The thing is that even though I use
.map(this::parser)
.onErrorResume(...
-1
votes
1
answer
19
views
Stay subscribed to Flux/Mono after client disconnect in Spring WebFlux
I have a service where a couple requests can be long running actions. Occasionally we have timeouts for these requests, and that causes bad state because steps of the flux stop executing after the ...
1
vote
2
answers
58
views
Spring 5 - Chain reactive webclient calls
I'm trying to learn how to use the reactive WebClient in Spring 5.
Let's say I have 3 calls I need to make:
Get an access token
Get a list of accounts
For each account, get a list of transactions.
...
0
votes
0
answers
17
views
Spring reactive WebSocketHandler mono does not reach doFinally block
I am trying to handle closing web socket session in WebSocketHandler. My intuition was to do it in this way:
webSocketClient.execute(
URI.create("some-ws-endpoint")
) { session: ...
0
votes
0
answers
21
views
Correct way of returning an object from an async operation to continue the Flux
I'd like to perform an async operation e.g. writing to a database, and after that continue the Flux with the same object that the database write function has received. The thing is that the response ...
1
vote
1
answer
50
views
How do I write a subscriber that accepts data from two related streams?
I'm new to reactive programming and Project Reactor so apologies in advance if I use the wrong words.
I have a simple procedural program that I'm converting to reactive, more as a learning activity ...
1
vote
0
answers
27
views
Is it fine to use block() in spring reactor tests for test state creation?
I'm writing integration tests using Spring, JUnit 5, JOOQ, TestContainers and Project Reactor. I want to test my repository layer, e.g. verify
Mono<Entity> get(Long id) method.
Therefore, my ...
0
votes
1
answer
35
views
Shuffle emissions of a Flux in Reactor?
I have a Flux and I would like to shuffle its emissions.
So for emissions:
1, 2, 3, ...
I would like to have:
32, 5, 9, ...
Is it possible in Reactor?
0
votes
1
answer
40
views
How do you add elements to a Redis Hash and set the expiration using ReactiveRedisTemplate without block?
If it wasn't reactive I'd do something like.
/**
* Given a refresh token and the user name, create a "payload" to represent
* secret data and store it into Redis as a hash and set it to ...
0
votes
0
answers
23
views
Is there a style guide/convention on using Project Reactor?
I am just trying to find examples of using Project Reactor hoping to see somewhat of a convention to use for the method and variable names for Flux/Mono and their respective generators.
Going through ...
0
votes
1
answer
30
views
Reactor: delayElement doesn't work with Flux
So I want to use Flux to run each Mono task and wait each task for 2 seconds. Here's what I have tried
val test = { x: Int ->
Mono.just(x)
.log()
.doOnNext { println("...
1
vote
1
answer
42
views
Pause kafka read in reactive Kafka
I have an application that reads from kafka using reactive kafka. Below is the code to read from kafka:
public Flux<String> readFromKafka() {
return kafkaConsumerTemplate
...
0
votes
0
answers
47
views
Webflux, reactor connection reset by peer unhandled
So basically I am simulating a 'Connection reset by peer' locally using Toxiproxy or Wiremock (same behaviour for both of them) and I get the exception below.
The tricky part is that it will bypass ...
0
votes
1
answer
32
views
How do I build a mono from a callable that would return a mono?
I have this code that would call reactive Spring Redis Template to store some data I am stuck at the point of the switchIfEmpty where I want to create a new entry and store it and return the result.
...
0
votes
2
answers
58
views
How do I add a delay with data from the Mono?
I have a service that is returning a value which contains delay information.
public Mono<R> authenticate(
A authenticationRequest,
@RequestHeader Map<String, String> headers,
...
11
votes
0
answers
198
views
r2dbc - This is likely a bug leading to unpredictable outcome
We use spring-data-r2dbc:1.3.2 and dev.miku:r2dbc-mysql:0.8.2.RELEASE and experience a strange issue in production.
We don't understand what is the root cause and whether it is a recoverable or can ...
2
votes
1
answer
34
views
How do I apply a Reactive Java StepVerifier to a Pojo
I have the following code (please see below), which successfully uses a StepVerifier to call the ParsePerson::Parse method, and then expectNext for "John" and "Mary".
If the ...
1
vote
1
answer
478
views
java.lang.NoSuchMethodError: 'reactor.util.context.ContextView reactor.core.publisher.MonoSink.contextView()'
After upgrade Spring Boot version from 2.6.6 to 2.6.7, I'm getting below error while executing web client rest calls. Any idea?
Sample code
public void execute(BiConsumer<ResponseEntity<JsonNode&...
1
vote
1
answer
86
views
Whats the difference between flatMap, flatMapSequential and concatMap in Project Reactor?
I've read from the documentation that flatMap:
Transform the elements emitted by this Flux asynchronously into Publishers, then flatten these inner publishers into a single Flux through merging, ...
1
vote
1
answer
51
views
convert to Mono<List<String>> from Flux<List<String>> in Java
I have this entity class.
public class MyClass {
private String category;
private List<String> topics;
}
I am using firestore database with spring reactive. I am trying to return all ...
1
vote
1
answer
28
views
Is there equivalent of RxJava Completable in project reactor?
I don't want to return anything from my Mono, just wait for it to complete without errors. In RxJava I would use Completable, but what is recommended way of doing it in Project-Reactor?
0
votes
1
answer
24
views
Stream returns wrong type
I'm trying to understand reactive style. But stuck on this example.
public class ScriptServiceImpl implements ScriptService{
private static Logger log = LoggerFactory.getLogger(ScriptServiceImpl....
0
votes
0
answers
27
views
what is Mono.zipDelayError equal function in Flux
I want to Triger the below two calls parallelly and combine the Mono and flux.
Mono<EmpAddressDetail> empAddDetail = getTimeoutDuration()
.flatMapDelayError(duration -> ...
0
votes
1
answer
31
views
Spring Reactive programming if search is not there in database then prepare entity and save else return new instance of object
I am new to reactive world and trying to write for following logic:
find if entry is there in database corresponding to my service call. If the the data is not there in database then i have to preapre ...
1
vote
1
answer
68
views
What is the default scheduler in Project Reactor?
Is there a "default" scheduler in the Project Reactor? Which one is it? By "default" I mean the one used when no subscribeOn() nor publishOn() are called for the chain.
0
votes
0
answers
24
views
Is there a step that will not cause an error? bufferTimeout: "Could not emit buffer due to lack of requests"
API: Exception - reactor-code
The example works as follows:
Subscribes to the next one on departure. The incoming data comes from the Rabbit and it will be processed. This can take a relatively long ...
0
votes
1
answer
53
views
Why reactor does not process each element concurrently?
What I expect is reactor will create threads for each emitted element, meaning elements 1, 2, 3, 4, 5 from the source should be handled concurrently. But it's not true from my demo code output, but I ...
0
votes
1
answer
20
views
guaranteed order of delivery in reactor?
Does project-reactor guarantee the order in which a publisher delivers things to subscriber(s) by default? If not, is there a way to guarantee the order in which a publisher delivers things to ...