Rolling our own bg looper, working the RoomFragment

app/src/main/java/chat/rocket/android/BackgroundLooper.java

+package chat.rocket.android;
+
+
+import android.os.HandlerThread;
+import android.os.Looper;
+import android.os.Process;
+
+public class BackgroundLooper {
+
+  private static HandlerThread handlerThread;
+
+  public static Looper get() {
+    if (handlerThread == null) {
+      handlerThread =
+          new HandlerThread("BackgroundHandlerThread", Process.THREAD_PRIORITY_BACKGROUND);
+      handlerThread.start();
+    }
+
+    return handlerThread.getLooper();
+  }
+}

app/src/main/java/chat/rocket/android/fragment/chatroom/RoomContract.java

+package chat.rocket.android.fragment.chatroom;
+
+import android.support.annotation.NonNull;
+
+import chat.rocket.android.model.core.Room;
+
+public interface RoomContract {
+
+  interface View {
+
+    void render(Room room);
+
+    void updateHistoryState(boolean hasNext, boolean isLoaded);
+  }
+
+  interface Presenter {
+    void bindView(@NonNull View view);
+
+    void release();
+  }
+}

app/src/main/java/chat/rocket/android/fragment/chatroom/RoomFragment.java

@@ -51,14 +51,15 @@ import chat.rocket.android.layouthelper.extra_action.upload.ImageUploadActionIte
 import chat.rocket.android.layouthelper.extra_action.upload.VideoUploadActionItem;
 import chat.rocket.android.log.RCLog;
 import chat.rocket.android.model.SyncState;
+import chat.rocket.android.model.core.Room;
 import chat.rocket.android.model.ddp.RealmMessage;
 import chat.rocket.android.model.ddp.RoomSubscription;
 import chat.rocket.android.model.ddp.RealmUser;
 import chat.rocket.android.model.internal.LoadMessageProcedure;
 import chat.rocket.android.model.internal.Session;
+import chat.rocket.android.repositories.RealmRoomRepository;
 import chat.rocket.persistence.realm.RealmHelper;
 import chat.rocket.persistence.realm.RealmModelListAdapter;
-import chat.rocket.persistence.realm.RealmObjectObserver;
 import chat.rocket.persistence.realm.RealmStore;
 import chat.rocket.android.service.ConnectivityManager;
 import chat.rocket.android.widget.internal.ExtraActionPickerDialogFragment;
@@ -72,7 +73,7 @@ import permissions.dispatcher.RuntimePermissions;
 @RuntimePermissions
 public class RoomFragment extends AbstractChatRoomFragment
     implements OnBackPressListener, ExtraActionPickerDialogFragment.Callback,
-    RealmModelListAdapter.OnItemClickListener<PairedMessage> {
+    RealmModelListAdapter.OnItemClickListener<PairedMessage>, RoomContract.View {
 
   private static final int DIALOG_ID = 1;
   private static final String HOSTNAME = "hostname";
@@ -81,11 +82,9 @@ public class RoomFragment extends AbstractChatRoomFragment
   private String hostname;
   private RealmHelper realmHelper;
   private String roomId;
-  private RealmObjectObserver<RoomSubscription> roomObserver;
   private String userId;
   private String token;
   private LoadMoreScrollListener scrollListener;
-  private RealmObjectObserver<LoadMessageProcedure> procedureObserver;
   private MessageFormManager messageFormManager;
   private RecyclerViewAutoScrollManager autoScrollManager;
   private AbstractNewMessageIndicatorManager newMessageIndicatorManager;
@@ -94,6 +93,8 @@ public class RoomFragment extends AbstractChatRoomFragment
 
   private List<AbstractExtraActionItem> extraActionItems;
 
+  private RoomContract.Presenter presenter;
+
   public RoomFragment() {
   }
 
@@ -112,8 +113,10 @@ public class RoomFragment extends AbstractChatRoomFragment
     Bundle args = new Bundle();
     args.putString(HOSTNAME, hostname);
     args.putString(ROOM_ID, roomId);
+
     RoomFragment fragment = new RoomFragment();
     fragment.setArguments(args);
+
     return fragment;
   }
 
@@ -123,21 +126,20 @@ public class RoomFragment extends AbstractChatRoomFragment
 
     Bundle args = getArguments();
     hostname = args.getString(HOSTNAME);
-    realmHelper = RealmStore.get(hostname);
     roomId = args.getString(ROOM_ID);
+
+    presenter = new RoomPresenter(
+        roomId,
+        new RealmRoomRepository(hostname)
+    );
+
+    realmHelper = RealmStore.get(hostname);
+
     userId = realmHelper.executeTransactionForRead(realm ->
         RealmUser.queryCurrentUser(realm).findFirst()).getId();
+
     token = realmHelper.executeTransactionForRead(realm ->
         Session.queryDefaultSession(realm).findFirst()).getToken();
-    roomObserver = realmHelper
-        .createObjectObserver(
-            realm -> realm.where(RoomSubscription.class).equalTo(RoomSubscription.ROOM_ID, roomId))
-        .setOnUpdateListener(this::onRenderRoom);
-
-    procedureObserver = realmHelper
-        .createObjectObserver(realm ->
-            realm.where(LoadMessageProcedure.class).equalTo(LoadMessageProcedure.ID, roomId))
-        .setOnUpdateListener(this::onUpdateLoadMessageProcedure);
 
     if (savedInstanceState == null) {
       initialRequest();
@@ -347,12 +349,8 @@ public class RoomFragment extends AbstractChatRoomFragment
     }
   }
 
