Trivial ViewState pattern (#16)

* Finally subscribe onActive as observers are needed for that

* Example with dummy ViewState preventing Activity to handle errors

* RxLiveData without inheritance

client-data/src/main/java/com/jraska/github/client/rx/RxLiveData.java

@@ -1,60 +1,56 @@
 package com.jraska.github.client.rx;
 
-import android.arch.lifecycle.LifecycleOwner;
 import android.arch.lifecycle.LiveData;
-import android.arch.lifecycle.Observer;
 import android.support.annotation.Nullable;
 
+import io.reactivex.Maybe;
 import io.reactivex.Observable;
 import io.reactivex.Single;
 import io.reactivex.disposables.Disposable;
 import io.reactivex.functions.Consumer;
 
-public abstract class RxLiveData<T> extends LiveData<T> {
+public final class RxLiveData<T> extends LiveData<T> {
+  public static <T> RxLiveData<T> from(Observable<T> observable) {
+    return create(new ObservableAdapter<>(observable));
+  }
+
   public static <T> RxLiveData<T> from(Single<T> single) {
-    return new SingleAdapter<>(single);
+    return create(new SingleAdapter<>(single));
   }
 
-  public static <T> RxLiveData<T> from(Observable<T> observable) {
-    return new ObservableAdapter<>(observable);
+  public static <T> RxLiveData<T> from(Maybe<T> maybe) {
+    return create(new MaybeAdapter<>(maybe));
   }
 
-  protected Consumer<? super Throwable> onError;
+  private static <T> RxLiveData<T> create(SubscriberAdapter<T> adapter) {
+    return new RxLiveData<>(adapter);
+  }
+
+  private final SubscriberAdapter<T> subscriberAdapter;
+
   @Nullable private Disposable subscription;
 
-  RxLiveData() {
+  RxLiveData(SubscriberAdapter<T> subscriberAdapter) {
+    this.subscriberAdapter = subscriberAdapter;
   }
 
-  @Override public void observe(LifecycleOwner owner, Observer<T> observer) {
-    super.observe(owner, observer);
+  @Override protected void onActive() {
+    super.onActive();
     if (subscription == null) {
       subscription = subscribe();
     }
   }
 
-  @Override public void removeObserver(Observer<T> observer) {
-    super.removeObserver(observer);
-    if (!hasObservers()) {
-      dispose();
-    }
+  @Override protected void onInactive() {
+    dispose();
+    super.onInactive();
   }
 
-  public RxLiveData<T> resubscribe() {
+  public void resubscribe() {
     if (subscription != null) {
       dispose();
       subscription = subscribe();
     }
-
-    return this;
-  }
-
-  // TODO: 07/06/17 Error handling exposes RxLiveData everywhere. Too invasive
-  // Solution is to make LiveData never fail -> LiveData<ViewState>
-  public RxLiveData<T> observe(LifecycleOwner owner, Observer<T> observer,
-                               Consumer<? super Throwable> onError) {
-    this.onError = onError;
-    observe(owner, observer);
-    return this;
   }
 
   private void dispose() {
@@ -64,42 +60,51 @@ private void dispose() {
     }
   }
 
-  abstract Disposable subscribe();
+  private Disposable subscribe() {
+    return subscriberAdapter.subscribe(this::setValueInternal);
+  }
 
   void setValueInternal(T value) {
     setValue(value);
   }
 
-  static final class SingleAdapter<T> extends RxLiveData<T> {
+  interface SubscriberAdapter<T> {
+    Disposable subscribe(Consumer<T> onValue);
+  }
+
+  static final class ObservableAdapter<T> implements SubscriberAdapter<T> {
+    private final Observable<T> observable;
+
+    ObservableAdapter(Observable<T> observable) {
+      this.observable = observable;
+    }
+
+    @Override public Disposable subscribe(Consumer<T> onNext) {
+      return observable.subscribe(onNext);
+    }
+  }
+
+  static final class SingleAdapter<T> implements SubscriberAdapter<T> {
     private final Single<T> single;
 
-    private SingleAdapter(Single<T> single) {
+    SingleAdapter(Single<T> single) {
       this.single = single;
     }
 
-    @Override
-    Disposable subscribe() {
-      if (onError == null) {
-        return single.subscribe(this::setValueInternal);
-      } else {
-        return single.subscribe(this::setValueInternal, onError);
-      }
+    @Override public Disposable subscribe(Consumer<T> onSuccess) {
+      return single.subscribe(onSuccess);
     }
   }
 
-  static final class ObservableAdapter<T> extends RxLiveData<T> {
-    private final Observable<T> observable;
+  static final class MaybeAdapter<T> implements SubscriberAdapter<T> {
+    private final Maybe<T> maybe;
 
-    private ObservableAdapter(Observable<T> observable) {
-      this.observable = observable;
+    MaybeAdapter(Maybe<T> maybe) {
+      this.maybe = maybe;
     }
 
-    @Override Disposable subscribe() {
-      if (onError == null) {
-        return observable.subscribe(this::setValueInternal);
-      } else {
-        return observable.subscribe(this::setValueInternal, onError);
-      }
+    @Override public Disposable subscribe(Consumer<T> onSuccess) {
+      return maybe.subscribe(onSuccess);
     }
   }
 }

client-data/src/main/java/com/jraska/github/client/users/UserDetailViewModel.java

@@ -1,24 +1,26 @@
 package com.jraska.github.client.users;
 
 import android.arch.lifecycle.ViewModel;
+
 import com.jraska.github.client.Navigator;
 import com.jraska.github.client.Urls;
 import com.jraska.github.client.analytics.AnalyticsEvent;
 import com.jraska.github.client.analytics.EventAnalytics;
 import com.jraska.github.client.rx.AppSchedulers;
 import com.jraska.github.client.rx.RxLiveData;
-import io.reactivex.Observable;
 
 import java.util.HashMap;
 import java.util.Map;
 
+import io.reactivex.Observable;
+
 public class UserDetailViewModel extends ViewModel {
   private final UsersRepository usersRepository;
   private final AppSchedulers schedulers;
   private final Navigator navigator;
   private final EventAnalytics eventAnalytics;
 
-  private final Map<String, RxLiveData<UserDetail>> liveDataMapping = new HashMap<>();
+  private final Map<String, RxLiveData<ViewState>> liveDataMapping = new HashMap<>();
 
   UserDetailViewModel(UsersRepository usersRepository, AppSchedulers schedulers,
                       Navigator navigator, EventAnalytics eventAnalytics) {
@@ -28,8 +30,8 @@ public class UserDetailViewModel extends ViewModel {
     this.eventAnalytics = eventAnalytics;
   }
 
-  public RxLiveData<UserDetail> userDetail(String login) {
-    RxLiveData<UserDetail> liveData = liveDataMapping.get(login);
+  public RxLiveData<ViewState> userDetail(String login) {
+    RxLiveData<ViewState> liveData = liveDataMapping.get(login);
     if (liveData == null) {
       liveData = newUserLiveData(login);
       liveDataMapping.put(login, liveData);
@@ -38,12 +40,15 @@ public RxLiveData<UserDetail> userDetail(String login) {
     return liveData;
   }
 
-  private RxLiveData<UserDetail> newUserLiveData(String login) {
-    Observable<UserDetail> detailObservable = usersRepository.getUserDetail(login)
+  private RxLiveData<ViewState> newUserLiveData(String login) {
+    Observable<ViewState> viewStateObservable = usersRepository.getUserDetail(login)
       .subscribeOn(schedulers.io())
-      .observeOn(schedulers.mainThread());
+      .observeOn(schedulers.mainThread())
+      .map((userDetail -> new ViewState(null, userDetail)))
+      .onErrorReturn((error) -> new ViewState(error, null))
+      .startWith(new ViewState(null, null));
 
-    return RxLiveData.from(detailObservable);
+    return RxLiveData.from(viewStateObservable);
   }
 
   public void onUserGitHubIconClick(String login) {
@@ -55,4 +60,26 @@ public void onUserGitHubIconClick(String login) {
 
     navigator.launchOnWeb(Urls.user(login));
   }
+
+  public static class ViewState {
+    private final Throwable error;
+    private final UserDetail result;
+
+    public ViewState(Throwable error, UserDetail result) {
+      this.error = error;
+      this.result = result;
+    }
+
+    public boolean isLoading() {
+      return (result == null || result.basicStats == null) && error == null;
+    }
+
+    public Throwable error() {
+      return error;
+    }
+
+    public UserDetail result() {
+      return result;
+    }
+  }
 }

client-data/src/main/java/com/jraska/github/client/users/UsersViewModel.java

@@ -1,47 +1,66 @@
 package com.jraska.github.client.users;
 
+import android.arch.lifecycle.LiveData;
 import android.arch.lifecycle.ViewModel;
+
 import com.jraska.github.client.Navigator;
 import com.jraska.github.client.Urls;
 import com.jraska.github.client.analytics.AnalyticsEvent;
 import com.jraska.github.client.analytics.EventAnalytics;
 import com.jraska.github.client.rx.AppSchedulers;
 import com.jraska.github.client.rx.RxLiveData;
-import io.reactivex.Single;
 
 import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
 
-public class UsersViewModel extends ViewModel{
+import io.reactivex.Observable;
+import io.reactivex.Single;
+import io.reactivex.SingleEmitter;
+import io.reactivex.SingleOnSubscribe;
+
+public class UsersViewModel extends ViewModel {
 
   private final UsersRepository usersRepository;
   private final AppSchedulers appSchedulers;
   private final Navigator navigator;
   private final EventAnalytics eventAnalytics;
 
-  private final RxLiveData<List<User>> users;
+  private final RxLiveData<ViewState> users;
+  private OnSubscribeRefreshingCache<List<User>> refreshingCache;
 
   UsersViewModel(UsersRepository usersRepository, AppSchedulers appSchedulers,
-                        Navigator navigator, EventAnalytics eventAnalytics) {
+
+                 Navigator navigator, EventAnalytics eventAnalytics) {
     this.usersRepository = usersRepository;
     this.appSchedulers = appSchedulers;
     this.navigator = navigator;
     this.eventAnalytics = eventAnalytics;
 
-    users = RxLiveData.from(usersInternal());
+    Observable<ViewState> viewStateObservable = usersInternal()
+      .map((data) -> new ViewState(null, data))
+      .onErrorReturn((error) -> new ViewState(error, null))
+      .toObservable()
+      .startWith(new ViewState(null, null));
+
+    users = RxLiveData.from(viewStateObservable);
   }
 
-  public RxLiveData<List<User>> users() {
+  public LiveData<ViewState> users() {
     return users;
   }
 
   public void onRefresh() {
+    refreshingCache.invalidate();
     users.resubscribe();
   }
 
   private Single<List<User>> usersInternal() {
-    return usersRepository.getUsers(0)
+    Single<List<User>> single = usersRepository.getUsers(0)
       .subscribeOn(appSchedulers.io())
       .observeOn(appSchedulers.mainThread());
+
+    refreshingCache = new OnSubscribeRefreshingCache<>(single);
+    return Single.create(refreshingCache);
   }
 
   public void onUserClicked(User user) {
@@ -63,4 +82,49 @@ public void onUserGitHubIconClicked(User user) {
 
     navigator.launchOnWeb(Urls.user(user.login));
   }
+
+  public static final class ViewState {
+    private final Throwable error;
+    private final List<User> result;
+
+    public ViewState(Throwable error, List<User> result) {
+      this.error = error;
+      this.result = result;
+    }
+
+    public boolean isLoading() {
+      return result == null && error == null;
+    }
+
+    public Throwable error() {
+      return error;
+    }
+
+    public List<User> result() {
+      return result;
+    }
+  }
+
+  public static class OnSubscribeRefreshingCache<T> implements SingleOnSubscribe<T> {
+
+    private final AtomicBoolean refresh = new AtomicBoolean(true);
+    private final Single<T> source;
+    private volatile Single<T> current;
+
+    public OnSubscribeRefreshingCache(Single<T> source) {
+      this.source = source;
+      this.current = source;
+    }
+
+    public void invalidate() {
+      refresh.set(true);
+    }
+
+    @Override public void subscribe(SingleEmitter<T> e) throws Exception {
+      if (refresh.compareAndSet(true, false)) {
+        current = source.cache();
+      }
+      current.subscribe(e::onSuccess, e::onError);
+    }
+  }
 }