Say those classes are
Client1 and Client2. I want to call each in a Mono.zip so that they can be done concurrently
I am trying to verify that they are being done concurrently by using a VirtualTime scheduler
except I cannot figure out how to create a mock delayed response to force me to add the right schedulers to my
zip method arguments
public Mono<Tuple2<String, String>> getPair() {
return Mono.zip(
client1.getFirstPart(),
client2.getFirstPart());
}
i am pretty certain i will have to schedule those client methods on their own schedulers
but i am not sure
Thanks to everyone who takes part in the development. I feel the reactor - as very beautiful, tasty reactive solution (not only for Big EE). Gorgeous Documentation Quality, and Clean Java 8 Integration. It is a pity that in Android - space - it is still perceived wildly and non-obviously)
@kota_sunilkumar_twitter
@bsideup , Hi am new to reactive prog.i watched your talk at @s1p@Austin.
There, you gave an example for errors are side effects to.
Use handle method for throwing exceptions but not throw new SomeException()..like this.
@bsideup , Can please give best usage and example for usage Flux#handle method...I want to learn
@bsideup , Hi am new to reactive prog.i watched your talk at @s1p@Austin.
There, you gave an example for errors are side effects to.
Use handle method for throwing exceptions but not throw new SomeException()..like this.
@bsideup , Can please give best usage and example for usage Flux#handle method...I want to learn
With sink.error(new SomeException())
This is my reference from your talk
what this sink.next() and sink.error() methods will do there..any basic example to better understand "sink".Please help me
think callback-based API:
function doSomething(item, callback) {
if (itemIsGood(item)) {
callback(convertItem, null);
} else {
callback(null, new Error("something is wrong with the item"));
}
}
and what is the major difference of throw new exception() vs sink.error()?
and how throw new exception() is a side effect?
and how throw new exception() is a side effect?
although
Also, as mentioned in the talk, when exception is checked, you must wrap it with some unchecked wrapper (like
Last but not least, you're making it more explicit, what error cases you handle and how you process them, instead of just throwing things around, imperative-style.
You can also use
.map will handle the error and propagate it downstream, it is an "exceptional" case, hence side-effect.Also, as mentioned in the talk, when exception is checked, you must wrap it with some unchecked wrapper (like
RuntimeException).Last but not least, you're making it more explicit, what error cases you handle and how you process them, instead of just throwing things around, imperative-style.
You can also use
sink.success() (or sink.complete() if Flux) instead of passing an error and handling it later in the same operator (in the catch block) to use less operators
I have 1 requirement...like
We are developing application with reactor but database is Oracle which blocking
Now the problem is ...while am doing 10 validations, each requires Oracle db connection.. total 10 db connections.
While hitting the API for performance testing with just 60 requests using gatling ...hikari connection pool size is 30..
We are getting unable to get db connection exceptions
Plz pullout me from this situation
We are developing application with reactor but database is Oracle which blocking
Now the problem is ...while am doing 10 validations, each requires Oracle db connection.. total 10 db connections.
While hitting the API for performance testing with just 60 requests using gatling ...hikari connection pool size is 30..
We are getting unable to get db connection exceptions
Plz pullout me from this situation
@bsideup @simonbasle
@bsideup, any suggestions on my above puzzle?
make sure that you actually use the pool and not create a new connection every time
I’m creating a hot flux which publishes parallel scheduler. A subscriber which is consuming this flux is running in parallel on a new bounded elastic scheduler. While doing some testing I set the thread cap and queue cap to 1 and program started throwing error which baffled me at first but after closely going through the documentation I figured out it’s because when no more task can be enqueued
Now I’m little confused because of I increase the queue task cap to 4-5 it works again but it’s queue should have been overflown in this case too as it’s processing 100’s of elements
I'm kinda new to reactor, I've read the basis and all but i'm a bit stuck on my implementation, I need some help
I've created an endpoint, which wraps my business endpoint with data model transformation. My new endpoint call 3 methods which returns a Mono<myobject>. The first one transform "outsideModel" to "internalModel" (through a microserviceA), then I call my business microserviceB, finally I call-again my first microserviceA to transform back from internalModel to outsideModel. As my microserviceA is stateless, I need the original data from the "outsideModel" from step 1. My 3 Mono calls are chained, and I'm troubled on how can I store/save the original "outsideModel" and reuse it when I need it for my last step. Can I simply store it in a local variable ?
Hi! Hope this is the right place to ask, would really appreacieate any input.
I have a flux instance, that I'm iterating through using toIterable. I can't find a way to cancel the underlaying generator if anything goes wrong on the consumer side, it will only close once I've iterated through enough items to exhaust it. Is there any way to trigger a cancelation event to a flux that is being consumed through toIterable()?
I have a flux instance, that I'm iterating through using toIterable. I can't find a way to cancel the underlaying generator if anything goes wrong on the consumer side, it will only close once I've iterated through enough items to exhaust it. Is there any way to trigger a cancelation event to a flux that is being consumed through toIterable()?
question about 'request' method
final Flux<Boolean> randoms = Flux.generate(sink -> sink.next(true));
randoms.filter(x -> x == true).subscribe(x ->System.out.println(x));as I know , for the above code , LambdaSubscriber will request data from FilterFuseableSubscriber , just like downstream(LambdaSubscriber) request data from upstream(FilterFuseableSubscriber) , and for FilterFuseableSubscriber it will request data from upstream, my question is :
@Override
public void request(long n) {
s.request(n);
}why the request number is the same as the downstream, if downstream request data number is 5 , for filter function , the requestion should more than 5 , since that it will filter some data , so can somebody tell me the logic :)
.onErrorResume then ?
@bsideup thank you for the response , I am trying to understand why do we have
requestOnSeparateThread flag when we have separate schedulers for blocking and non-blocking reactor calls. I have used requestOnSeparateThread in my application but couldn't see any difference on setting requestOnSeparateThread = true/false