Observable<Void> Usage or VoidObservable

[NiteshKant]

While working on RxNetty, a networking library that layers RxJava on top of netty, I have come across a common case of expressing result of an operation as an Observable<Void>. A typical case being the result of a write on a Connection. The API for write looks like:

public Observable<Void> write(Observable<T> msgs);

The return value just represents the asynchronous result of the write and hence does not emit any item, so has the type as Void.

This works pretty well, however, it becomes awkward when we start doing a bit more complex operations over this write. eg: If I am writing a TCP client that sends a message on the connection and then reads the response from the server, I start doing things like:

connection.write(Observable.just("hello"))
                 .cast(String.class)
                 .concatWith(connection.getInput())

(Actual example here)

In this example, connection.getInput() returns an Observable<T> emitting items as read on the connection.

and in case, I do not want to wait for write to complete, the code looks like:

connection.write(Observable.just("hello"))
                 .cast(String.class)
                 .mergeWith(connection.getInput())

What is awkward in the above example, is the cast(String.class). It is completely safe as the write never emits an item (Void) so I can merge/concat heterogeneous streams but it is just awkward.

My motivation behind creating this issue is to first see if other folks have come across the same scenario, if so, how have they handled it. Secondly, if this does sound like something worth spending time to make it better, I would like to propose some abstraction on the lines of a VoidObservable (for the lack of a better name) which basically provides better constructs around:

• Concat with a different type: write().cast().concatWith() example above, where it is expected to continue with another stream of a different type (String in the above example)
• Merge with a different type: write().cast().mergeWith() example above, where it is expected to eagerly subscribe to the second Observable but listen only for errors from the Observable<Void>

[akarnokd]

I see two problems:

Type inference limitations

The lack of cast-down type-inference in the Java language. For example: if I have an Observable<Double> a and an Observable<Integer> b, I can't use mergeWith() because it expects the parameter to extend the type of the source, which Double and Integer doesn't do to each other. You'd expect c = a.mergeWith(b) have a common supertype of Observable<Number> which type conversion can't be expressed in Java:

interface IObservable<T> {
    <U, R super T & U> IObservable<R> mergeWith(IObservable<U> other);
}

Here, super is not allowed and gives a compiler error.

If I try to cheat this out via extends I have to add a second type parameter to the observable:

interface IObservable<T extends R, R> {
    void subscribe(Subscriber<R> child);
    default <U extends R> IObservable<R, R> mergeWith(IObservable<U, R> other) {
        return null;
    }
}

This requires me to specify some common basetype so the merge can fall back onto it, however, it has limitations too:

IObservable<Double, Number> io1 = s -> {};
IObservable<Integer, Number> io2 = s -> {};
IObservable<Integer, Object> io3 = s -> {};

io1.mergeWith(io2);
io2.mergeWith(io1);
io1.mergeWith(io3); // won't compile, incompatible types.

The only thing that works is the static merge:

Observable<? extends Number> merge = Observable.merge(o1, o2);

But still, it forces the user to have ? extends Number carried everywhere. Rx.NET falls into this category so it seems to be a no-issue there and due to declaration-site variance, you don't have super and extends everywhere.

Operator methods are final

Even if you have your own, fixed type observable, the following won't compile because our operator methods are final:

public static final class VoidObservable extends Observable<Void> {

    protected VoidObservable(OnSubscribe<Void> onSubscribe) {
        super(onSubscribe);
    }

    public <T> Observable<T> mergeWith(Observable<T> other) {
        return null;
    }
}

The compiler will most likely complain about same-erasure non-overriding methods or just simply about override attempt of a final method. You'd have to name this method something else and remember to use it instead of the standard.

[abersnaze]

I took a crack at changing RxNetty to use Observable<?> but ran into generics problems when it came to its usage of onErrorResumeNext(Func1). Now that we have doOnError(Action1) that doesn't have to return Observable<?> it might work.

[NiteshKant]

@akarnokd @abersnaze Thanks for your comments, let me answer them one at a time.

I see two problems:

You are absolutely correct in identifying these issues & I should have clarified my intent better. What I was intending to is to introduce a class like:

public class VoidObservable extends Observable<Void> {

