rx.Future/Task/Async/Single

Explore whether an rx.Future makes sense to allow better representation of Scalar responses.

Comment below explaining reasoning: #1594 (comment)

I really would love to see this because of 2 reasons:

There is no good (async & composable) Future type in the Java standard library if you have to target < Java8. Even with CompleteableFuture available this uses executors for the continuations. For an RxJava integration using an Rx Scheduler is a better match - we can then receive Observable events and Future continuations on the same scheduler.
The alternative is using an AsyncSubject. The problem with AsyncSubject is that the user only sees an Observable in the API and thereby won't know that it emits only a single item. He also won't know if it's hot (the function call which returns the Observable will immediatly start an operation without the need for Subscribe() or not (needs Subscribe()). Therefore I think the ability to model such single return-value async operations with Futures makes an API better understandable.

https://github.com/BoltsFramework/Bolts-Java is the java version of .net's task.
Rx.net has seamless integration between observable and task.

However without async await syntax, tasks/futures do not provide much advantage. I don't ever expect async await to be added to java BTW.

Rx.net has seamless integration between observable and task.

Yes, but RxJava hasn't. An integration/interfacing with Bolts could be an option instead of separate implementation, but it would probably require som changes to Bolts as well. Bolts, unlike RxJava, rely on Executors.

However without async await syntax, tasks/futures do not provide much advantage

Over what? Observables? Simpler syntax and easier learning curve would be one. Right now I'm using Observables for Scalar data (HTTP responses) because that's the only thing Retrofit supports, but it's been a challenge to sift through the wiki finding just the methods I need for these kind of tasks. continueWith=flatMap? Not intuitive. I simply feel I'm using the wrong tool for the job.

it's been a challenge to sift through the wiki finding just the methods I need for these kind of tasks

The flatMap operator would almost certainly exist on the rx.Future. For example, see the Twitter Future that has flatMap: https://twitter.github.io/util/docs/index.html#com.twitter.util.Future

What makes the use of flatMap not intuitive on an Observable that has a single value versus multiple values?

What makes the use of flatMap not intuitive on an Observable that has a single value versus multiple values?

First and foremost the naming. "FlatMap" doesn't tell me anything. What's "unflat" about a Scalar value? What part of the operation involves mapping of any kind? Promise.then is much more descriptive to me.

But I realize it might be my background coming from Tasks in C# to Promises in Javascript and then Observables that might be "polluting my mind". I also might have misunderstood this isse. Maybe you could elaborate, since you're the OP?

flatMap in RxJava is selectMany in C#. Since you mention you know Tasks form C# you should read this http://blogs.msdn.com/b/pfxteam/archive/2013/04/03/tasks-monads-and-linq.aspx. Technically speaking the Task API is a so-called co-monad, but since they are continuations, Task can also be formulated as a monad. In Rx .NET we had manySelect for a while that corresponds to continueWith.

What's "unflat" about a Scalar value?

An async continuation that returns another Future needs to be flattened:

aFuture.flatMap(value -> {
   return doWithThatReturnsAnotherFuture(value);
})

In that case the flat part is flattening the returned Future, otherwise you'd end up with a Future<Future<T>> rather than a Future<T>.

What part of the operation involves mapping of any kind

The function can map from T to R however you wish.

Future<R> Future.flatMap(Func1<T, Future<R>> mapper)

If there is no need to flatten then it would be map:

Future<R> Future.map(Func1<T, R> mapper)

If it is purely a side-effect, then the naming is up in the air but in RxJava thus far it is doOn* method names to signal the side-effects, so something like doOnSuccess/doOnError on a scalar Future and just pass the value through:

Future<T> Future.doOnSuccess(Action1<T> sideEffect)
Future<T> Future.doOnError(Action1<Throwable> sideEffect)

Ok thanks for taking the time to answer my questions, it made things clearer.

