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
1 view

Reactor Kafka Graceful Shutdown and Pause Consumer

Spring Application uses Reactor Kafka to consume messages. Question 1: Is there standard convention to pause message consumption and finish processing inflight messages during application shutdown? ...
0 votes
1 answer
15 views

How to write filter for a reactive Spring Integration flow?

I'm using RSocket as InboundGateway in my integration flow, using interaction model REQUEST_CHANNEL. It emits a Flux of my business objects. I am able to "decorate" these objects using ...
  • 631
1 vote
0 answers
13 views

the webflux to consume server sent events have no valid response

I want to use webflux to consume some server sent events in Java 11, this is the minimal runnable code looks like: public static void main(String[] args) { consumeServerSentEvent(); } ...
  • 23.7k
0 votes
0 answers
15 views

Spring WebFlux Controller default pools

I need advice, there are two controller methods: Method with body parameter @RequestMapping(value = "/test1", method = RequestMethod.POST) public Mono<SuccessEmptyResponse> ...
  • 399
1 vote
0 answers
29 views

Spring Webflux : Not able to start kafka consumer

When I try to start my application connecting to my local kafka broker everything is working fine. But when I try to connect to my hosted broker using ssl authentication, I am getting following error :...
  • 3,561
0 votes
1 answer
21 views

Custom Method Security Expression Evaluator for Spring Boot 3 Webflux Application

I'd like to use method security annotations and a custom expression evaluator in my Spring Boot 3 Webflux application, and I'm not sure how to do it. My endpoint with its method security annotation ...
  • 4,692
0 votes
0 answers
50 views

Execute two Monos in parallel but processing them individually

I am trying to execute two Monos in parallel(like the code below) but process each Mono individually with doOnNext for each Mono. I have tried to use Mono.zip but it is not useful for my case because ...
  • 1,639
0 votes
1 answer
25 views

Can I reuse Mono in case of unexpected error

I have the following code Mono<FileInfo> fileInfoMono = getFileInfo(fileName, filelocation); return fileInfoMono.map(info -> fileReader.read(info) .doOnNext(this::log) ...
  • 341
0 votes
0 answers
21 views

Reactor Mono in Spring Webflux evaluates to empty

Suppose I have the following endpoint to create a user in my Spring Webflux web service: @PostMapping("/user") public Mono<ResponseEntity<UserResponse>> userCreate(Mono<...
  • 4,692
0 votes
0 answers
28 views

"Unsupported array type" while saving object with one to many relation

I have two entities User and UserAuthority with one to many relation like following @Getter @Setter @Table(name = "users") @Entity public class User extends BaseEntity { @OneToMany(...
0 votes
1 answer
21 views

Download a file using Spring WebClient and store it on S3 using S3AsyncClient

I want to download a file using WebClient and put it on S3. The problem I face is that I need to have a "Content-Lenght" value in order to pass it to S3. I can get it from WebClient, but I ...
  • 24.5k
0 votes
1 answer
24 views

Call a method right before start consuming Flux elements

How to call a non-reactive blocking method in the API pipeline, but call it only if there is at least one item present in the Flux? For example, before calling itemsRepo.saveItem, I want to call ...
0 votes
0 answers
24 views

How to gracefully cancel/dispose in-flight items in project Reactor?

According to the docs of project reactor, for a Flux or Mono, cancellation is a signal that the source should stop producing elements. However, it is NOT guaranteed to be immediate: Some sources might ...
  • 2,764
0 votes
0 answers
21 views

Is there a way to know if the retry was successful or not in the doAfterRetry method in Spring webflux?

yThe failure attribute of the RetrySignal seems to hold the original exception, and is not cleared upon successful retry. Is success at all possible to detect in this method or is there another way? ...
0 votes
0 answers
14 views

A public reactive MockAPI like jsonplaceholder.typicode to return Flux

Is there a Mock public Api website that returns a reactive response. Like a Stream of Items instead of a List of items. like we have jsonplaceholder.typicode.com which is non reactive version
0 votes
0 answers
20 views

how to call outer doOnNext of kafka flux from the new flux returned from onErrorResume()

I am using Reactive Kafka with ReactiveKafkaConsumerTemplate class as below: @Autowired private ReactiveKafkaConsumerTemplate<String, MyData> consumerTemplate; @EventListener(...
0 votes
1 answer
36 views

what is the right functional way to iterate over a Flux of N elements twice with Spring WebFlux library without hitting the db twice?

I'm new to Spring Reactor, and I'm not sure what is the right way to do the following: I have two reactive repository of type MyEntity and SumEntity, repo and sumRepo. I need to do the following in ...
  • 1,946
0 votes
1 answer
40 views

Dynamic Backoff duration in RetryBackoffSpec in spring webflux

Mono.fromCallable(() -> sendRequestToServer(params)) .doOnSuccess(result -> handleResponse(result, grantMapEntry, deviceInfo)) .retryWhen(RetryBackoffSpec .backoff(5, ...
  • 802
0 votes
0 answers
48 views

Blocking in reactor thread for validation

Spring boot 3.0.2 - I want to block in reactor thread to validate an event. Unless the event is validated , I do not want to move ahead. How do we achieve it ? .block() -> throws exception , so it ...
1 vote
2 answers
51 views

How to get the data in Flux onError

The question is not specific to database, but depicted it with the database example I fetch some entities from database with a reactive driver Convert them in Item entity using itemMapper. save them ...
-1 votes
0 answers
35 views

Java Webflux - how to trigger an action but not to wait for the completion

How to trigger an action but not to wait for the completion and insead just go to the next step? For example, I have a service which can send notifications. I want to call it, then trigger saving the ...
0 votes
1 answer
42 views

Reactor Mono zip+map vs flatMap/Map

when calling three different services and waiting for the response in parallel, such as this: Mono<A> mono1 = service1.call().scheduleOn(Schedulers.parallel()); Mono<B> mono2 = service2....
  • 1,886
0 votes
0 answers
60 views

Error occurs java.lang.NoSuchMethodError: reactor.core.publisher.MonoSink.contextView() after updating Quarkus version

I have updated the quarkus version of the project from 2.13.final to 2.16.4 final. Then it gives an error as follows. java.lang.NoSuchMethodError: reactor.core.publisher.MonoSink.contextView() The ...
1 vote
0 answers
23 views

Spring Boot, reactive streams hot publisher and GraphQL subscriptions. Doesn't work as expected

I'm using spring boot v3.0.2 with websockets and faced problem with graphQL subscriptions. The task is pretty simple - notify client when backend entity changes. So, I have EntityChangeNotifier - @...
  • 107
0 votes
1 answer
25 views

Java Reactor equivalent of an ArrayBlockingQueue

Imagine the following pattern: execute(): The execute() method is called once and a reactor code scans the entire directory, does parse files and puts the result into a blocking queue. fetch(): The ...
-1 votes
1 answer
27 views

execute 2 Mono sequentialy not supporte blocking

i have the two bellow functions Mono<Void> messageEvent(MessageEvent messageEvent); Mono<Victim> getPersonById(String personId); i want when the function messageEvent is executed ...
0 votes
1 answer
54 views

How to properly generate a Flux from a function that supplies a Mono?

I want to write a piece of code that would iterate over a database reactively (the next query depends on the result of a previous query). My question could be simplified as follows: Lets say we have a ...
0 votes
2 answers
86 views

How to return Flux of objects after calling a mono boolean method

I have a validator method that returns a Mono<Boolean>. The validation is happening as expected and returns bad request exception in case of invalid id1 or id2. public static Mono<Boolean> ...
  • 3,041
0 votes
1 answer
59 views

Handling partial failures in case of Mono zip

I am building a reactive Java application in which I'm creating a zip of three different functions, say funA(), funB() and funC() in the following way Mono<Type> combined = Mono.zip(funA(), funB(...
0 votes
0 answers
39 views

Spring Webflux Returning Mono which will be populated asynchronously

I am trying to return a response to a http GET call which will be filled asynchronously (via a MQ). IE user calls the /hello url, the url fires off a request to some back-end system asynchronously, ...
  • 11
0 votes
1 answer
21 views

How to combinate the Reactor operation that simplify the code struct which like if or else?

Summary I am learning about the reactive programming framework that Reactor. I had read the offical operation document with a simple cogition and when i try to do some practices , i meet the ...
  • 23
2 votes
2 answers
101 views

SubscribeOn uses only 1 thread from the pool

There are many questions on subscribeOn and in theory it is simple: it makes the subscriber process run with specific Scheduler. However when I run this: @Test void fluxWithSubscribeOnTest() { ...
0 votes
1 answer
12 views

what the difference between handle and create?

Article background i am trying to figure out that the difference betweenhandleand create; i had refer to the reactor offical documents , but the sample which offical provided , could not point me ...
  • 23
0 votes
2 answers
40 views

Connecting a Flux to a Mono

I have a Flux<T> that returns multiple Ts. I need to wait until all the items are emitted from my Flux before continuing on to the Mono. When I was doing similarly with just Monos, I was able to ...
  • 5,893
0 votes
0 answers
30 views

Reactor Netty IdleStateHandler as a keepalive mechanism breaks http2 connections

I'm using Spring Webflux for sending Server Sent Events over HTTP: @GetMapping("/dummysse") public Flux<ServerSentEvent<UpdateResponse>> handleSse(@RequestParam(required = ...
0 votes
2 answers
48 views

How to include a total count of items when doing pagination with skip() and take() in Webflux?

I am trying to implement a pagination in Webflux. Let's say CustomersService.getCustomers() is returning a Flux<Customer>. public Mono<CustomersResponse> getCustomers(Integer page, Integer ...
  • 15
0 votes
0 answers
42 views

Java card polling reader with reactor

I've built the following function to execute a CardTerminal for reading the ICCID of a SIM Cards. Goal: Terminal Polling and wait for a CardTerminal connect Card Polling and wait for a Card connect ...
  • 3
0 votes
0 answers
18 views

"exhaustMap" in Project Reactor?

Is there an equivalent operator in Reactor for exhaustMap from RxJs (https://rxjs.dev/api/operators/exhaustMap)? A function that returns an Observable containing projected Observables of each item of ...
1 vote
2 answers
32 views

Reactor subscriber not executed in the specified scheduler

I'm starting with java reactor and facing an issue when testing a code that is querying couchbase and processing result documents. My code looks like that Flux<String> documentIds = ...
  • 793
0 votes
1 answer
42 views

How to control the kafka connection retries in Reactor Kafka/Reactive Kafka

I am using Reactor-Kafka in spring boot application. I am trying to handle the transient errors where If kafka goes down then I want to keep retrying to connect to kafka indefinitely. I have observed ...
0 votes
1 answer
49 views

Spring ReactiveCrudRepository @Query with PostgreSql throws PostgresqlBadGrammarException: [42P01]

Given a ReactiveCrudRepository with specified @Query which passes unit test flawlessly using MariaDB backend but throws exception using PostgreSQL. Other Spring repository queries succeeds (findById, ...
  • 329
0 votes
1 answer
26 views

How to test buffer in webflux

I want to write a unit test for my below gist private Flux<String> replayWithData(final Replay replay, final Checkpoint checkpoint) { return service.findAll(replay.getData()) ...
  • 1,119
1 vote
2 answers
58 views

How can I work with specific version of dependency in maven?

I have a project in springboot 3, I want to consume some service from internet, I implemented a webclient, here my webclien definition @Bean public WebClient createWebClient(@Value("${...
  • 137
1 vote
0 answers
23 views

onNext and onComplete not called when using anonymous subscriber

I have the following code Flux<String> myFlux=getFlux(); documentIds .windowTimeout(WINDOW, Duration.ofSeconds(5)) .onBackpressureBuffer() .takeWhile(d -> !...
  • 793
0 votes
1 answer
39 views

Execute Monos sequentially/parallel

I have this situation: Mono<List<String>> resultAMono = listA(); Mono<Void> resultBMono = loadB(resultAMono.block()); Mono<Void> resultCMono = loadC(); I want to execute ...
0 votes
0 answers
26 views

return a flux inside a mono

I have the below code, in which i want to return a flux inside a mono, but i have this error no instance(s) of type variable(s) R exist so that Flux conforms to Mono<? extends R> return ...
0 votes
0 answers
6 views

How to build a distributed service with reacty-netty that could get the userInfo and the connection which distributed in different service?

as use the reactor-netty to build a message proxy server , i meet the confuse that i have no idea about how to slove userinfo may caught by different machine when i use distributed service . i ...
  • 23
0 votes
1 answer
29 views

Sequential calls with reactor-netty http client seem to free ByteBuf of the first response and causes IllegalReferenceCountException

I have a requirement to either make two http calls in sequence with one being dependent on another AND in another case to make two parallel calls and combine their results. For Case1: Sequential calls ...
  • 1
0 votes
0 answers
34 views

List mapping in Webflux: map() + stream() vs. flatMapMany() + map() + collectList()

When working with Lists in Webflux, there are multiple ways how to iterate over nested lists. Let's say we have a class Customer: public class Customer { public String name; public List<...
  • 15
1 vote
0 answers
45 views

Micronaut Data MonogDB Reactive Transactions

I am looking for guidance on implementation transactions when using Micronaut Data MongoDB (Reactive) Below I have given code snippets from what I am trying to implement: Data dependencies included: ...

1
2 3 4 5
61