    protected VoidObservable(OnSubscribe<Void> f) {
        super(f);
    }

    public <T> Observable<T> continueWith(Observable<T> next) {
        final Observable<T> cast = unsafeCast();
        return cast.concatWith(next);
    }

    public <T> Observable<T> mergeErrorWith(Observable<T> next) {
        final Observable<T> cast = unsafeCast();
        return cast.mergeWith(next);
    }

    private <T> Observable<T> unsafeCast() {
        @SuppressWarnings("rawtypes")
        final Observable rawSource = this;
        @SuppressWarnings("unchecked")
        final Observable<T> cast = rawSource;
        return cast;
    }
}

and add any other methods specific to VoidObservable. This is basically clarifying the intent of the operation better than .cast().XWith().

Do you see this working or making sense?

I took a crack at changing RxNetty to use Observable<?>

Can you elaborate more i.e. were you trying to make RxNetty's Connection.write() return Observable<?> ?

[abersnaze]

I took a crack at changing RxNetty to use Observable<?>

Can you elaborate more i.e. were you trying to make RxNetty's Connection.write() return Observable<?> ?

The observer of the return value from Connection.write() only cares about the terminal state/time of that Observable and therefore shouldn't put any restrictions on the type of data that it is ultimately going to ignore. If there are no restriction on the type then there wouldn't be any need for casting it to Observable<Void>.

[NiteshKant]

@abersnaze I agree with you on why <?> could be better in terms of indicating that the subscriber should not care about the type, although in this context, wouldn't it have the same problem? If you keep aside the RxNetty connection and look at this simple case,

        Observable<?> source = Observable.just("Hi");
        source.concatWith(Observable.just(1));

The above isn't valid as Observable<?> can not be concatenated with Observable<Integer> in this case. So, you will have to do:

        Observable<?> source = Observable.just("Hi");
        source.cast(Integer.class).concatWith(Observable.just(1));

which is the same thing as Observable<Void>, correct?

[abersnaze]

You wouldn't use concat because it doesn't make sense. Concat is concatenating the data you don't care about the data you are only using concat because enforces an order and propagation of errors. Here is code that does what you want.

Observable<?> source = Observable.just("Hi");
source.lift(new Operator<Integer, Object>() {
    @Override
    public Subscriber<? super Object> call(Subscriber<? super Integer> child) {
        return new Subscriber<Object>(child) {
            @Override
            public void onCompleted() {
                Observable.just(1).subscribe(child);
            }

            @Override
            public void onError(Throwable e) {
                child.onError(e);
            }

            @Override
            public void onNext(Object t) {
                // ignore data
            }
        };
    }
});

[NiteshKant]

Sure, I think the point here is whether using <?> is any better than <Void>, I do not really see a difference in terms of applying any higher order functions on the Observable, which is the context for this discussion.

Now, if I take out <?> argument, I think the problem still is similar i.e. how to make the use case of continueWith and mergeErrorWith, I outlined above, more palatable to the users. Whether the actual implementation of continueWith is .cast().concatWith() or a specific operator like @abersnaze mentioned can be discussed later.

[benjchristensen]

I don't think VoidObservable will work very well due to how generics behave. I can't see how it would make concatWith behave any better.

I think something like continueWith would work better, which by contract is saying that it ignores the generics of the previous stream and continues with the next stream onCompletion of the previous.

Observable<Void> someStream ...
Observable<T> otherStream ...
Observable<T> combined = someStream.continueWith(otherStream);

This achieves the goal of ignoring output (a 'Void' or ignoreElements()), ignoring the type (such as via cast(Class.class), and concatting when onComplete emits while also emitting onError.

[NiteshKant]

I see, so what you are saying is add .continueWith and possible mergeError in Observable instead of creating an extension like VoidObservable.

The only ambiguity will be, what if, someone uses it with a non-void Observable? We can possibly do something like: raising an error if the first stream emits any element.

I am fine with either approaches, if and when we have a consensus here, I can submit a PR for this change!

[benjchristensen]

Yes, it would behave like single() which emits an error if it emits an onNext.

[abersnaze]

I'm okay with that restriction because if someone wanted to use the continueWith (I like switchEmpty more) but wanted to ensure the error doesn't happen then they can compose it with ignoreElements.