Add support for rx.Single

[dalewking]

Single is the alternate version of Observable and should be supported to automatically unsubscribe based on Lifecycle events as well.

[dlew]

Seems reasonable to me.

BTW, you know that you misspelled "boycott", right?

[JakeWharton]

Also "GitHub"

[dalewking]

Note that it will be slightly more difficult because Single does not actually have a takeUntil operator so you will have to build that support yourself. Not sure how you will differentiate Observable vs. Single because you can't have a type implement both Observable.Transformer and Single.Transformer. You probably either have to have separate methods or your return value from bind methods is an object that has another method to get a Single version.

The workaround for now is do toObservable before the compose call and toSingle after it.

[dlew]

After pondering this for a while, it seems to me that instead of having two different versions - one for Single and another for Observable - that people should just use Single.toObservable() once they want to use compose(RxLifecycle) on a sequence.

Justifications:

1. Single.toObservable() is simple and doesn't have any negative side effects.
2. Often times you'll end up using compose(RxLifecycle) near the end of your chain, at which point it probably would've turned into an Observable by then anyways.
3. Supporting both means two different method names for each, which will be confusing.
4. The lack of takeUntil support in Single seems to be an oversight IMO. That indicates to me that Single isn't fully supported yet.

Any counterarguments?

[bensandee]

+1.

If you're following RxJava 2.x dev discussions it was suggested just today that Single is half-baked in 1.x (it is in fact marked as @Experimental). In 2.x it has more comprehensive support (e.g. a complete list of operators) and when migrating to support that it might be a nice time to revisit this.

[JakeWharton]

migrating to support

Migrating to 2.x means no longer doing Android as it'll be years if Android
ever sees the APIs required to run 2.x.

On Tue, Sep 29, 2015 at 6:26 PM Ben Sandee [email protected] wrote:

+1.

If you're following RxJava 2.x dev discussions it was suggested just today
that Single is half-baked in 1.x (it is in fact marked as @experimental).
In 2.x it has more comprehensive support (e.g. a complete list of
operators) and when migrating to support that it might be a nice time to
revisit this.

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

[dalewking]

Regarding your points:

1. It would likely be a chain of .toObservable().compose(...).toSingle()
2. Oddly I tend to do the compose at the beginning of the chain since I want to stop the chain as early as possible.
3. As I said it wouldn't be easy.
4. I'll agree it is experimental in 1.x as others have said.

[phajduk]

Update: Single has been updated to public API (more precisely to the @Beta) in RxJava 1.1.0.

The lack of takeUntil support in Single seems to be an oversight IMO. That indicates to me that Single isn't fully supported yet.

I believe it's not oversight as there is lack of more features (like retry).
Additionally, are you sure that we should expect takeUntil operator in Single? Single emits only either one value or error. Taking values until some predicate is true have no sense as there will be always one value.

[dlew]

@phajduk If a Single has a significantly delayed emission (as is the case with a network call) then you could takeUntil zero values have been emitted (which is essentially what this library does).

[phajduk]

@dlew You're right. I meant that there is no reason to implement takeUntil(predicate) not takeUntil(observable) which one is of course used in RxLifecycle. There could be also variant takeUntil(single) but it's RxJava issue.

Then 👍 from my side for using Single.toObservable().

[ZacSweers]

I just added takeUntil support to Single with an eye toward this. It'll now emit a CancellationException downstream if it's unsubscribed due to takeUntil, which would allow us to resume if necessary. See ReactiveX/RxJava#3712

Completable is easy, just Completable.amb(other). There's a bug that will be fixed in the next version that currently prevents it from being usable here though. ReactiveX/RxJava#3707

[dlew]

Awesome, we'll get those integrated once they're released.

[ZacSweers]

RxJava 1.1.2 is out and should work for both Single and Completable.

I have a branch open right now that does this, but wanted to get feedback on the API design. Specifically, we can't have overloads of bind that return Single.Transformer and Completable.CompletableTransformer since they still shared the same signature. Options are:

A. Make overrides specifically named (bind -> bindObservable, bindActivity -> bindObservableToActivity, etc).

B. Make separate classes for Single and Completable with the same API. RxLifecycle, RxSingleLifecycle, RxCompletableLifecycle.

---

Open to suggestions. B seems like the cleaner approach, and much of the shared logic can be pulled out into package-local internals.

[dalewking]

I think it would be bad to change any of the existing method names. Another alternative (an A.5) would be to not change existing calls, but add versions of the methods with Single and Completable in the name.

But I agree that B seems the best choice.

[ZacSweers]

A sort of moonshot option C:

generic return type and have the method take a Class to indicate which type it needs. Then create the appropriate transformer and (cast) return it. Loses type safety, but avoids all the duplicate code.

[dlew]

What'd be really nice is if Observable.Transformer, Single.Transformer and Completable.CompletableTransformer didn't exist at all, and instead the method signature just used Func1<T, T> where T == Observable, Single or Completable. Then one could write a transformer for all three that can be input into any of them.

I don't think we can get away with that with the current type system (though we could simplify our own implementation by having a single class that implements all three interfaces).

I'll be thinking on this for a bit. I don't have any qualms with changing method signatures, TBH, we'll just deprecate/warn again. But it still seems like that route would create a method explosion. C seems like the most reasonable route to me so far, since we can guarantee our own type safety.

[ZacSweers]

I did try the custom transformer that implements all three, but unfortunately there are conflicting method signatures that it can't resolve :/

[dlew]

You'd have to cast in the compose() method for that to work, unfortunately.

[dlew]

After thinking on it for a while, I think I'd prefer option C. Mainly because there's a combinatorial explosion that'll happen each time we add a new method (if we duplicate methods for each type).

