How to create an ObservableList?

[fhaust]

This is of course easy when you do it like this:

Observable<List<T>> oMap = Observable.create(...);

But this gives you no informations about added or removed entries. In effect it forces you to iterate through all entries each time the map was changed. Additionally, if the List<T> contains Observables itself, you have no way to unsubscribe from removed elements or react on their removal.

My approach is something like this:

List<T> wrapped = new ArrayList<>();

public void onInit() {
  List<Action1<T>> addedEntryObservers = new ArrayList<>();
  Observer<T> addedEntries = Observable.create(
    observer -> observer.add(t -> observer.onNext(t))
  );
}

public void add(T t) {
  wrapped.add(t);
  addedEntryObservers.forEach(o -> o.call(t));
}

But this requires me to write some strange combinators if I just want to get all entries currently in the list.

Is there a canonical way to do this (yet)?

[akarnokd]

Hi. Could you elaborate on what you try to achieve in the first place?

[fhaust]

I have a dynamic collection of streams and want to migrate them to use Observables. While it is easy to model the streams as Observable<T>s, I struggle to find a (nice) way to get the stream added and stream removed notifications into the concept.

My current approach is to just use three Observables:

Observable<List<Observable<T>>> streams;
Observable<Observable<T>> addedStreams;
Observable<Observable<T>> removedStreams;

I could of course use a single Observable<List<Observable<?>>, but as I wrote above there is no way to react to individual changes.

(This could be done by delay()ing the Map and comparing the results ... but this will probably quite ineffective.)

[akarnokd]

Sending out notification and observing them can be achieved via Subjects. You'd need PublishSubject for both addedStreams and removedStreams.

public class StreamManager<T> {
    List<Observable<T>> streams = new ArrayList<>();
    public final PublishSubject<Observable<T>> streamAdded = PublishSubject.create();
    public final PublishSubject<Observable<T>> streamRemoved = PublishSubject.create();
    public final PublishSubject<List<Observable<T>>> streamChanged = PublishSubject.create();
    public void add(Observable<T> stream) {
        List<Observable<T>> current;
        sychronized  (this) {
            streams = new ArrayList<>(streams);
            streams.add(stream);
            current = streams;
        }
        streamChanged.onNext(current);
        addedStreams.onNext(stream);
    }
    public void remove(Observable<T> stream) {
        List<Observable<T>> current;
        synchronized (this) {
            streams = new ArrayList<>(streams);
            streams.remove(stream);
            current = streams;
        }
        removedStreams.onNext(stream);
        streamChanged.onNext(current);
    }
}

[fhaust]

Thanks! I'll have to try that, but it looks useful.

2014-02-03 akarnokd notifications@github.com

Sending out notification and observing them can be achieved via Subjects.
You'd need PublishSubject for both addedStreams and removedStreams.

public class StreamManager {
List<Observable> streams = new ArrayList<>();
public final PublishSubject<Observable> streamAdded = PublishSubject.create();
public final PublishSubject<Observable> streamRemoved = PublishSubject.create();
public final PublishSubject<List<Observable>> streamChanged = PublishSubject.create();
public void add(Observable stream) {
List<Observable> current;
sychronized (this) {
streams = new ArrayList<>(streams);
streams.add(stream);
current = streams;
}
streamChanged.onNext(current);
addedStreams.onNext(stream);
}
public void remove(Observable stream) {
List<Observable> current;
synchronized (this) {
streams = new ArrayList<>(streams);
streams.remove(stream);
current = streams;
}
removedStreams.onNext(stream);
streamChanged.onNext(current);
}}

Reply to this email directly or view it on GitHubhttps://github.com//issues/798#issuecomment-33956998
.

[fhaust]

Is there a canonical way to connect these StreamManagers?

Eg I might have a ReadManager and a WriteManager. ReadManager will add streams as they appear and WriteManager will subscribe to streamAdded and every announced stream.

Currently I'd just give WriteManager a reference to the ReadManager via its constructor and it will then subscribe to the necessary Observables.

[abersnaze]

General we implement dynamic collects of Obersvables as Obersvable<Obersvable<T>>. The outer Observable emits with onNext any new Obersvables as they are created and removal is the achieved by having the inner Obersvable emit an onCompleted.

Could your problem be solved with that? It would probably make it a lot easier to implement. Check out the implementation of groupBy()

[fhaust]

The problem is probably that I'd have a Observable<Observable<?>> where the type could be almost any basic or composite type. This is not known until runtime.

While I get to know the type before the first events come in that information would be lost if I just use a Observable<Observable<?>>.

[fhaust]

Maybe this question is better: How would you create composable blocks of functionality (like the Read- and WriteManager above) using the RXJava framework?

Note that there are multiple streams of data that run in parallel (and may be combined and/or splitted) in the individual blocks.

[benjchristensen]

Closing out as this conversation went stale ... if you'd like to followup on these questions, please post to https://groups.google.com/forum/#!forum/rxjava