RxRiddle 1xx

[rcd27]

Sometimes it comes to a case when we need to have cold observable which shares its emissions between multiple observers. For example, presentation layer calls for data and we want to store it in repository at the same call. I use publish().autoConnect(N) in that cases (N - number of observers). Here is an example of such usage:

  public Observable<User> getUsersAndStore() {
    Observable<User> listObservable = api.getUsers()
        .publish()
        .autoConnect(3);

// Here we store the result to ArrayMap<String, User>, 1st observer
    listObservable
        .collect(ArrayMap<String, User>::new,
            (destinationMap, user) -> destinationMap.put(user.type, user))
        .subscribe(typeWalletMap::putAll, Timber::e);

// 2nd
    listObservable
        .collect(ArrayMap<String, User>::new,
            (destinationMap, user) -> destinationMap.put(user.id, user))
        .subscribe(idWalletMap::putAll, Timber::e);

    return listObservable; // and finally return needed observable, the code which subscribes to it will be 
                                       // 3rd observer. That subscriber will initiate the emissions in the source 
                                       // observable.
  }

This approach seems to have side-effects(storing data to maps in this case), however we can't (or maybe just I can't) have code 100% free of side-effects, unfortunately. But the point is that all 3 observers use the same emissions without making source produce them 3 times.

Used literature: Thomas Nield, Learning RxJava '2017: Chapter 5 - Automatic connection

[vanniktech]

Theoretically you can subscribe to the returned Observable much more often. There's also autoRef() which could be used. Maybe we can have a riddle for autoRef() and that one would not need 1xx instead it could be in the lower digits.

[vanniktech]

I'm closing this issue due to inactivity. If you have any further input on the issue, don't hesitate to reopen this issue or post a new one.

[thomasnield]

I understand the intent here... but typically what I do in these cases is I coldly map() or flatMap() each operation.

val myStuff = Observable.just(1,2,3).toList()

myStuff.flatMap { 
    saveToFirstDestination(it)
}.flatMap { 
    saveToSecondDestination(it)
}.subscribe { 
   saveToFinalDestination(it)
}

That way the Observable is cold and can be repeatedly called multiple times.