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,134
questions
-3
votes
0answers
57 views
spring weblux and kotlin coroutines + BlockHound: reactor.blockhound.BlockingOperationError: Blocking call
I have the code:
@RestController
@RequestMapping("/v1/accounts")
class AccountController(private val accountService: AccountService) {
@Operation(
summary = "Запрос ...
0
votes
2answers
25 views
Combine all elements to a flux to mono
This is a simplified example of the problem I am facing.
Flux<String> fluxString = Flux.just("A","B","C");
Mono<String> monoString = Mono.just("Original&...
1
vote
0answers
29 views
When is Reactive programming preferred over Java Streams and vice-versa? [closed]
Here I am having two pieces of code. Both look very similar on the outside. I would like to know when each of the following are preferred
Reactor
// import reactor.core.publisher.Flux;
// import ...
0
votes
0answers
36 views
How to convert Flux<List<Object>> into Mono<List<Object>>
I have a service which returns Flux<List<Integer>> and I would like to convert it into Mono<List<Integer>> to be used inside transform()
Here is what I did by using flatMap and ...
-1
votes
0answers
7 views
How can we implement Confluent kafka Replicator using Reactor kafka Api
Actually we are using Reactor kafka to connect enterprise confluent kafka. But how can we implement Confluent kafka Replicator to Replicating Data Between Clusters.
how can we integrate timestamp-...
0
votes
0answers
33 views
Blocking operation in coroutines
I am working on webflux application using coroutines. It's main purpose is "backend for frontend" for mobile app. Most of request is handled by fetching data and merging data from ...
0
votes
0answers
16 views
Debounce similar requests with reactor-grpc
In order to offload my database, I would like to debounce similar requests in a gRPC service (say for instance that they share the same id part of the request) that serves an API which does not have ...
0
votes
0answers
35 views
Parse JSON with Spring WebFlux after error occured
I'm receiving JSON from REST API looks like:
{
"items": [
{
"id": 60659,
"name": "Display",
"active": ...
0
votes
0answers
41 views
Blocking calls in Reactive Stream
I wonder if normal java API calls (I mean methods without I/O) should be threaded as a "mini blocking calls"? Is it acceptable to implement Reactive Streams like this (invoking method before ...
-1
votes
0answers
12 views
How to read correctly kafka-topic with multiple reactor consumers?
I want to process messages in parallel between partitions of topic.
I found an example via grouping in the official reactor doc :
https://projectreactor.io/docs/kafka/release/reference/#concurrent-...
-1
votes
1answer
20 views
Mono response from a method which returns void
I have a service method which does not result anything, but can return an HttpException.
example
class Service{
public void myService() throws HttpException{
//do something
}
}
My calling class has a ...
0
votes
0answers
20 views
Continue zip(), if one source is completed
I have some troubles with .zip() operator.
Let me simplify my problem on a small example.
Flux<Integer> flux1 = Flux.just(9, 8, 3, -2);
Flux<Integer> flux2 = Flux.just(7);
Flux&...
0
votes
1answer
29 views
Mono webflux handling a specific error properly
I am using WebFlux into a micro-service and I am having issues with client. I need handle and process 404 error from client, other errors proxy.
Is this proper way to do it?
client.getSomeData()
....
0
votes
1answer
55 views
Does the thread ever change once inside a reactor execution?
It's been hammered into my head that I shouldn't use ThreadLocal with Reactor. But I want to know if I can use ThreadLocal within a single execution of a reactor function.
Specifically, when inside a ...
2
votes
1answer
55 views
Reactor groupBy: What happens with remaining items after GroupedFlux is canceled?
I need to group infinite Flux by key with high cardinality.
For example:
group key is domain url
calls to one domain should be strictly sequential (next call happens after previous one is completed)
...
1
vote
1answer
42 views
Why context is not propagated to (RxJava) Single from (Reactor) Mono?
For example, let's say I have a WebFilter that writes some Context
public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {
return chain.filter(exchange)
.contextWrite(...
1
vote
2answers
50 views
return a Mono<String> from FilePart (Java)
a class UploadService receives a FilePart, representing the content of the JSON below, from a REST API.
The target is to return a Mono<String> from this FilePart to build a JSONObject.
This is ...
0
votes
0answers
30 views
Why does my Grpc ClientInterceptor not have the right reactor Context?
I'm trying to understand the reactor Context and why I'm getting this error:
reactor.core.Exceptions$ErrorCallbackNotImplemented: java.util.NoSuchElementException: Context does not contain key
I ...
1
vote
1answer
19 views
logging on production for project reactor
i have a question related to logging on production for spring web-flux with project reactor.
I have the following code on a staging env which log a lot of info related to signals which of course is ...
0
votes
1answer
32 views
What is the Akka streams .via() equivalent in Project Reactor?
I'm kind of struggling with the Project Reactor documentation. I have some experience in Akka Streams but right now I'm working on a project that uses Project Reactor.
I need a Reactor operator that ...
-2
votes
0answers
23 views
Hooks.onEachOperator but for specific class only
Is there any instrument in Reactor to apply operator for the specific class and not for all operators in application?
Here's an example
public class MyCustomSubscriber<T> implements ...
0
votes
1answer
43 views
WebFlux using WebClient for blocking REST call
I have a situtation where I can either use blocking I/O library that access REST endpoint on outside service or I can directly call that REST with HttpClient like WebClient in this case. Now I wonder ...
1
vote
1answer
21 views
In Reactor Flux an onCancel call is not returned to calling method
I have a set of methods that handle a Flux<String> so
@GetMapping(value = "/web/{s}")
public Flux<String> getWeb(@PathVariable String s) {
return myService.getFlux(s);
}
In ...
0
votes
0answers
21 views
Regulate concurrency for reactor kafka consumer
When I increase my number of partitions in a topic, my consumers automatically increase equalling the number of partitions.
My question is how can we set the number of consumers that we want in our ...
1
vote
2answers
41 views
How to continually consume from Kafka topic using spring webflux?
I have a spring-webflux handler named WordListenerHandler which I can request a HTTP GET from the browser. This request is a stream of Long using Flux: .body(Flux.interval(Duration.ofSeconds(1)).log(),...
1
vote
1answer
25 views
project reactor calling multiple apis based on conditions and aggregating result
I am new to reactive programming and I am developing an API that aggregates the responses of 3 other APIs. The provided interfaces are as follows:
public interface UserServiceApi {
Mono<...
0
votes
1answer
42 views
Spring WebFlux endpoint performance logger
Do you know any good practices to log Spring WebFlux controller endpoint performance to keep reactive nature? Seems the following principle wouldn't work because it will block the IO as the ...
1
vote
2answers
47 views
Can I pass reactor Context into grpc clientInterceptor implicitly?
Here's what I want to do at a high-level
Capture some http headers in a WebFilter
In the Controller method, I make a grpc call
I'd like to propagate the http headers as grpc Metadata headers
...
0
votes
0answers
34 views
How can I save nested entities with Reactor?
I am using Spring with R2DBC and I need to do the following:
for (var q: quiz.quizzes()) {
Set<UUID> qsIds = new LinkedHashSet<>();
for (var qs: q.questions()) {
var ansids = ...
1
vote
1answer
39 views
How to invalidate cached Mono?
Is there a way to invalidate cached Mono? Either manually or with some condition?
Here is my case:
private Mono<String> cachedMono;
public MyService() {
this.cachedMono = cacheData()
....
1
vote
1answer
27 views
How to call switchIfEmpty when the flatMap returns an empty Mono?
My title sounds confusing so let me explain with some imperative pseudo code for what I'm trying to do
Mono<Void> func() {
Mono<MyThing> myThing = getMyThing();
if myThing is not empty:...
1
vote
0answers
17 views
Throwing exception outside of Flux/Mono reactive pipeline
In my spring JMS listener onMessage method, I am using a reactive pipeline as as below where I receive a JMS message from a queue and transform that to a different format and publishing to another ...
-1
votes
0answers
36 views
Spring Reactor Mono.zip loosing RequestScope variable
I have a springboot web application
Where there is a @RequestScope bean which hold the logged in user data.
Sample Working Code
@Component
@Scope(value = "request",proxyMode = ...
1
vote
1answer
21 views
move from imperative try-with-resource to reactive using, using()
i'm trying to move from imperative try-with-resources to reactive try-with-resources without success. I have the following piece of code i would like to move.
private final AmazonS3 amazonS3;
...
0
votes
0answers
47 views
What is the benefit of creating local variables in methods that are just copies of instance variables?
In some library implementations people use to create a lot of local variables that directly relate to instance variables. I don't really get the point. Maybe someone could shine some light in my ...
1
vote
1answer
39 views
Is it possible to create a dynamic filter/predicate in Mono/Flux?
The app is a simple processing - reading data, filtering it, process and write filtered data.
Here is a simple code which runs without any issue:
void run() {
Flux.interval(Duration.ofMillis(200)...
0
votes
1answer
28 views
Mixing two Mono and return first one with second in firsts body
I am facing a situation where I have to use 2 Mono where second one will be dependent on an Id field of first one and return the response of second one in the body of first Mono.
For example :
Mono<...
0
votes
3answers
103 views
How to use Spring WebClient to make non-blocking calls and send email after all calls complete?
I'm using Spring's 'WebClient` and project reactor to make non-blocking calls to a list of URLs. My requirements are:
Asynchronously call GET on a list of URLs
Log the URL when each URL is called
Log ...
1
vote
1answer
25 views
Spring WebFlux using RSocket: Kotlin coroutines Flow vs Reactor Flux message format
Is there any difference in IO traffic when we use Spring WebFlux's RSocket (over WebSockets) support to emit values using Kotlin Coroutines Flow and Reactor Flux?
@MessageMapping("stream")
...
0
votes
1answer
24 views
How do I conditionally return from a Mono within a WebFilter?
Not sure I'm asking this right so here's an example of what I want to do (in imperative style with blocking)
public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {
Mono&...
1
vote
1answer
44 views
How to create a Mono<Void> that actually emits an element?
So I want to create a Mono<Void> (or any Mono<SomeIgnorableType>) that actually emits an element.
Why? Because I actually want to run an effect, and the result just implies the effect was ...
1
vote
1answer
19 views
Spring Integration: How to consume QueueChannel on-demand as backpressure aware reactive (Flux) pipeline
I have Spring Integration QueueChannel that I want to consume in backpressure aware Flux pipeline.
Prefetch n Messages from the queue.
Make async calls to external system like fun remoteCall(message: ...
1
vote
1answer
44 views
Alternative to groupBy operator in Reactor
This is sort of followup question to this question. Proposed solution in answers is to use the groupBy operator. This is generally fine, but as mentioned in its docs, it is not recommended to use with ...
1
vote
1answer
27 views
How to use retry and repeat combined in Reactor?
I'm polling the queue messages, and I would like to do retry if some error happens or repeat if I don't receive any messages. The total count of retries and repeats should be equal to numRetries. What'...
0
votes
0answers
30 views
how to perform two operations on the same flux [duplicate]
I have a project reactor Flux. I want to perform two different operation on the same flux. I thought of doing the below way. But this is happening in sequential way. Is there any better way I can get ...
1
vote
1answer
24 views
Defining Two Workflows Using Using .filterWhen
I have an async predicate that checks whether a document exists in my DB. Based on whether it's present, I have two workflows. I want to achieve this in a single reactive pipeline:
Flux.fromIterable(...
0
votes
1answer
38 views
How to throw exception or handle filter condition in webflux?
1 customerRepo.findById(customerId) // finding the Customer
2 .switchIfEmpty(Mono.defer(
() -> Mono.error(new RuntimeException("not found")))) // if ...
0
votes
2answers
32 views
Testing Spring webflux complain about “Unfinished Step Verifier”. Ho finish the test properly and fully?
I am using SpringWebflux to playground some tests.
I am getting one highlight in IDE, saing "Unfinished Step Verifier"
Thats my code:
@Test
public void findAllNextMatches() {
...
1
vote
1answer
23 views
How to understand the difference between reactor.util.retry.Retry#backoff and reactor.util.retry.Retry#fixedDelay?
i tried the fixedDelay and backoff,is seems to be same... so what`s the difference of Them and the usage of them.
static AtomicInteger retryTime = new AtomicInteger();
public static void main(...
1
vote
0answers
26 views
'producer' type is unknown to ReactiveAdapterRegistry (WebFlux handler)
I have the following route in my application
@Bean
public RouterFunction<ServerResponse> route(UserHandler handler) {
return RouterFunctions.route(RequestPredicates.POST("/users"), ...