RxJava – Reactive Extensions for the JVM – a library for composing asynchronous and event-based programs using observable sequences for the Java VM.
Ah gotcha! So the main queue is only responsible for scalars from the following line:
if (t instanceof ScalarSynchronousObservable) {
tryEmit(((ScalarSynchronousObservable<? extends T>)t).get());
}Each inner will have its own queue and not use the main. @akarnokd ?
Greetings! So I wanted to try out Java 9 MrJars. Since in DESIGN.md it says "we want to use the multi-release jar file support" I have decided to practice MrJars by attempting to migrate RxJava. I have succeeded in making a gradle script that produces seemingly valid MrJar with some limitations. It seems to be that there's no way to use this MrJar yet with gradle, which means that this is probably not a production ready solution. I have decided not to submit a PR yet, but I would be glad if someone takes a look at changes in my repo.
Is it going to work if I just provide a link? https://github.com/gordeevbr/RxJava
So there are old build scripts (jar, sourcesJar, javadocJar) and new ones are mrJar, sourcesMrJar and javadocMrJar. I have tested them out and everything seems to be in order.
If I just include it into my classpath with gradle using
flatDir, I can't seem to use the Java 9 part.
Thanks in advance!
@gordeevbr Hi. Please don't submit a PR for multi-release jar. There is currently no value in having twice the JAR size and the duplication of the source files. In addition, you can't add new public API (such as Java 9+ features) via mr jars anyway and I'm not sure the build and release infrastructure could even handle mr jars properly.
I have made it so that there's no duplication of source files, the new API "extends" the old one, even though it does so in a not obvious manner.
It seems to me that Gradle can handle creating an MrJar, but I think there's no way to use these Jars yet. That's why I gave up on it right now.
Yeah, that won't work it seems.
I have read through this again.
Sorry for wasting your time.
Hey folks, i'am currently totally lost on a RxJava / functional-programing issue. Still new to all of this. I've documented my issue in this Stackoverflow question: https://stackoverflow.com/questions/48341012/rxjava-conditional-execution-of-observables-null-handling
Would be super awesome if maybe somebody could have a look
@faken You shoud read this article: http://blog.danlew.net/2015/06/22/loading-data-from-multiple-sources-with-rxjava/ and review the code example: https://github.com/dlew/rxjava-multiple-sources-sample
Hi,
Was wondering if the community could help me on how I could parallelize the 3 calls shown in the code below?
Observable<CheckoutDetailsDTO> o = Observable.zip(
orderIntegrationService.getRxOrderByOrderId(orderId),
shippingIntegrationService.getRxAddressListForUser(),
paymentIntegrationService.getRxPaymentInfoForUser(),
(order, addresslist, paymntinfo) -> new CheckoutDetailsDTO(order, addresslist, paymntinfo));The calls to order, shipping and payment happen to be called one after another in the code above, is there any way it can be parallelized as the 3 calls are independent of each other ?
I've tried the following to parallelize, but still see that its been called one after another
orderIntegrationService.getRxOrderByOrderId(orderId).subscribeOn(Schedulers.io()),
shippingIntegrationService.getRxAddressListForUser().subscribeOn(Schedulers.io()),
paymentIntegrationService.getRxPaymentInfoForUser().subscribeOn(Schedulers.io()),
i have a
BehaviorSubject that emits strings every time a barcode is scanned, i then want to try map this to a product in my room database. i also want to know if it failed to map so i cannot useObservable as 'empty' values are filtered out so i changed the sig to use Maybe<T>. Is there anything special i need to do to flatmap from my Observable to the Maybe?
Hi, I am working on a project using Hystrix and gRPC. I have a HystrixObservableCommand class - MemberCommand and I am subscribing using Schedulers.io()
memberCommand.toObservable().subscribeOn(Schedulers.io());In the construct method, I log the gRPC context but it returns null -
protected Observable<Greeter.GreeterReply> construct() {
logger.info("0 -- In MemberCommand, traceId from gRPC context = {}", ContextKeys.TRACE_ID_CTX_KEY.get());Any ideas, how I can propagate gRPC context through RxIoScheduler threads?
For the above I get,
0 -- In MemberCommand, traceId from gRPC context = null
@pbagchi Ask here: https://gitter.im/Netflix/Hystrix
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 );