Flatten operator

[mttkay]

This is more a request for discussion.

I'm not sure .NET's Rx has something like this, but one thing I miss in RxJava is an operator to flatten an Observable<List<T>> to Observable<T>. As far as I can tell, this currently requires one to write somewhat clunky code using flatMap:

Observable.from(...).toList().flatMap(new Func1<List<...>, Observable<...>>() {
        @Override
        public Observable<...> call(List<...> list) {
            return Observable.from(list);
        }
    });

i.e. there seems to be no standard reverse operation for toList. Is this something you would deem worth having in the library itself? It would simplify the above call to:

Observable.from(...).toList().flatten()

[michaeldejong]

It looks a bit similar to the switch operator on Observable. That takes Observable<Observable<T>> and turns it into Observable<T>. But perhaps a new operator which enumerates entries from an Observable<Collection<T>> is not such a bad idea.

Not only would this be the reverse operation of toList, it would also be a reverse operation for the buffer operation, which buffers incoming elements into lists of elements, which are then propagated.

Note that this does have some limitations. To me Observable.from(...).toList().flatten() seems impossible. The flatten operator should only be available / usable for Observables which produce Collections of elements, and not for Observables which produce non-Collection objects.

The only usable syntax I can come up with is: Observable.flatten(Observable.from(...).toList())

[mttkay]

Note that this does have some limitations. To me Observable.from(...).toList().flatten() seems impossible. The flatten operator should only be available / usable for Observables which produce Collections of elements, and not for Observables which produce non-Collection objects.

that's a good point actually, didn't think about this. If we can't have the extra readability by flatten being an instance method, it might even suffice to have a flatten Func1 for now, which could be used like this:

// have to check if T could actually be inferred by Java here
Observable.from(1, 2, 3).toList().flatMap(Functions.flatten())

This is not much longer, doesn't read much worse, and would have the benefit of not impacting the Observable class interface.

[daveray]

I don't have much opinion on the syntax, but I'd find this useful. I end up
doing this operation on the result of toSortedList to switch back to an
observable sequence and continue processing.

Dave

On Tuesday, June 18, 2013, Matthias Käppler wrote:

Note that this does have some limitations. To me
Observable.from(...).toList().flatten() seems impossible. The flatten
operator should only be available / usable for Observables which produce
Collections of elements, and not for Observables which produce
non-Collection objects.

that's a good point actually, didn't think about this. If we can't have
the extra readability by flatten being an instance method, it might even
suffice to have a flatten Func1 for now, which could be used like this:

// have to check if T could actually be inferred by Java here
Observable.from(1, 2, 3).toList().flatMap(Functions.flatten())

This is not much longer, doesn't read much worse, and would have the
benefit of not impacting the Observable class interface.

—
Reply to this email directly or view it on GitHubhttps://github.com//issues/295#issuecomment-19610646
.

[prabirshrestha]

Here is how to flatten IObservable<int[]> to IObservable<int> in .NET

var observable = Observable.Return(new[] { 1, 2, 3, 4, 5 });

observable                    // IObservable<int[]> observable
    .SelectMany(x => x)       // returns Observable<int>
    .Subscribe(Console.WriteLine);

Here is the SelectMany operator being used.

public static IObservable<TResult> SelectMany<TSource, TResult>(this IObservable<TSource> source, Func<TSource, IEnumerable<TResult>> selector);

[benjchristensen]

Can anyone see a way of doing this that is type-safe?

Perhaps flatten(Class type) is the only way to do it so that we pass in the type that it should return? If it can't cast everything to that then it would onError a ClassCastException. Perhaps flatten() could exist but would have to emit Observable<Object>.

[samuelgruetter]

I'm slightly confused by this discussion and I'm not sure if everyone is talking about the same thing... ;-)

