ReactiveX/RxJava

RxJava – Reactive Extensions for the JVM – a library for composing asynchronous and event-based programs using observable sequences for the Java VM.

Oleh Dokuka

@OlegDokuka

In case if your requirements sounds like I would like to execute all my Singles one by one in order in which they were passed you should refer to next example

<T> Flowable whenAll(List<Single<T>> singles) {
    return Single.concat(singles);
}

Phoenix

@wbinarytree

Hi is there an operator can ignore the onComplete from upstream? for example i use this

Observable.flatmap{
    Observable.just(it)
                            .onErrorReturn{error ->ErrorState(error)}
}

it won't complete when a error occur

Oleh Dokuka

@OlegDokuka

Unfortunatelly, as far as I know, there is no such operator out of the box, but in Reactor 3 this feature will be included in the next 3.2.x release

Phoenix

@wbinarytree

Thanks ! what's that operator called then? onErrorReplace or something?

Oleh Dokuka

@OlegDokuka

https://github.com/reactor/reactor-core/pull/840/files#diff-12cdd642dc76cc8d31d1d9605e8cef0fR3725

actually it will be called as .errorStrategyContinue, but you may join the discussion and propose and additional resume operator with fallback -> reactor/reactor-core#629

Phoenix

@wbinarytree

i will keep my flatMap then. Thanks . I will check it out in Project Reactor

Oleh Dokuka

@OlegDokuka

:+1:

Phoenix

@wbinarytree

Another question. Is there any operator is equivalent andThen for Completable in Observable? For example i use this really often

API.login
      .doOnNext(//save the login datas)
      .ignoreElements()
      .andThen(//other observable Calls)

I think concatWith is somehow the good one but it need downstream be the same type.

David Karnok

@akarnokd

Use flatMap to continue with some other type.

API.login
    .doOnNext(loginData -> { /* ... */ })
    .flatMap(loginData -> API.fetchFavorites(loginData.userId))

Phoenix

@wbinarytree

But it's not guarantee to be called after complete no? Also For example maybe the upstream emit more than one item(instead of login). But ignoreElements().anThen() will called only once after the upstream complete.

Oleh Dokuka

@OlegDokuka

use .concateMap instead. It guaranties that each call will be executed in order as they come,

Phoenix

@wbinarytree

~~yes. Then back to the problem of the same signature in concatMap~~

Oleh Dokuka

@OlegDokuka

so if you make a call to external API concate guaraties that the next call will not be executed until previois has been completed

Phoenix

@wbinarytree

I think concatMap is not the same. .ignoreElements().anThen() will execute only once no matter what number of item emitted by the upstream. but concatMap will map all element to the given transform function

Oleh Dokuka

@OlegDokuka

Yeah, you are right.

Mark Galea

@cloudmark

@akarnokd I'm reading the Rx Code and your blog and I have a question about the merge operator. Why is the main queue an SPSC instance (rather than MPSC) give that inner subscribers can call the parent.tryEmit(this, t); concurrently. Or is this impossible?

David Karnok

@akarnokd

That SPSC queue is for the the main Observables that turn out to be of scalar origin, such as just. Since these come through onNext, there is guarantted to be one producer of them. The drain will then read from this SPSC queue and the queues of the inner Subscribers from a guaranteed single thread at a time.

Mark Galea

@cloudmark

This message was deleted

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 ?

Mark Galea

@cloudmark

Why are scalars prioritised over inners in the emitLoop? From the code it seems that the main queue is dequeued first.

David Karnok

@akarnokd

Yes. There are users who flatMap onto just() and empty() regularly and this shortcut helped their performance greatly.

Mark Galea

@cloudmark

Thanks a lot!

Gordeev Boris

@gordeevbr

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!

David Karnok

@akarnokd

@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.

Gordeev Boris

@gordeevbr

Yeah, that's why I didn't submit a PR.

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.

Gordeev Boris

@gordeevbr

That's why I asked if someone could give it a look.

Gordeev Boris

@gordeevbr

Oh wait, I'm totally wrong.

Yeah, that won't work it seems.

I have read through this again.

http://kto.so/2017/09/30/the-practical-truth-about-multi-release-jars/

Sorry for wasting your time.

Sascha Held

@faken

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

Rafael Guillen

@rguillens

@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

mreddimasi

@mreddimasi

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 ?

mreddimasi

@mreddimasi

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()),

Troy Patrick

@teasp00n

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?

pbagchi

@pbagchi

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

Rafael Guillen

@rguillens

@teasp00n I recommend you wrapping the scanned String with something like Optional, this way you can send Optional.empty() emissions when room database map fails.

Rafael Guillen

@rguillens

@mreddimasi Read about and try with mergeDelayError operator.

Where communities thrive

People

Repo info

Activity