Examples with Interop

We need some examples for the README to demonstrate usage.

Perhaps we can add unit test examples with dependencies on other libraries to demonstrate interop?

@smaldini

/cc @smaldini

I’m currently looking into trying out some interop with Akka Streams, are artifacts published somewhere?

You’ll have to hold off. It’s not spec compliant yet. Once it is I’ll get some artefacts published.

Bummer; I guess I’ll look into ratpack and Reactor then (want to show something at conferences next week).

I’m actively working on this right now. Should have it working in the next day or two.

Cool, would you please ping this issue when I should try it out? Thanks!

@benjchristensen

What I've just pushed should be compliant and I think is good enough for an initial release. There's probably bugs and there are certainly performance problems, but I think it can go out.

@benjchristensen how do we get a binary out?

I'll push the buttons (since we haven't yet got the fully self-serve Travis based release process working yet like RxScala).

@alkemist

0.2.0 is released on BinTray (https://bintray.com/reactivex/RxJava/RxJavaReactiveStreams/0.2.0/view) and is making its way to Maven Central.

@alkemist Thank you for your work on this. Would you mind providing a section on the README or in the Wiki with basic usage examples?

@alkemist

Looking forward to it for my next demos :D

On Wed, Oct 29, 2014 at 5:23 PM, Ben Christensen notifications@github.com
wrote:

0.2.0 is released on BinTray (
https://bintray.com/reactivex/RxJava/RxJavaReactiveStreams/0.2.0/view)
and is making its way to Maven Central.

@alkemist https://github.com/alkemist Thank you for your work on this.
Would you mind providing a section on the README or in the Wiki with basic
usage examples?

—
Reply to this email directly or view it on GitHub
#5 (comment)
.

Stéphane

It can be seen on Maven Central now: http://repo1.maven.org/maven2/io/reactivex/rxjava-reactive-streams/0.2.0/

@benjchristensen

Thanks a lot, @benjchristensen and @alkemist!

Pushed one example of interop with Ratpack: https://github.com/ReactiveX/RxJavaReactiveStreams/blob/0.x/examples/ratpack/src/test/java/rx/reactivestreams/example/ratpack/RatpackExamples.java#L46-46

Leaving this ticket open for more examples and some stuff in the README.

Cool, nice to see that code, that's very helpful.

@alkemist

@alkemist I’m trying to extract the sample code you linked to (since I cannot figure out how to run the tests you point to). The problem I am facing is that I cannot figure out how to publish ratpack 0.9.10-SNAPSHOT locally, what is the magic incantation? I tried adding the maven-publish plugin but that does not do anything when I say ./gradlew publishToMavenLocal (and ratpack-rx:publishToMavenLocal does not exist).

@rkuhn

@rkuhn Are you okay with me using the following slide? Is there anything you'd like me to change?

If you can provide me a more realistic example I'd happily change. For example an Akka Actor that acts as a source to RxJava for consumption.

@smaldini

@smaldini I can't find a logo for Reactor, is there something I should use?

@alkemist

@alkemist Is there an example you'd like me to show in my presentation at QCon? Are you close to publishing to Maven Central a ratpack-rx version that supports RxJava 1.0? I tried with 0.9.9 but that isn't working (only 17 more days until 1.0 Final and no more breaking changes!).

@benjchristensen

@benjchristensen 0.9.10 was released today. It's in Jcenter but a successful sync to Central is proving elusive. Regarding what to demo, I just pushed examples for SSE and WebSocket streaming using publishers. Chunks, SSE and WebSocket are really the only useful ways we leverage RS streams ATM.

Up to you what you want to show.

@benjchristensen

@benjchristensen The Akka part of the sample is fine, you can make it arbitrarily complex with all the operators out there.

The problem you are seeing very likely is that Akka streams buffer less elements, exposing the deadlock in your sample code: following a groupBy with a flatMap can only work under very specific circumstances, in general it is not a safe combination under bounded memory processing (since the even numbers will need to be buffered for as long as it takes to finish all the odd ones).

Concerning the exception itself: looking at the lines referenced in the stack trace it seems that data are just pushed too eagerly within the part leading up to observeOn, either by not respecting the signaled demand or by observeOn signaling more than it can take in. Do the RxJava or Reactor versions also work if you insert a Thread.sleep(1) within the map operation?

@alkemist

@alkemist Thanks, I'll try that.

@rkuhn

@rkuhn

very likely is that Akka streams buffer less elements

How much it buffers shouldn't matter, as request(n) will determine how much it requests. One impl should be able to request(1) and another request(Long.MAX_VALUE) and it work.

exposing the deadlock in your sample code: following a groupBy with a flatMap can only work under very specific circumstances

What do you mean by this? The groupBy and flatMap uses cases work and correctly slow down (i.e. apply backpressure) to whatever the slowest group is. It would only "dead lock" as you call it if you completely blocked one group, which is exactly what should happen in a backpressure case. The user is free to decouple fast and slow streams if they wish by using a multitude of different operators that will make it behave differently (such as sample, throttle, onBackpressureBuffer, onBackpressureDrop, debounce, buffer, window, etc).

it seems that data are just pushed too eagerly within the part leading up to observeOn

That's correct, hence the problem. It works with Rx+Reactor and with Rx by itself, which implies that the request(n) is not correctly being composed through when it is Rx+Akka. It may be the RxJavaReactiveStream implementation, but Rx+Reactor are working with the exact same use case.

I have not yet spent the time to debug where too much is being requested or the request is being lost.

I'll paste examples below using Thread.sleep to demonstrate that it works with RxJava by itself or with RxJava+Reactor.

Example of RxJava + Reactor where the even group takes 10ms per item and the odd group takes 1ms per item

package reactive_streams_interop;

import java.util.concurrent.atomic.AtomicInteger;

import org.reactivestreams.Publisher;

import reactor.rx.Stream;
import reactor.rx.Streams;
import rx.Observable;
import rx.RxReactiveStreams;
import rx.observables.GroupedObservable;
import rx.schedulers.Schedulers;

public class RxReactor {

    public static void main(String... args) {

        final AtomicInteger numEmitted = new AtomicInteger();

        // RxJava Observable
        Observable<GroupedObservable<Boolean, Integer>> oddAndEvenGroups = Observable.range(1, 1000000)
                .doOnNext(i -> numEmitted.incrementAndGet())
                .groupBy(i -> i % 2 == 0)
                .take(2);

        Observable<String> strings = oddAndEvenGroups.<String> flatMap(group -> {
            // schedule odd and even on different event loops
                Observable<Integer> asyncGroup = group.observeOn(Schedulers.computation()).map(i -> {
                    if (group.getKey()) {
                        // make 'even' group be slow
                        try {
                            Thread.sleep(10);
                        } catch (Exception e) {
                        }
                    } else {
                        // make 'odd' group be faster but still have some "computational cost"
                        try {
                            Thread.sleep(1);
                        } catch (Exception e) {
                        }
                    }
                    return i;
                });

                // convert to Reactive Streams Publisher
                Publisher<Integer> groupPublisher = RxReactiveStreams.toPublisher(asyncGroup);

                // Convert to Reactor Stream
                final Stream<String> linesStream = Streams.create(groupPublisher).map(i -> i + " " + group.getKey()).take(2000);

                // convert back from Reactor Stream to Rx Observable
                return RxReactiveStreams.toObservable(linesStream);
            });

        strings.toBlocking().forEach(System.out::println);
        System.out.println("Number emitted from source (should be < 6000): " + numEmitted.get());
    }
}

You end up with output such as this:

1 false
3 false
5 false
7 false
9 false
11 false
13 false
15 false
17 false
2 true
19 false
21 false
23 false
...
3990 true
3992 true
3994 true
3996 true
3998 true
4000 true
Number emitted from source (should be < 6000): 4097

RxJava by itself doing groupBy + flatMap:

package reactive_streams_interop;

import java.util.concurrent.atomic.AtomicInteger;

import rx.Observable;
import rx.observables.GroupedObservable;
import rx.schedulers.Schedulers;

public class RxJavaGroupByFlatMap {

    public static void main(String... args) {

        final AtomicInteger numEmitted = new AtomicInteger();

        // RxJava Observable
        Observable<GroupedObservable<Boolean, Integer>> oddAndEvenGroups = Observable.range(1, 1000000)
                .doOnNext(i -> numEmitted.incrementAndGet())
                .groupBy(i -> i % 2 == 0)
                .take(2);

        Observable<String> strings = oddAndEvenGroups.<String> flatMap(group -> {
            // schedule odd and even on different event loops
                return group.observeOn(Schedulers.computation()).map(i -> {
                    if (group.getKey()) {
                        // make 'even' group be slow
                        try {
                            Thread.sleep(10);
                        } catch (Exception e) {
                        }
                    } else {
                        // make 'odd' group be faster but still have some "computational cost"
                        try {
                            Thread.sleep(1);
                        } catch (Exception e) {
                        }
                    }
                    return i;
                }).map(i -> i + " " + group.getKey()).take(2000);

            });

        strings.toBlocking().forEach(System.out::println);
        System.out.println("Number emitted from source (should be < 6000): " + numEmitted.get());
    }
}

Here is one that multicasts a stream (use publish) creates 2 async/parallel streams off of it (odd and even using filtering) then zips them together with one of the streams slow.

package reactive_streams_interop;

import java.util.concurrent.atomic.AtomicInteger;

import rx.Observable;
import rx.schedulers.Schedulers;

public class RxJavaPublishZip {

    public static void main(String... args) {
        final AtomicInteger numEmitted = new AtomicInteger();

        Observable<Object> strings = Observable.range(1, 1000000).doOnNext(i -> numEmitted.incrementAndGet())
                .publish(oi -> {
                    // schedule it so we are async and need backpressure
                        Observable<String> odd = oi.observeOn(Schedulers.computation())
                                .filter(i -> i % 2 != 0).map(i -> i + "-odd").map(s -> {
                                    // make odd slow
                                        try {
                                            Thread.sleep(1);
                                        } catch (Exception e1) {
                                        }
                                        return s;
                                    });
                        Observable<String> even = oi.observeOn(Schedulers.computation())
                                .filter(i -> i % 2 == 0).map(i -> i + "-even");
                        return Observable.zip(odd, even, (o, e) -> o + " " + e + "   Thread: " + Thread.currentThread());
                    }).take(2000);

        strings.toBlocking().forEach(System.out::println);
        System.out.println("Number emitted from source (should be ~4000): " + numEmitted.get());
    }
}

Output would be like this:

1-odd 2-even   Thread: Thread[RxComputationThreadPool-3,5,main]
3-odd 4-even   Thread: Thread[RxComputationThreadPool-3,5,main]
5-odd 6-even   Thread: Thread[RxComputationThreadPool-3,5,main]
7-odd 8-even   Thread: Thread[RxComputationThreadPool-3,5,main]
9-odd 10-even   Thread: Thread[RxComputationThreadPool-3,5,main]
11-odd 12-even   Thread: Thread[RxComputationThreadPool-4,5,main]
13-odd 14-even   Thread: Thread[RxComputationThreadPool-3,5,main]
... 
3995-odd 3996-even   Thread: Thread[RxComputationThreadPool-3,5,main]
3997-odd 3998-even   Thread: Thread[RxComputationThreadPool-3,5,main]
3999-odd 4000-even   Thread: Thread[RxComputationThreadPool-3,5,main]
Number emitted from source (should be ~4000): 4864

You can see the backpressure happening because the num emitted is < 5000 in this case rather than spinning and emitting all 1000000. It is not exactly 4000 because we are async and allow buffers in observeOn to fill.

If I put take on the individual streams that it would be exact as it would propagate through, but via zip it can't do that as it splits streams, so each stream maintains its own backpressure and zip also maintains its own.

@benjchristensen

@benjchristensen do you have in mind anything more than what we have now in terms of the general pattern? Where the proposed pattern is to have more projects in examples/ integrating different libraries/frameworks.

I'd like to encourage the respective library/framework owners to contribute examples. I don't intend to do it.

Looking at the readme again, we probably do want at least one example of real interop there.

@alkemist

@alkemist The examples above are sufficient enough, but we should get them into /examples or something like that as you suggest, and probably have the simple ones presented in the README or wiki similar to how my slides give a quick intro: https://speakerdeck.com/benjchristensen/reactive-programming-with-rx-at-qconsf-2014?slide=158

@rkuhn

@rkuhn Any thoughts on the examples above? I'm interested in your perspective since you stated that groupBy and flatMap are not safe.

Sorry for the delay, I’m traveling too much this month; I’ll try to elaborate later this week, otherwise we can talk it through next week at React.

Navigation Menu

Repository navigation

Examples with Interop #5

Stéphane

17 remaining items

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Examples with Interop #5

Description

Activity

benjchristensen commented on Oct 1, 2014

rkuhn commented on Oct 29, 2014

ldaley commented on Oct 29, 2014

rkuhn commented on Oct 29, 2014

ldaley commented on Oct 29, 2014

rkuhn commented on Oct 29, 2014

ldaley commented on Oct 29, 2014

benjchristensen commented on Oct 29, 2014

benjchristensen commented on Oct 29, 2014

smaldini commented on Oct 29, 2014

Stéphane

benjchristensen commented on Oct 29, 2014

rkuhn commented on Oct 29, 2014

ldaley commented on Oct 30, 2014

benjchristensen commented on Oct 30, 2014

rkuhn commented on Oct 30, 2014

17 remaining items

benjchristensen commented on Nov 2, 2014

benjchristensen commented on Nov 2, 2014

benjchristensen commented on Nov 2, 2014

ldaley commented on Nov 2, 2014

rkuhn commented on Nov 3, 2014

benjchristensen commented on Nov 3, 2014

benjchristensen commented on Nov 3, 2014

benjchristensen commented on Nov 3, 2014

benjchristensen commented on Nov 3, 2014

benjchristensen commented on Nov 3, 2014

ldaley commented on Nov 8, 2014

ldaley commented on Nov 8, 2014

benjchristensen commented on Nov 8, 2014

benjchristensen commented on Nov 11, 2014

rkuhn commented on Nov 11, 2014

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions