Removed repeated subscriptions to the same room when registering observers and...

Removed repeated subscriptions to the same room when registering observers and unnecessary refresh calls to the RoomFragment to update its UI

app/src/main/java/chat/rocket/android/activity/MainActivity.java

@@ -34,7 +34,6 @@ public class MainActivity extends AbstractAuthedActivity implements MainContract
   private StatusTicker statusTicker;
 
   private MainContract.Presenter presenter;
-  private RoomFragment roomFragment;
 
   @Override
   protected int getLayoutContainerForFragment() {
@@ -180,8 +179,7 @@ public class MainActivity extends AbstractAuthedActivity implements MainContract
 
   @Override
   public void showRoom(String hostname, String roomId) {
-    roomFragment = RoomFragment.create(hostname, roomId);
-    showFragment(roomFragment);
+    showFragment(RoomFragment.create(hostname, roomId));
     closeSidebarIfNeeded();
     KeyboardHelper.hideSoftKeyboard(this);
   }
@@ -224,9 +222,6 @@ public class MainActivity extends AbstractAuthedActivity implements MainContract
   @Override
   public void showConnectionOk() {
     statusTicker.updateStatus(StatusTicker.STATUS_DISMISS, null);
-    if (roomFragment != null) {
-      roomFragment.refreshRoom();
-    }
   }
 
   //TODO: consider this class to define in layouthelper for more complicated operation.

app/src/main/java/chat/rocket/android/activity/MainPresenter.java

@@ -98,8 +98,10 @@ public class MainPresenter extends BasePresenter<MainContract.View>
 
   @Override
   public void onRetryLogin() {
-    view.showConnecting();
-    connectivityManagerApi.keepAliveServer();
+    final Disposable subscription = sessionInteractor.retryLogin()
+            .subscribe();
+
+    addSubscription(subscription);
   }
 
   private void openRoom() {

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

@@ -602,8 +602,4 @@ public class RoomFragment extends AbstractChatRoomFragment implements
     edittingMessage = message;
     messageFormManager.setEditMessage(message.getMessage());
   }
-
-  public void refreshRoom() {
-    presenter.loadMessages();
-  }
 }
\ No newline at end of file

app/src/main/java/chat/rocket/android/service/RocketChatWebSocketThread.java

@@ -338,6 +338,12 @@ public class RocketChatWebSocketThread extends HandlerThread {
           registrable.register();
           listeners.add(registrable);
         }
+
+      } catch (Exception exception) {
+        RCLog.w(exception, "Failed to register listeners!!");
+      }
+    }
+
     // Register for room stream messages
     String roomId = rocketChatCache.getSelectedRoomId();
     if (roomId != null && !roomId.isEmpty()) {
@@ -347,10 +353,6 @@ public class RocketChatWebSocketThread extends HandlerThread {
       streamRoomMessage.register();
       listeners.add(streamRoomMessage);
     }
-      } catch (Exception exception) {
-        RCLog.w(exception, "Failed to register listeners!!");
-      }
-    }
   }
 
   @DebugLog

persistence-realm/src/main/java/chat/rocket/persistence/realm/repositories/RealmMessageRepository.java

@@ -135,13 +135,15 @@ public class RealmMessageRepository extends RealmRepository implements MessageRe
         () -> new Pair<>(RealmStore.getRealm(hostname), Looper.myLooper()),
         pair -> RxJavaInterop.toV2Flowable(pair.first.where(RealmMessage.class)
             .equalTo(RealmMessage.ROOM_ID, room.getRoomId())
+            .isNotNull(RealmMessage.USER)
             .findAllSorted(RealmMessage.TIMESTAMP, Sort.DESCENDING)
             .asObservable()),
         pair -> close(pair.first, pair.second)
     )
         .unsubscribeOn(AndroidSchedulers.from(Looper.myLooper()))
         .filter(it -> it.isLoaded() && it.isValid())
-        .map(this::toList));
+        .map(this::toList)
+        .distinctUntilChanged());
   }
 
   @Override