RxJava – Reactive Extensions for the JVM – a library for composing asynchronous and event-based programs using observable sequences for the Java VM.
Hi, for the question I had above, I am trying around with RxJavaHooks. Can somebody point me to what the Runnable is inside the call() method?
RxJavaHooks.setOnIOScheduler(new Func1<Scheduler, Scheduler>() {
@Override
public Scheduler call(Scheduler scheduler) {
//context.attach();
return scheduler;
}
});
rx.Observable<Greeter.GreeterReply> memberReplyObservable = memberCommand.toObservable().subscribeOn(Schedulers.io());The version of RxJava Hystrix is using is 1.2.0.
You can hook into the task scheduling via setOnScheduleAction. Get the current context, create a new Runnable which restores that context and calls the original Runnable.
@akarnokd - So, I did this
rx.Observable<Greeter.GreeterReply> worldReplyObservable = worldCommand.observe().subscribeOn(Schedulers.computation());
rx.Observable<Greeter.GreeterReply> teamReplyObservable = teamCommand.observe().subscribeOn(Schedulers.computation());
Context context = Context.current();
RxJavaHooks.setOnScheduleAction(new Func1<Action0, Action0>() {
@Override
public Action0 call(Action0 action0) {
context.attach();
logger.info("traceId in gRPC context = {} ************", ContextKeys.TRACE_ID_CTX_KEY.get());
Runnable r = new Runnable() {
@Override
public void run() {
context.attach();
logger.info("**************** traceId in gRPC context = {}", ContextKeys.TRACE_ID_CTX_KEY.get());
}
};
r.run();
return action0;
}
});
rx.Observable<Greeter.GreeterReply> memberReplyObservable = memberCommand.toObservable().subscribeOn(Schedulers.io());Going into an endless loop because return action0; is eventually calling action0 again. What am I doing wrong?
I have three Observables - two on computation scheduler and one on IO. As far as I understand setOnScheduleAction will get called for all three which isn't a problem per se. But I am seeing RxComputationScheduler threads which have the gRPC context but then one RxIOScheduler thread which still doesn't have the context.
Runnable r = new Runnable() {
@Override
public void run() {
context.attach();
logger.info("**************** traceId in gRPC context = {}", ContextKeys.TRACE_ID_CTX_KEY.get());
action0.run(); // <--------------------------------------------------------------------------------------------
}
};
return r; // <------------------------------------------------------------------------------@akarnokd - Sorry, for bugging, but Action0 does not implement Runnable so I can't return r or do action0.run()
RxJavaHooks.setOnScheduleAction(new Func1<Action0, Action0>() {
@Override
public Action0 call(Action0 action0) {
context.attach();
logger.info("traceId in gRPC context = {} ************", ContextKeys.TRACE_ID_CTX_KEY.get());
Runnable r = new Runnable() {
@Override
public void run() {
context.attach();
logger.info("**************** traceId in gRPC context = {}", ContextKeys.TRACE_ID_CTX_KEY.get());
action0.run();
}
};
return r;
}
});I can do action0.call() but that would be an infinite loop.
Standalone, sorry I don't understand? I have three services that I am reaching out to. Two on computation scheduler and one on IO.
I would like them all scheduled independently and in parallel not waiting on each other.
I would like them all scheduled independently and in parallel not waiting on each other.
rx.Observable<Greeter.GreeterReply> worldReplyObservable = worldCommand.observe().subscribeOn(Schedulers.computation());
rx.Observable<Greeter.GreeterReply> teamReplyObservable = teamCommand.observe().subscribeOn(Schedulers.computation());
rx.Observable<Greeter.GreeterReply> memberReplyObservable = memberCommand.toObservable().subscribeOn(Schedulers.io());
Observable.zip(worldReplyObservable, teamReplyObservable, memberReplyObservable, new Func3<Greeter.GreeterReply, Greeter.GreeterReply, Greeter.GreeterReply, Object>() {
@Override
public Object call(Greeter.GreeterReply worldReply, Greeter.GreeterReply teamReply, Greeter.GreeterReply memberReply) {
Your "infinite loop" comment doesn't make sense to me. Whenever a task needs to be scheduled, that hook body will be executed as well as the
Action0 wrapper. Providing standalone means code that captures the essence of your flows and uses only plain JDK types and RxJava so I can try it and see what could be wrong.
I have a blog post about writing a scheduler that copies thread-local data around.
Hi,
I saw this "After an Observer calls an Observable's subscribe method..." and was wondering how an Observer can call an Observable's subscribe method.
Is that sentence correct?
Can anyone help with what I'm doing wrong here? I'm trying to have the messages in the messages Observable pop out at random intervals.. This is based off some code from stack overflow, but if I grok it right, it's trying to create one observable stream that loops that has the switchMap to cause it to wait a random amount (via the call to randomTime() that in my impl just returns '2').. and then zipWith laces the two together.. meaning each message has to wait for the next delay to pop.. except it doesn't work.. and I don't understand why...
Observable<String> delay = Observable.just("")
.switchMap(dummy -> Observable.timer(randomTime(), TimeUnit.SECONDS))
.map( a -> String.valueOf(a) )
.repeat();
Observable<String> messages = Observable.just("Test") //eventually lines from a file...
.repeat();
messages.zipWith(delay, (d, msg) -> ""+d+" "+msg ).subscribe( System.out::println );
it shouldn't quit tho right? I'm zipping across 2 infinite streams
I mean.. it's great to know it works for you.. you've no idea how nice it is to know my half brained understanding of this stuff is at least on the right tracks ;p
Flowable<String> delay = Flowable.just("")
.switchMap(dummy -> Flowable.timer(randomTime(), TimeUnit.SECONDS))
.map( a -> String.valueOf(a) )
.repeat();
Flowable<String> messages = Flowable.just("Test") //eventually lines from a file...
.repeat();
delay.zipWith(messages, (d, msg) -> ""+d+" "+msg ).subscribe( System.out::println );
So I tried that... and now it just exits immediately with no output =)
ahh.. so I need to subscribe on the current thread somehow?
so when it runs out of lines from the file, ideally I want it to start over again.. it's never supposed to exit, or rather, I'm fine exiting it by killing the process
like.. I understand that Flowable.timer( delay, unit) produces an observable that returns 0L after the specified delay
ahh.. and now I figure out why my next change didn't work either..
It might be worth reading an RxJava 2 book: https://www.packtpub.com/application-development/learning-rxjava
I'm playing with WebFlux, RxJava and Lagom all at the same time..
not in the same app, thankfully
but in parallel, trying to do the same thing in each