But I think @benjchristensen you're looking for something like this:

    // in Observable.java:
    public static <T> Observable<T> flatten(Observable<? extends List<? extends T>> source) {
        Func1<List<? extends T>, Observable<T>> f = new Func1<List<? extends T>, Observable<T>>() {
            public Observable<T> call(List<? extends T> t1) {
                return Observable.from(t1);
            }
        };
        return Observable.concat(source.map(f));
    }
    // in some test file: 
    @Test public void testFlattenFromList() {
        Observable<List<Integer>> o = Observable.from(Arrays.asList(1, 2), Arrays.asList(3, 4), Arrays.asList(5, 6));
        for (Integer i : flatten(o).toBlockingObservable().toIterable()) System.out.println(i);
    }

[akarnokd]

Yes, this looks straightforward. Rx.NET has a SelectMany overload for this kind of operation. I think using a List is too strong type here, maybe an Iteratable will suffice.

[MikolajKakol]

With new lift operation on Observables it's quite easy to create flatten operator. I've created something like this:

public static class ListToItemsOperator<T> implements Operator<T, Iterable<T>> {

    @Override
    public Subscriber<? super Iterable<T>> call(final Subscriber<? super T> child) {
        return new Subscriber<Iterable<T>>(child) {

            @Override
            public void onCompleted() {
                child.onCompleted();
            }

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

            @Override
            public void onNext(Iterable<T> list) {
                for (T t : list) {
                    if (child.isUnsubscribed()) {
                        break;
                    }
                    try {
                        child.onNext(t);
                    } catch (Exception e) {
          onError(OnErrorThrowable.addValueAsLastCause(e, t));
                        return;
                    }
                }
            }
        };
    }
}

Usage:

    Observable<List<Integer>> listObs = Observable.from(Arrays.asList(1, 2, 3, 4, 5, 6)).toList();
    listObs.lift(new ListToItemsOperator<Integer>()).subscribe(DumpObs.make());
    listObs.lift(new ListToItemsOperator<Integer>()).take(2).subscribe(Dump.Obs.make());

I'm not sure how good is that code, but it looks doable with type safety so It might be good moment to add it do core operators?

[benjchristensen]

That works well to pull in via lift() because you can provide the correctly typed implementation for the Observable you're working on, but it can't be put on Observable as an instance method unless we do casting (and blow up if invalid) since that only works when T is Iterable<T> (i.e. Observable<Iterable<T>>) and an Observable can be anything.

It could be done as a static Observable.flatten(Observable<Iterable<T>> o) (as @samuelgruetter mentions above) but then we'd need many different overloads of flatten and that doesn't work with the chaining pattern that is wanted.

In short, Java doesn't allow conditional extension methods on a class based on its generic type, so a flatten instance method can't be typed to Iterable<T>, or Iterable<Iterable<T>> or other such variants. It will always only receive whatever T is.

[MikolajKakol]

Sure, we shouldn't force to add that to Observable I was rather thinking to expand rx.operators package with FlattenOperator. Since this implementation is immutable we might might have single instance of this operator to slightly reduce objects creation.

[abersnaze]

Other notable places where we've done this are MathObservable and StringObservable. The contradiction is the that dematerialize() is an instance method on Observable<T> when it should only exist on instances Observable<Notification<T>>.

We could go crazy with the overloads to make things type safe

class Observable<T> {
    NotificationObservable<T> materialize()
    Observable<R> map(Func1<T, R>)
    NotificationObservable<R> map(Func1<T, Notification<R>>)
    ... a billion methods later ...
}
class NotificationObservable<T> extends Observable<Notification<T>> {
    Observable<T> dematerialize();
    // just in case someone materializes a materialized Observable.... turtles all the way down.
    NotificationNotificationObservable<T> materialize()
}
class NotificationNotificationObservable<T> extends NotificationObservable<Notification<T>> {
    NotificationObservable<T> dematerialize()
}

ok that was a silly idea.

[benjchristensen]

Yes, dematerialize is the counter-example where we do casting an throw an error if the type is not a Notification.

[benjchristensen]

I was rather thinking to expand rx.operators package with FlattenOperator

We have yet to start exposing custom operators for use via lift like this, and the rx.operators package is not the right place, since those are internal implementations and can change at any time.