I'm thinking that we'd still default to Observable for the base methods (at least for now), but then provide a second way to do it for any type. If someone ends up using it a lot they could always write their own wrapper for a particular method.

[ZacSweers]

Been hacking at option C, but unfortunately I can' seem to get it to a state where it doesn't at least complain about casting in line.

The following code:

@NonNull
@CheckResult
public static <R> R bindActivity(
        @NonNull final Class<? super R> transformerClass,
        @NonNull final Observable<ActivityEvent> lifecycle) {
    return bind(transformerClass, lifecycle, ACTIVITY_LIFECYCLE);
}

@SuppressWarnings("unchecked")
@NonNull
@CheckResult
public static <T, R> R bind(
        @NonNull final Class<?> transformerClass,
        @NonNull final Observable<T> lifecycle,
        @Nullable final Func1<T, T> correspondingEvents) {
    checkNotNull(transformerClass, "transformerClass == null");
    checkNotNull(lifecycle, "lifecycle == null");

    // Make sure we're truly comparing a single stream to itself
    final Observable<T> sharedLifecycle = lifecycle.share();

    final Observable<?> eventMappingObservable = correspondingEvents == null
            ? lifecycle
            : eventMappingObservable(sharedLifecycle, correspondingEvents);

    // Keep emitting from source until the corresponding event occurs in the lifecycle
    if (Completable.CompletableTransformer.class.isAssignableFrom(transformerClass)) {
        return (R) new Completable.CompletableTransformer() {
            @Override
            public Completable call(Completable completable) {
                return completable.ambWith(eventMappingObservable.toCompletable());
            }
        };
    } else if (Observable.Transformer.class.isAssignableFrom(transformerClass)) {
        return (R) new Observable.Transformer<T, T>() {
            @Override
            public Observable<T> call(Observable<T> source) {
                return source.takeUntil(eventMappingObservable);
            }
        };
    } else if (Observable.Transformer.class.isAssignableFrom(transformerClass)) {
        return (R) new Single.Transformer<T, T>() {
            @Override
            public Single<T> call(Single<T> source) {
                return source.takeUntil(eventMappingObservable);
            }
        };
    } else {
        throw new IllegalArgumentException("Unsupported target class");
    }
}

Results in this:

[dlew]

This seems to work for me:

public static <T, R> T bind(Observable<R> lifecycle, Class<T> transformerType) {
  // Implement here
}

[ZacSweers]

It works, but lint yells at me. You don't get a lint warning?

[dlew]

Ah, yeah, I do get that. That's because you can't define the generic type of the returned Transformer - e.g. it's returning Transformer instead of Transformer<T, T>.

I played around with it a bit and got a little further, but haven't quite solved it yet. You can do this:

public static <T extends Func1<R, R>, R> T basicTransformer() {
  // Do whatever here
}

Then you can do this:

Observable.Transformer<String, String> transformer = basicTransformer();

And it doesn't complain. But if you try to use basicTransformer() inside of compose() it doesn't figure it out.

[dalewking]

There has always been issues with the generic type inference with RxLifecycle and the compose method. See issues #44, #58, and #59. As it stands today in Kotlin I have to explicitly add a type parameter.

Would be great if the type inference can be fixed once and for all, but the problem may actually be with RxJava's declaration of compose as well.

[ZacSweers]

Yeah unfortunately mixing doesn't really work either since Single and Observable aren't the same type (so the method signature in the transformer would conflict) and Completable is a different one entirely. Starting to think the slicing approach of option B is probably the best bet for now, and trying to refactor and share as much of the logic as possible.

[dalewking]

Any updates on this? Seems to have stalled a month ago. In the process of converting som Observables to Single and Completable

[dlew]

It's stalled out due to soul searching.

The way the API is would need to be setup means that supporting Single and Completable would increase the complexity. A small complexity increase is alright, but this seems to go beyond that since we'd need 3x of every method. Maintenance would become more difficult as well, as we'd essentially be maintaining three separate code paths (even though they all fundamentally do exactly the same thing).

I feel like this is going to be a problem for any library built on top of RxJava, so I'm a bit frustrated with how the design shook out.

As a result, what I've been considering is simply not supporting Single and Completable. One can still use toX() to add Single and Completable support as necessary without making this library significantly more complex.

[dalewking]

I can say with lots of calls, the whole toObservable() toX() pairing is a
tedious and just seems wrong. It is unfortunate that Rx designers did not
forsee the need to create one object that could translate all three.

I wonder if this might be a solution. Right now all the bind calls return
Observable.Transformer<T, T>. What if you changed that to your own exension
interface like this:

interface RxLifecycleTransformer<T> extends Observable.Transformer<T, T>
{
    Single.Transformer<T, T> forSingle();
    Func1<Completable, Completable> forCompletable();
}

Where you implement the methods to convert for single and completable.

Then for users, they just have to do something like:

 .compose(RxLifecycle.bindToLifecycle().forSingle())

[dlew]

That's an interesting solution! It looks like a decent mixture of common case/back-compat support but also being able to support the other types. I think I'll explore that sometime this week.

[dalewking]

Seems that parts of your API were missed. ActivitiyLifecycleProvider does not have the change to support Single and Completable and also Navi component does not have it.

[dlew]

D'oh. Will fix.

[dalewking]

FYI have submitted PR for RxJava2 that will allow a single object to act as transformer for all reactive types: ReactiveX/RxJava#4672

[dlew]

Excellent, thanks!

[dalewking]

And the PR was just merged, so RxLifeCycle for RxJava2 won't need this workaround.