I think the issue here is that flatmap is a description what happens in general whereas then or continueWith describe it in a more concrete way for the Future type.
Some time ago I had no experience with FP programming, so flatmap didn't tell me anything. However continueWith is more self-describing so I could understand it and it's easier to discover in an API documentation. I like it. But as I later learned that then/continueWith are equivalent to flatmap so now I understand it too. Maybe the same applies to other users?
Why not expose both function signatures? Don't think it would be a problem and I think it would make the familarity higher for OO as well as FP programmers.

However I thought the more general question was whether RxJava would get a Task/Future like structure at all? I would still like to see it. Bolts looks good, but having a single library for all kinds of asynchronous things and with the possibility for interaction between structures (futures to observables and the other way around, sharing of schedulers, ...) would be even nicer.

However I thought the more general question was whether RxJava would get a Task/Future like structure at all?

Yes, that's the more general question. I tend to think we should.

having a single library for all kinds of asynchronous things and with the possibility for interaction between structures ... would be even nicer.

Agreed, which is why I want it part of RxJava. Most everything would be shared, mostly just the type signature would differ.

Can someone explain what this Future/Task should do, or more specifically, what is the underlying issue we want to solve that Observable or Async can't do (efficiency?)?

Represent request/response of a single value. It is purely a design thing, not functional. Many, many people would prefer to represent scalar values such as IO request/response (think HTTP, REST, RPC) as a Future/Task but be able to merge them into Observable streams. It is one of the most common questions I get.

Rx.Net has it because .Net already has Task that works nicely with Observable since they were built for each other. The Java Future is unusable in this sense and CompletableFuture in Java 8 can be made to work but it's entire API design is incredibly different and they are always "hot".

@benjchristensen

@benjchristensen I meant potential postixes, i.e., ScalarTask, ScalarObservable or ScalarFuture, etc.

Since we will use the experimental annotation, let's proceed with Single and come back to the name in a few months once enough use experience has been accumulated.

I like Single but just to throw more alternatives, Observable1 or OneObservable can be considered to avoid mixing up with other class names/libraries.

Here are examples of using Single:

import java.util.concurrent.TimeUnit;

import rx.Observable;
import rx.Single;
import rx.Single.Promise;
import rx.schedulers.Schedulers;

public class SingleExperiment {

    public static void main(String... args) {
        Observable<String> a_merge_b = getDataA().mergeWith(getDataB());

        Single<String> a_and_b = getDataA().flatMap(t -> {
            return getDataB().map(r -> t + r);
        });

        Single<String> a_transformed = getDataA().map(a -> a + "_response");

        Observable<String> a_concat_b = getDataA().concatWith(getDataB());

        /**
         * This shows flatMapping a Single with an Observable
         */
        Observable<String> a_with_c = getDataA().flatMapObservable(a -> {
            return getDataC().map(c -> a + "-" + c);
        });

        Single<String> aWithTimeout = getDataA().timeout(100, TimeUnit.MILLISECONDS);

        Single<String> aWithTimeoutAndSimpleDefaultValue = aWithTimeout.onErrorReturn(t -> "DefaultValue");

        /**
         * We can use toObservable() for a Single to interact with an Observable.
         * Or we can add the various overloads to Observable.merge/concat/zip/mergeWith/concatWith/zipWith/etc
         * Adding the overloads though does result in a lot of API clutter.
         */
        getDataC().mergeWith(getDataA().toObservable());

    }

    public static Single<String> getDataA() {
        return Single.<String> create(o -> {
            o.onSuccess("DataA");
        }).subscribeOn(Schedulers.io());
    }

    public static Single<String> getDataB() {
        return Single.just("DataB")
                .subscribeOn(Schedulers.io());
    }

    public static Observable<String> getDataC() {
        return Observable.interval(10, TimeUnit.MILLISECONDS).map(i -> "c" + i);
    }

    /**
     * Example of Single.Promise
     */
    public static Single<String> getDataD() {
        Promise<String> p = Single.Promise.create();
        new Thread(() -> {
            p.onSuccess("data-d-from-thread");
        }).start();
        return p.getSingle();
    }
}

@benjchristensen

@benjchristensen I like this new proposal but could you explain why there is a need for this embedded Promise type?

Reposting a modified version of #2641 (comment) to provide reasoning for this proposal:

