Merge pull request #388 from RocketChat/fix/filtering_get_messages_flood

Avoid Realm queries for messages with null users

android-ddp/src/main/java/chat/rocket/android_ddp/DDPClient.java

 package chat.rocket.android_ddp;
 
 import android.support.annotation.Nullable;
-import io.reactivex.Flowable;
+
 import org.json.JSONArray;
 
 import bolts.Task;
 import bolts.TaskCompletionSource;
 import chat.rocket.android_ddp.rx.RxWebSocketCallback;
+import io.reactivex.Flowable;
+import io.reactivex.Maybe;
 import okhttp3.OkHttpClient;
 
 public class DDPClient {
@@ -34,6 +36,10 @@ public class DDPClient {
     return task.getTask();
   }
 
+  public Maybe<DDPClientCallback.Base> doPing(@Nullable String id) {
+    return impl.ping(id);
+  }
+
   public Task<DDPClientCallback.RPC> rpc(String method, JSONArray params, String id,
                                          long timeoutMs) {
     TaskCompletionSource<DDPClientCallback.RPC> task = new TaskCompletionSource<>();

android-ddp/src/main/java/chat/rocket/android_ddp/DDPClientCallback.java

 package chat.rocket.android_ddp;
 
+import android.support.annotation.NonNull;
 import android.support.annotation.Nullable;
 import org.json.JSONObject;
 
@@ -53,6 +54,15 @@ public class DDPClientCallback {
       this.id = id;
     }
 
+    public static class UnMatched extends Base {
+      @NonNull public String id;
+
+      public UnMatched(DDPClient client, @NonNull String id) {
+        super(client);
+        this.id = id;
+      }
+    }
+
     public static class Timeout extends BaseException {
       public Timeout(DDPClient client) {
         super(Timeout.class, client);

android-ddp/src/main/java/chat/rocket/android_ddp/DDPClientImpl.java

@@ -17,6 +17,7 @@ import chat.rocket.android.log.RCLog;
 import chat.rocket.android_ddp.rx.RxWebSocket;
 import chat.rocket.android_ddp.rx.RxWebSocketCallback;
 import io.reactivex.Flowable;
+import io.reactivex.Maybe;
 import io.reactivex.disposables.CompositeDisposable;
 import okhttp3.OkHttpClient;
 
@@ -106,40 +107,75 @@ public class DDPClientImpl {
     }
   }
 
-  public void ping(final TaskCompletionSource<DDPClientCallback.Ping> task,
-                   @Nullable final String id) {
+  public Maybe<DDPClientCallback.Base> ping(@Nullable final String id) {
 
     final boolean requested = (TextUtils.isEmpty(id)) ?
         sendMessage("ping", null) :
         sendMessage("ping", json -> json.put("id", id));
 
     if (requested) {
-      CompositeDisposable disposables = new CompositeDisposable();
-
-      disposables.add(
-          flowable.filter(callback -> callback instanceof RxWebSocketCallback.Message)
-              .timeout(8, TimeUnit.SECONDS)
+      return flowable.filter(callback -> callback instanceof RxWebSocketCallback.Message)
               .map(callback -> ((RxWebSocketCallback.Message) callback).responseBodyString)
               .map(DDPClientImpl::toJson)
-              .subscribe(
-                  response -> {
-                    String msg = extractMsg(response);
-                    if ("pong".equals(msg)) {
-                      if (response.isNull("id")) {
-                        task.setResult(new DDPClientCallback.Ping(client, null));
-                        disposables.clear();
-                      } else {
-                        String _id = response.optString("id");
-                        if (id.equals(_id)) {
-                          task.setResult(new DDPClientCallback.Ping(client, id));
-                          disposables.clear();
-                        }
-                      }
-                      disposables.clear();
+              .filter(response -> "pong".equalsIgnoreCase(extractMsg(response)))
+              .doOnError(error -> {
+                RCLog.e(error, "Heartbeat ping[%s] xxx failed xxx", id);
+              })
+              .map(response -> {
+                String msg = extractMsg(response);
+                if ("pong".equals(msg)) {
+                  RCLog.d("pong[%s] <", id);
+                  if (response.isNull("id")) {
+                    return new DDPClientCallback.Ping(client, null);
+                  } else {
+                    String _id = response.optString("id");
+                    if (id.equals(_id)) {
+                      return new DDPClientCallback.Ping(client, _id);
+                    } else {
+                      return new DDPClientCallback.Ping.UnMatched(client, _id);
                     }
-                  },
-                  err -> task.setError(new DDPClientCallback.Ping.Timeout(client))
-              )
+                  }
+                }
+                // if we receive anything other than a pong throw an exception
+                throw new DDPClientCallback.RPC.Error(client, id, response);
+              }).firstElement();
+    } else {
+      return Maybe.error(new DDPClientCallback.Closed(client));
+    }
+  }
+
+  public void ping(final TaskCompletionSource<DDPClientCallback.Ping> task,
+                   @Nullable final String id) {
+
+    final boolean requested = (TextUtils.isEmpty(id)) ?
+            sendMessage("ping", null) :
+            sendMessage("ping", json -> json.put("id", id));
+
+    if (requested) {
+      CompositeDisposable disposables = new CompositeDisposable();
+
+      disposables.add(
+              flowable.filter(callback -> callback instanceof RxWebSocketCallback.Message)
+                      .timeout(8, TimeUnit.SECONDS)
+                      .map(callback -> ((RxWebSocketCallback.Message) callback).responseBodyString)
+                      .map(DDPClientImpl::toJson)
+                      .subscribe(
+                              response -> {
+                                String msg = extractMsg(response);
+                                if ("pong".equals(msg)) {
+                                  if (response.isNull("id")) {
+                                    task.setResult(new DDPClientCallback.Ping(client, null));
+                                  } else {
+                                    String _id = response.optString("id");
+                                    if (id.equals(_id)) {
+                                      task.setResult(new DDPClientCallback.Ping(client, id));
+                                    }
+                                  }
+                                  disposables.clear();
+                                }
+                              },
+                              err -> task.setError(new DDPClientCallback.Ping.Timeout(client))
+                      )
       );
 
       addErrorCallback(disposables, task);
@@ -368,12 +404,11 @@ public class DDPClientImpl {
     try {
       JSONObject origJson = new JSONObject().put("msg", msg);
       String msg2 = (json == null ? origJson : json.create(origJson)).toString();
-      websocket.sendText(msg2);
+      return websocket.sendText(msg2);
     } catch (Exception e) {
       RCLog.e(e);
       return false;
     }
-    return true; // ignore exception here.
   }
 
   private void sendMessage(String msg, @Nullable JSONBuilder json,
@@ -387,6 +422,9 @@ public class DDPClientImpl {
     disposables.add(
         flowable.subscribe(
             base -> {
+              if (base instanceof RxWebSocketCallback.Close) {
+                task.trySetError(new Exception(((RxWebSocketCallback.Close) base).reason));
+              }
             },
             err -> {
               task.trySetError(new Exception(err));

android-ddp/src/main/java/chat/rocket/android_ddp/rx/RxWebSocket.java

@@ -62,14 +62,20 @@ public class RxWebSocket {
               }
             }),
         BackpressureStrategy.BUFFER
-    ).delay(4, TimeUnit.SECONDS).publish();
+    ).publish();
   }
 
   public boolean sendText(String message) throws IOException {
+    if (webSocket == null) {
+        return false;
+    }
     return webSocket.send(message);
   }
 
   public boolean close(int code, String reason) throws IOException {
+    if (webSocket == null) {
+      return false;
+    }
     return webSocket.close(code, reason);
   }
 }

app/build.gradle

@@ -142,9 +142,15 @@ dependencies {
 
     compile "com.github.hotchemi:permissionsdispatcher:$permissionsdispatcherVersion"
     annotationProcessor "com.github.hotchemi:permissionsdispatcher-processor:$permissionsdispatcherVersion"
+
     compile('com.crashlytics.sdk.android:crashlytics:2.6.8@aar') {
         transitive = true;
     }
+
+    provided 'com.parse.bolts:bolts-tasks:1.4.0'
+    provided 'io.reactivex.rxjava2:rxjava:2.1.0'
+    provided 'io.reactivex:rxjava:1.3.0'
+    provided "com.github.akarnokd:rxjava2-interop:0.10.2"
 }
 
 apply plugin: 'com.google.gms.google-services'

app/src/main/AndroidManifest.xml

@@ -69,8 +69,18 @@
                 <action android:name="com.google.android.gms.iid.InstanceID"/>
             </intent-filter>
         </service>
+
+        <service android:name=".service.TaskService"
+            android:permission="com.google.android.gms.permission.BIND_NETWORK_TASK_SERVICE"
+            android:exported="true">
+            <intent-filter>
+                <action android:name="com.google.android.gms.gcm.ACTION_TASK_READY"/>
+            </intent-filter>
+        </service>
+
         <meta-data
             android:name="io.fabric.ApiKey"
             android:value="12ac6e94f850aaffcdff52001af77ca415d06a43" />
+
     </application>
 </manifest>
\ No newline at end of file

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

@@ -3,11 +3,13 @@ package chat.rocket.android;
 import android.content.Context;
 import android.content.SharedPreferences;
 
-import io.reactivex.BackpressureStrategy;
-import io.reactivex.Flowable;
+import com.hadisatrio.optional.Optional;
 
 import java.util.UUID;
 
+import io.reactivex.BackpressureStrategy;
+import io.reactivex.Flowable;
+
 /**
  * sharedpreference-based cache.
  */
@@ -51,11 +53,11 @@ public class RocketChatCache {
     return preferences.getString(KEY_PUSH_ID, null);
   }
 
-  public Flowable<String> getSelectedServerHostnamePublisher() {
+  public Flowable<Optional<String>> getSelectedServerHostnamePublisher() {
     return getValuePublisher(KEY_SELECTED_SERVER_HOSTNAME);
   }
 
-  public Flowable<String> getSelectedRoomIdPublisher() {
+  public Flowable<Optional<String>> getSelectedRoomIdPublisher() {
     return getValuePublisher(KEY_SELECTED_ROOM_ID);
   }
 
@@ -75,12 +77,13 @@ public class RocketChatCache {
     getEditor().putString(key, value).apply();
   }
 
-  private Flowable<String> getValuePublisher(final String key) {
+  private Flowable<Optional<String>> getValuePublisher(final String key) {
     return Flowable.create(emitter -> {
       SharedPreferences.OnSharedPreferenceChangeListener
           listener = (sharedPreferences, changedKey) -> {
         if (key.equals(changedKey) && !emitter.isCancelled()) {
-          emitter.onNext(getString(key, null));
+          String value = getString(key, null);
+          emitter.onNext(Optional.of(value));
         }
       };
 

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

@@ -4,6 +4,8 @@ import android.content.Intent;
 import android.os.Bundle;
 import android.support.annotation.Nullable;
 
+import com.hadisatrio.optional.Optional;
+
 import java.util.List;
 
 import chat.rocket.android.LaunchUtil;
@@ -179,6 +181,7 @@ abstract class AbstractAuthedActivity extends AbstractFragmentActivity {
   private void subscribeToConfigChanges() {
     compositeDisposable.add(
         rocketChatCache.getSelectedServerHostnamePublisher()
+            .map(Optional::get)
             .distinctUntilChanged()
             .subscribeOn(Schedulers.io())
             .observeOn(AndroidSchedulers.mainThread())
@@ -190,6 +193,7 @@ abstract class AbstractAuthedActivity extends AbstractFragmentActivity {
 
     compositeDisposable.add(
         rocketChatCache.getSelectedRoomIdPublisher()
+            .map(Optional::get)
             .distinctUntilChanged()
             .subscribeOn(Schedulers.io())
             .observeOn(AndroidSchedulers.mainThread())

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/api/DDPClientWrapper.java

@@ -15,6 +15,7 @@ import chat.rocket.android_ddp.DDPClient;
 import chat.rocket.android_ddp.DDPClientCallback;
 import chat.rocket.android_ddp.DDPSubscription;
 import io.reactivex.Flowable;
+import io.reactivex.Maybe;
 
 /**
  * DDP client wrapper.
@@ -124,4 +125,13 @@ public class DDPClientWrapper {
           }
         });
   }
+
+  /**
+   * check WebSocket connectivity with ping.
+   */
+  public Maybe<DDPClientCallback.Base> doPing() {
+    final String pingId = UUID.randomUUID().toString();
+    RCLog.d("ping[%s] >", pingId);
+    return ddpClient.doPing(pingId);
+  }
 }

app/src/main/java/chat/rocket/android/api/MethodCallHelper.java

@@ -82,6 +82,10 @@ public class MethodCallHelper {
     return task.continueWithTask(_task -> {
       if (_task.isFaulted()) {
         Exception exception = _task.getError();
+        // If wet get any error, close the socket to let the RocketChatWebSocketThread aware of it.
+        // FIXME: when rewriting the network layer we should get rid of this MethodCallHelper
+        // monolith concept. It decouples a lot the socket from the rest of the app.
+        ddpClientRef.get().close();
         if (exception instanceof MethodCall.Error) {
           String errMessageJson = exception.getMessage();
           if (TextUtils.isEmpty(errMessageJson)) {
@@ -93,7 +97,6 @@ public class MethodCallHelper {
           if (TwoStepAuthException.TYPE.equals(errType)) {
             return Task.forError(new TwoStepAuthException(errMessage));
           }
-
           return Task.forError(new Exception(errMessage));
         } else if (exception instanceof DDPClientCallback.RPC.Error) {
           String errMessage = ((DDPClientCallback.RPC.Error) exception).error.getString("message");

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/RealmBasedConnectivityManager.java

@@ -63,6 +63,7 @@ import rx.subjects.PublishSubject;
     }
   }
 
+  @DebugLog
   @Override
   public void ensureConnections() {
     for (String hostname : serverConnectivityList.keySet()) {
@@ -146,6 +147,7 @@ import rx.subjects.PublishSubject;
     return Observable.concat(Observable.from(getCurrentConnectivityList()), connectivitySubject);
   }
 
+  @DebugLog
   private Single<Boolean> connectToServerIfNeeded(String hostname, boolean forceConnect) {
     return Single.defer(() -> {
       final int connectivity = serverConnectivityList.get(hostname);
@@ -163,8 +165,8 @@ import rx.subjects.PublishSubject;
       }
 
       return connectToServer(hostname)
-          //.doOnError(RCLog::e)
-          .retryWhen(RxHelper.exponentialBackoff(3, 500, TimeUnit.MILLISECONDS));
+          .doOnError(RCLog::e)
+          .retryWhen(RxHelper.exponentialBackoff(Integer.MAX_VALUE, 500, TimeUnit.MILLISECONDS));
     });
   }
 
@@ -191,7 +193,7 @@ import rx.subjects.PublishSubject;
     });
   }
 
-
+  @DebugLog
   private Single<Boolean> waitForConnected(String hostname) {
     return connectivitySubject
         .filter(serverConnectivity -> hostname.equals(serverConnectivity.hostname))
@@ -207,6 +209,7 @@ import rx.subjects.PublishSubject;
                 : Single.error(new ServerConnectivity.DisconnectedException()));
   }
 
+  @DebugLog
   private Single<Boolean> waitForDisconnected(String hostname) {
     return connectivitySubject
         .filter(serverConnectivity -> hostname.equals(serverConnectivity.hostname))

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

+package chat.rocket.android.service;
+
+import com.google.android.gms.gcm.GcmNetworkManager;
+import com.google.android.gms.gcm.GcmTaskService;
+import com.google.android.gms.gcm.TaskParams;
+
+public class TaskService extends GcmTaskService {
+
+    public static final String TAG_KEEP_ALIVE = "TAG_KEEP_ALIVE";
+
+    @Override
+    public int onRunTask(TaskParams taskParams) {
+        switch (taskParams.getTag()) {
+            case TAG_KEEP_ALIVE:
+                ConnectivityManager.getInstance(getApplicationContext()).keepAliveServer();
+                return GcmNetworkManager.RESULT_SUCCESS;
+            default:
+                return GcmNetworkManager.RESULT_FAILURE;
+        }
+    }
+}

app/src/main/java/chat/rocket/android/service/ddp/base/AbstractBaseSubscriber.java

@@ -20,7 +20,7 @@ abstract class AbstractBaseSubscriber extends AbstractDDPDocEventSubscriber {
 
   @Override
   protected final boolean shouldTruncateTableOnInitialize() {
-    return true;
+    return false;
   }
 
   protected abstract String getSubscriptionCallbackName();

app/src/main/java/chat/rocket/android/service/internal/AbstractRocketChatCacheObserver.java

@@ -2,6 +2,8 @@ package chat.rocket.android.service.internal;
 
 import android.content.Context;
 
+import com.hadisatrio.optional.Optional;
+
 import chat.rocket.android.log.RCLog;
 import io.reactivex.disposables.CompositeDisposable;
 
@@ -48,6 +50,7 @@ public abstract class AbstractRocketChatCacheObserver implements Registrable {
     compositeDisposable.add(
         new RocketChatCache(context)
             .getSelectedRoomIdPublisher()
+            .map(Optional::get)
             .subscribe(this::updateRoomIdWith, RCLog::e)
     );
   }

app/src/main/java/chat/rocket/android/service/internal/StreamRoomMessageManager.java

@@ -4,6 +4,7 @@ import android.content.Context;
 import android.os.Handler;
 import android.os.Looper;
 
+import chat.rocket.android.RocketChatCache;
 import chat.rocket.persistence.realm.RealmHelper;
 import chat.rocket.android.service.DDPClientRef;
 import chat.rocket.android.service.Registrable;
@@ -19,6 +20,7 @@ public class StreamRoomMessageManager implements Registrable {
   private final DDPClientRef ddpClientRef;
   private final AbstractRocketChatCacheObserver cacheObserver;
   private final Handler handler;
+  private final RocketChatCache rocketChatCache;
   private StreamRoomMessage streamRoomMessage;
 
   public StreamRoomMessageManager(Context context, String hostname,
@@ -27,6 +29,7 @@ public class StreamRoomMessageManager implements Registrable {
     this.hostname = hostname;
     this.realmHelper = realmHelper;
     this.ddpClientRef = ddpClientRef;
+    this.rocketChatCache = new RocketChatCache(context);
 
     cacheObserver = new AbstractRocketChatCacheObserver(context, realmHelper) {
       @Override
@@ -57,6 +60,11 @@ public class StreamRoomMessageManager implements Registrable {
   @Override
   public void register() {
     cacheObserver.register();
+    String selectedRoomId = rocketChatCache.getSelectedRoomId();
+    if (selectedRoomId == null) {
+      return;
+    }
+    registerStreamNotifyMessage(selectedRoomId);
   }
 
   @Override

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