-  private void onRenderRoom(RoomSubscription roomSubscription) {
-    if (roomSubscription == null) {
-      return;
-    }
-
-    String type = roomSubscription.getType();
+  private void onRenderRoom(Room room) {
+    String type = room.getType();
     if (RoomSubscription.TYPE_CHANNEL.equals(type)) {
       setToolbarRoomIcon(R.drawable.ic_hashtag_gray_24dp);
     } else if (RoomSubscription.TYPE_PRIVATE.equals(type)) {
@@ -362,34 +360,15 @@ public class RoomFragment extends AbstractChatRoomFragment
     } else {
       setToolbarRoomIcon(0);
     }
-    setToolbarTitle(roomSubscription.getName());
+    setToolbarTitle(room.getName());
 
-    boolean unreadMessageExists = roomSubscription.isAlert();
+    boolean unreadMessageExists = room.isAlert();
     if (newMessageIndicatorManager != null && previousUnreadMessageExists && !unreadMessageExists) {
       newMessageIndicatorManager.reset();
     }
     previousUnreadMessageExists = unreadMessageExists;
   }
 
-  private void onUpdateLoadMessageProcedure(LoadMessageProcedure procedure) {
-    if (procedure == null) {
-      return;
-    }
-    RecyclerView listView = (RecyclerView) rootView.findViewById(R.id.recyclerview);
-    if (listView != null && listView.getAdapter() instanceof MessageListAdapter) {
-      MessageListAdapter adapter = (MessageListAdapter) listView.getAdapter();
-      final int syncState = procedure.getSyncState();
-      final boolean hasNext = procedure.hasNext();
-      RCLog.d("hasNext: %s syncstate: %d", hasNext, syncState);
-      if (syncState == SyncState.SYNCED || syncState == SyncState.FAILED) {
-        scrollListener.setLoadingDone();
-        adapter.updateFooter(hasNext, true);
-      } else {
-        adapter.updateFooter(hasNext, false);
-      }
-    }
-  }
-
   private void initialRequest() {
     realmHelper.executeTransaction(realm -> {
       realm.createOrUpdateObjectFromJson(LoadMessageProcedure.class, new JSONObject()
@@ -439,15 +418,13 @@ public class RoomFragment extends AbstractChatRoomFragment
   @Override
   public void onResume() {
     super.onResume();
-    roomObserver.sub();
-    procedureObserver.sub();
+    presenter.bindView(this);
     closeSideMenuIfNeeded();
   }
 
   @Override
   public void onPause() {
-    procedureObserver.unsub();
-    roomObserver.unsub();
+    presenter.release();
     super.onPause();
   }
 
@@ -539,4 +516,23 @@ public class RoomFragment extends AbstractChatRoomFragment
           return null;
         });
   }
+
+  @Override
+  public void render(Room room) {
+    onRenderRoom(room);
+  }
+
+  @Override
+  public void updateHistoryState(boolean hasNext, boolean isLoaded) {
+    RecyclerView listView = (RecyclerView) rootView.findViewById(R.id.recyclerview);
+    if (listView == null || !(listView.getAdapter() instanceof MessageListAdapter)) {
+      return;
+    }
+
+    MessageListAdapter adapter = (MessageListAdapter) listView.getAdapter();
+    if (isLoaded) {
+      scrollListener.setLoadingDone();
+    }
+    adapter.updateFooter(hasNext, isLoaded);
+  }
 }

app/src/main/java/chat/rocket/android/fragment/chatroom/RoomPresenter.java

+package chat.rocket.android.fragment.chatroom;
+
+import android.support.annotation.NonNull;
+
+import chat.rocket.android.BackgroundLooper;
+import chat.rocket.android.model.SyncState;
+import chat.rocket.android.repositories.RoomRepository;
+import rx.Subscription;
+import rx.android.schedulers.AndroidSchedulers;
+import rx.subscriptions.CompositeSubscription;
+
+public class RoomPresenter implements RoomContract.Presenter {
+
+  private final String roomId;
+  private final RoomRepository roomRepository;
+
+  private CompositeSubscription compositeSubscription = new CompositeSubscription();
+  private RoomContract.View view;
+
+  public RoomPresenter(String roomId, RoomRepository roomRepository) {
+    this.roomId = roomId;
+    this.roomRepository = roomRepository;
+  }
+
+  @Override
+  public void bindView(@NonNull RoomContract.View view) {
+    this.view = view;
+
+    getRoomInfo();
+    getRoomHistoryStateInfo();
+  }
+
+  @Override
+  public void release() {
+    compositeSubscription.unsubscribe();
+    this.view = null;
+  }
+
+  private void getRoomInfo() {
+    final Subscription subscription = roomRepository.getById(roomId)
+        .distinctUntilChanged()
+        .subscribeOn(AndroidSchedulers.from(BackgroundLooper.get()))
+        .observeOn(AndroidSchedulers.mainThread())
+        .subscribe(
+            room -> view.render(room)
+        );
+
+    compositeSubscription.add(subscription);
+  }
+
+  private void getRoomHistoryStateInfo() {
+    final Subscription subscription = roomRepository.getHistoryStateByRoomId(roomId)
+        .distinctUntilChanged()
+        .subscribeOn(AndroidSchedulers.from(BackgroundLooper.get()))
+        .observeOn(AndroidSchedulers.mainThread())
+        .subscribe(
+            roomHistoryState -> {
+              int syncState = roomHistoryState.getSyncState();
+              view.updateHistoryState(
+                  !roomHistoryState.isComplete(),
+                  syncState == SyncState.SYNCED || syncState == SyncState.FAILED
+              );
+            }
+        );
+
+    compositeSubscription.add(subscription);
+  }
+}

app/src/main/java/chat/rocket/android/fragment/sidebar/SidebarMainFragment.java

@@ -28,7 +28,6 @@ import chat.rocket.android.renderer.UserRenderer;
 import chat.rocket.android.repositories.RealmRoomRepository;
 import chat.rocket.android.repositories.RealmUserRepository;
 import chat.rocket.android.widget.RocketChatAvatar;
-import chat.rocket.persistence.realm.RealmStore;
 
 public class SidebarMainFragment extends AbstractFragment implements SidebarMainContract.View {
 
@@ -65,22 +64,22 @@ public class SidebarMainFragment extends AbstractFragment implements SidebarMain
 
     presenter = new SidebarMainPresenter(
         hostname,
-        new RealmRoomRepository(RealmStore.getRealm(hostname)),
-        new RealmUserRepository(RealmStore.getRealm(hostname)),
+        new RealmRoomRepository(hostname),
+        new RealmUserRepository(hostname),
         TextUtils.isEmpty(hostname) ? null : new MethodCallHelper(getContext(), hostname)
     );
   }
 
   @Override
-  public void onViewCreated(View view, @Nullable Bundle savedInstanceState) {
-    super.onViewCreated(view, savedInstanceState);
+  public void onResume() {
+    super.onResume();
     presenter.bindView(this);
   }
 
   @Override
-  public void onDestroyView() {
+  public void onPause() {
     presenter.release();
-    super.onDestroyView();
+    super.onPause();
   }
 
   @Override

app/src/main/java/chat/rocket/android/fragment/sidebar/SidebarMainPresenter.java

@@ -2,6 +2,7 @@ package chat.rocket.android.fragment.sidebar;
 
 import android.support.annotation.NonNull;
 
+import chat.rocket.android.BackgroundLooper;
 import chat.rocket.android.api.MethodCallHelper;
 import chat.rocket.android.helper.LogcatIfError;
 import chat.rocket.android.helper.TextUtils;
@@ -9,6 +10,8 @@ import chat.rocket.android.model.core.User;
 import chat.rocket.android.repositories.RoomRepository;
 import chat.rocket.android.repositories.UserRepository;
 import rx.Subscription;
+import rx.android.schedulers.AndroidSchedulers;
+import rx.schedulers.Schedulers;
 import rx.subscriptions.CompositeSubscription;
 
 public class SidebarMainPresenter implements SidebarMainContract.Presenter {
@@ -81,6 +84,8 @@ public class SidebarMainPresenter implements SidebarMainContract.Presenter {
   private void subscribeToRooms() {
     final Subscription subscription = roomRepository.getOpenRooms()
         .distinctUntilChanged()
+        .subscribeOn(AndroidSchedulers.from(BackgroundLooper.get()))
+        .observeOn(AndroidSchedulers.mainThread())
         .subscribe(
             rooms -> view.showRoomList(rooms)
         );
@@ -91,6 +96,8 @@ public class SidebarMainPresenter implements SidebarMainContract.Presenter {
   private void subscribeToUser() {
     final Subscription subscription = userRepository.getCurrentUser()
         .distinctUntilChanged()
+        .subscribeOn(AndroidSchedulers.from(BackgroundLooper.get()))
+        .observeOn(AndroidSchedulers.mainThread())
         .subscribe(user -> view.showUser(user));
 
     compositeSubscription.add(subscription);

app/src/main/java/chat/rocket/android/model/core/RoomHistoryState.java

+package chat.rocket.android.model.core;
+
+import com.google.auto.value.AutoValue;
+
+@AutoValue
+public abstract class RoomHistoryState {
+
+  public abstract String getRoomId();
+
+  public abstract int getSyncState();
+
+  public abstract boolean isReset();
+
+  public abstract long getTimestamp();
+
+  public abstract int getCount();
+
+  public abstract boolean isComplete();
+
+  public static Builder builder() {
+    return new AutoValue_RoomHistoryState.Builder();
+  }
+
+  @AutoValue.Builder
+  public abstract static class Builder {
+
+    public abstract Builder setRoomId(String roomId);
+
+    public abstract Builder setSyncState(int syncState);
+
+    public abstract Builder setReset(boolean isReset);
+
+    public abstract Builder setTimestamp(long timestamp);
+
+    public abstract Builder setCount(int count);
+
+    public abstract Builder setComplete(boolean isComplete);
+
+    public abstract RoomHistoryState build();
+  }
+}

app/src/main/java/chat/rocket/android/model/internal/LoadMessageProcedure.java

@@ -3,6 +3,8 @@ package chat.rocket.android.model.internal;
 import io.realm.RealmObject;
 import io.realm.annotations.PrimaryKey;
 
+import chat.rocket.android.model.core.RoomHistoryState;
+
 /**
  * Load messages in the room.
  */
@@ -72,4 +74,15 @@ public class LoadMessageProcedure extends RealmObject {
   public void setHasNext(boolean hasNext) {
     this.hasNext = hasNext;
   }
+
+  public RoomHistoryState asRoomHistoryState() {
+    return RoomHistoryState.builder()
+        .setRoomId(roomId)
+        .setSyncState(syncstate)
+        .setReset(reset)
+        .setTimestamp(timestamp)
+        .setCount(count)
+        .setComplete(!hasNext)
+        .build();
+  }
 }

app/src/main/java/chat/rocket/android/repositories/RealmRoomRepository.java

@@ -6,26 +6,75 @@ import io.realm.RealmResults;
 import java.util.ArrayList;
 import java.util.List;
 import chat.rocket.android.model.core.Room;
+import chat.rocket.android.model.core.RoomHistoryState;
 import chat.rocket.android.model.ddp.RoomSubscription;
+import chat.rocket.android.model.internal.LoadMessageProcedure;
+import chat.rocket.persistence.realm.RealmStore;
 import rx.Observable;
 
 public class RealmRoomRepository implements RoomRepository {
 
-  private final Realm realm;
+  private final String hostname;
 
-  public RealmRoomRepository(Realm realm) {
-    this.realm = realm;
+  public RealmRoomRepository(String hostname) {
+    this.hostname = hostname;
   }
 
   @Override
   public Observable<List<Room>> getOpenRooms() {
+    return Observable.defer(() -> {
+      final Realm realm = RealmStore.getRealm(hostname);
+
+      if (realm == null) {
+        return Observable.just(null);
+      }
+
       return realm.where(RoomSubscription.class)
           .equalTo(RoomSubscription.OPEN, true)
-        .findAllAsync()
+          .findAll()
           .asObservable()
           .filter(roomSubscriptions -> roomSubscriptions != null && roomSubscriptions.isLoaded()
               && roomSubscriptions.isValid())
           .map(roomSubscriptions -> toList(roomSubscriptions));
+    });
+  }
+
+  @Override
+  public Observable<Room> getById(String roomId) {
+    return Observable.defer(() -> {
+      final Realm realm = RealmStore.getRealm(hostname);
+
+      if (realm == null) {
+        return Observable.just(null);
+      }
+
+      return realm.where(RoomSubscription.class)
+          .equalTo(RoomSubscription.ROOM_ID, roomId)
+          .findFirstAsync()
+          .<RoomSubscription>asObservable()
+          .filter(roomSubscription -> roomSubscription != null && roomSubscription.isLoaded()
+              && roomSubscription.isValid())
+          .map(roomSubscription -> roomSubscription.asRoom());
+    });
+  }
+
+  @Override
+  public Observable<RoomHistoryState> getHistoryStateByRoomId(String roomId) {
+    return Observable.defer(() -> {
+      final Realm realm = RealmStore.getRealm(hostname);
+
+      if (realm == null) {
+        return Observable.just(null);
+      }
+
+      return realm.where(LoadMessageProcedure.class)
+          .equalTo(LoadMessageProcedure.ID, roomId)
+          .findFirstAsync()
+          .<LoadMessageProcedure>asObservable()
+          .filter(loadMessageProcedure -> loadMessageProcedure != null
+              && loadMessageProcedure.isLoaded() && loadMessageProcedure.isValid())
+          .map(loadMessageProcedure -> loadMessageProcedure.asRoomHistoryState());
+    });
   }
 
   private List<Room> toList(RealmResults<RoomSubscription> roomSubscriptions) {

app/src/main/java/chat/rocket/android/repositories/RealmUserRepository.java

@@ -4,23 +4,38 @@ import io.realm.Realm;
 
 import chat.rocket.android.model.core.User;
 import chat.rocket.android.model.ddp.RealmUser;
+import chat.rocket.persistence.realm.RealmStore;
 import rx.Observable;
 
 public class RealmUserRepository implements UserRepository {
 
-  private final Realm realm;
+  private final String hostname;
 
-  public RealmUserRepository(Realm realm) {
-    this.realm = realm;
+  public RealmUserRepository(String hostname) {
+    this.hostname = hostname;
   }
 
   @Override
   public Observable<User> getCurrentUser() {
-    return realm.where(RealmUser.class)
+    return Observable.defer(() -> {
+      final Realm realm = RealmStore.getRealm(hostname);
+
+      if (realm == null) {
+        return Observable.just(null);
+      }
+
+      final RealmUser realmUser = realm.where(RealmUser.class)
           .isNotEmpty(RealmUser.EMAILS)
-        .findFirstAsync()
+          .findFirst();
+
+      if (realmUser == null) {
+        return Observable.just(null);
+      }
+
+      return realmUser
           .<RealmUser>asObservable()
-        .filter(realmUser -> realmUser != null && realmUser.isLoaded() && realmUser.isValid())
-        .map(realmUser -> realmUser.asUser());
+          .filter(it -> it != null && it.isLoaded() && it.isValid())
+          .map(it -> it.asUser());
+    });
   }
 }

app/src/main/java/chat/rocket/android/repositories/RoomRepository.java

@@ -2,9 +2,14 @@ package chat.rocket.android.repositories;
 
 import java.util.List;
 import chat.rocket.android.model.core.Room;
+import chat.rocket.android.model.core.RoomHistoryState;
 import rx.Observable;
 
 public interface RoomRepository {
 
   Observable<List<Room>> getOpenRooms();
+
+  Observable<Room> getById(String roomId);
+
+  Observable<RoomHistoryState> getHistoryStateByRoomId(String roomId);
 }