The evolution of thinking after using RxJava for a couple years is that libraries that expose public APIs can benefit from being able to communicate if something is scalar or vector – despite the "functional" argument that a scalar is just a "list of one". We are able to choose to use T or List<T> when creating a synchronous API, so why not for async as well? In other words, why should the use of RxJava exclude communicating when we do know something is a scalar, which is a very common use case, and results in a simpler mental model?

When composition occurs that results in multiple values then the type changes to an Observable. Not all composition and transformation requires this though – zip, amb, map, flatMap all can return Single. Others like merge will return an Observable.

Another consideration is that APIs that expose data access are generally separate from the consuming code that composes them, and it is these data access APIs where the scalar Single is most effective at providing communicating via a static type such as:

Single<T> getData()

versus

Observable<T> getData()

does this actually simplify anything?

There is one thing I think it simplifies that is more than just making it easier due to familiarity. It communicates when a user does NOT need to think about multiple values, and that does indeed simplify code.

A Single will always behave in one of 3 ways:

respond with an error
never respond
respond with a success

An Observable on the other hand may:

response with an error
never respond
respond successfully with no data and terminate
respond successfully with a single value and terminate
respond successfully with multiple values and terminate
respond successfully with one or more values and never terminate (waiting for more data)

The mental model for these extra possibilities do definitely increase the complexity and are a big part of the pushback for using RxJava that I have seen for request/response models.

These more complex models do still end up happening as soon as composition occurs, but now that composition is declared by the consumer, and the consumer knows the contract of the producer to be simpler if it is a Single.

For example, in "service layer" type code we could have getters like this:

Single<T> getDataA(arg);
Single<R> getDataB(arg);
Observable<R> getDataC(arg);

This is now simpler to understand because the types narrow the scope for A and B so we know we don't need to consider multi-value or infinite cases (which generally need to be explained in the javadoc). The C type in contrast clearly indicates that it wants to support multiple values otherwise it would have just been a Single.

familiarity can always be learned, but complexity is harder to avoid

My thinking has evolved to think that trying to force everything to be a vector results in increased complexity of a different kind that can be addressed by a slight increase in complexity of the RxJava APIs (2 types instead of 1). Static types communicate common information that simplifies the mental model and constrains available states and operations that can be applied. This seems worth the increase in API surface area.

@sdeleuze

could you explain why there is a need for this embedded Promise type

@sdeleuze it is not required and I should probably remove it. I added it to prove a common programming model.

Below are two ways of achieving the same thing. The first eliminates nesting but does involve an extra type. However, Promises are "hot" whereas the Single.create approach supports hot and cold execution.

    public static Single<String> getDataD() {
        Promise<String> p = Single.Promise.create();
        doStuffWithCallback(d -> p.onSuccess(d + "-from-thread"));
        return p.getSingle();
    }

    public static Single<String> getDataE() {
        return Single.create(o -> {
            doStuffWithCallback(d -> {
                o.onSuccess(d + "-from-thread");
            });
        });
    }

rx.Single

An @Experimental version was merged in #3012

Navigation Menu

Repository navigation

rx.Future/Task/Async/Single #1594

15 remaining items

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

rx.Future/Task/Async/Single #1594

Description

Activity

Matthias247 commented on Sep 18, 2014

headinthebox commented on Sep 19, 2014

Nilzor commented on Nov 24, 2014

benjchristensen commented on Nov 24, 2014

Nilzor commented on Nov 24, 2014

headinthebox commented on Nov 24, 2014

benjchristensen commented on Nov 25, 2014

Nilzor commented on Nov 25, 2014

Matthias247 commented on Nov 25, 2014

benjchristensen commented on Nov 25, 2014

akarnokd commented on Feb 5, 2015

benjchristensen commented on Feb 6, 2015

15 remaining items

akarnokd commented on May 11, 2015

schrepfler commented on May 11, 2015

benjchristensen commented on May 12, 2015

sdeleuze commented on May 12, 2015

benjchristensen commented on May 12, 2015

benjchristensen commented on May 12, 2015

benjchristensen commented on Jun 10, 2015

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions