implement loading message for room.

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

@@ -3,10 +3,15 @@ package chat.rocket.android.fragment.chatroom;
 import android.os.Bundle;
 import android.support.annotation.Nullable;
 import chat.rocket.android.R;
+import chat.rocket.android.helper.LogcatIfError;
+import chat.rocket.android.model.LoadMessageProcedure;
+import chat.rocket.android.model.SyncState;
 import chat.rocket.android.model.ddp.RoomSubscription;
 import io.realm.Realm;
 import io.realm.RealmQuery;
 import jp.co.crowdworks.realm_java_helpers.RealmObjectObserver;
+import jp.co.crowdworks.realm_java_helpers_bolts.RealmHelperBolts;
+import org.json.JSONObject;
 
 /**
  * Chat room screen.
@@ -42,6 +47,18 @@ public class RoomFragment extends AbstractChatRoomFragment {
 
   @Override protected void onSetupView() {
 
+    // TODO: just a sample!!
+    RealmHelperBolts.executeTransaction(realm -> {
+      final String serverConfigId = realm.where(RoomSubscription.class)
+          .equalTo("rid", roomId).findFirst().getServerConfigId();
+      realm.createOrUpdateObjectFromJson(LoadMessageProcedure.class, new JSONObject()
+          .put("serverConfigId", serverConfigId)
+          .put("roomId", roomId)
+          .put("syncstate", SyncState.NOT_SYNCED)
+          .put("count", 50)
+          .put("reset", true));
+      return null;
+    }).continueWith(new LogcatIfError());
   }
 
   private RealmObjectObserver<RoomSubscription> roomObserver =

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

@@ -4,8 +4,9 @@ import android.util.Patterns;
 import bolts.Continuation;
 import bolts.Task;
 import chat.rocket.android.model.MethodCall;
-import chat.rocket.android.model.ddp.RoomSubscription;
 import chat.rocket.android.model.ServerConfig;
+import chat.rocket.android.model.ddp.Message;
+import chat.rocket.android.model.ddp.RoomSubscription;
 import chat.rocket.android.ws.RocketChatWebSocketAPI;
 import chat.rocket.android_ddp.DDPClientCallback;
 import java.util.UUID;
@@ -219,4 +220,32 @@ public class MethodCallHelper {
           }
         });
   }
+
+  /**
+   * Load messages for room.
+   */
+  public Task<Void> loadHistory(final String roomId, final long timestamp,
+      final int count, final long lastSeen) {
+    return call("loadHistory", TIMEOUT_MS, params -> params
+            .put(roomId)
+            .put(timestamp > 0 ? new JSONObject().put("$date", timestamp) : JSONObject.NULL)
+            .put(count)
+            .put(lastSeen > 0 ? new JSONObject().put("$date", lastSeen) : JSONObject.NULL),
+        task -> {
+          JSONObject result = task.getResult();
+          final JSONArray messages = result.getJSONArray("messages");
+          for (int i = 0; i < messages.length(); i++) {
+            Message.customizeJson(messages.getJSONObject(i));
+          }
+
+          return RealmHelperBolts.executeTransaction(realm -> {
+            if (timestamp == 0) {
+              realm.where(Message.class).equalTo("rid", roomId).findAll().deleteAllFromRealm();
+            }
+            realm.createOrUpdateAllFromJson(Message.class, messages);
+            return null;
+          });
+        });
+  }
+
 }

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

+package chat.rocket.android.model;
+
+import io.realm.RealmObject;
+import io.realm.annotations.PrimaryKey;
+
+/**
+ * Load messages in the room.
+ */
+public class LoadMessageProcedure extends RealmObject {
+  @PrimaryKey private String roomId;
+  private String serverConfigId;
+  private int syncstate;
+
+  private boolean reset;
+  private long timestamp;
+  private int count;
+
+  private boolean hasNext;
+
+  public String getRoomId() {
+    return roomId;
+  }
+
+  public void setRoomId(String roomId) {
+    this.roomId = roomId;
+  }
+
+  public String getServerConfigId() {
+    return serverConfigId;
+  }
+
+  public void setServerConfigId(String serverConfigId) {
+    this.serverConfigId = serverConfigId;
+  }
+
+  public int getSyncstate() {
+    return syncstate;
+  }
+
+  public void setSyncstate(int syncstate) {
+    this.syncstate = syncstate;
+  }
+
+  public boolean isReset() {
+    return reset;
+  }
+
+  public void setReset(boolean reset) {
+    this.reset = reset;
+  }
+
+  public long getTimestamp() {
+    return timestamp;
+  }
+
+  public void setTimestamp(long timestamp) {
+    this.timestamp = timestamp;
+  }
+
+  public int getCount() {
+    return count;
+  }
+
+  public void setCount(int count) {
+    this.count = count;
+  }
+
+  public boolean isHasNext() {
+    return hasNext;
+  }
+
+  public void setHasNext(boolean hasNext) {
+    this.hasNext = hasNext;
+  }
+}

app/src/main/java/chat/rocket/android/model/ddp/Message.java

+package chat.rocket.android.model.ddp;
+
+import io.realm.RealmList;
+import io.realm.RealmObject;
+import io.realm.annotations.PrimaryKey;
+import org.json.JSONException;
+import org.json.JSONObject;
+
+/**
+ * Message.
+ */
+@SuppressWarnings({"PMD.ShortClassName", "PMD.ShortVariable",
+    "PMD.MethodNamingConventions", "PMD.VariableNamingConventions"})
+public class Message extends RealmObject {
+  @PrimaryKey private String _id;
+  private String t; //type:
+  private String rid; //roomId.
+  private long ts;
+  private String msg;
+  private User u;
+  private boolean groupable;
+  private RealmList<MessageAttachment> attachments;
+  private RealmList<MessageUrl> urls;
+
+
+  public static JSONObject customizeJson(JSONObject messageJson) throws JSONException {
+    long ts = messageJson.getJSONObject("ts").getLong("$date");
+    messageJson.remove("ts");
+    messageJson.put("ts", ts);
+
+    return messageJson;
+  }
+}

app/src/main/java/chat/rocket/android/model/ddp/MessageAttachment.java

+package chat.rocket.android.model.ddp;
+
+import io.realm.RealmObject;
+
+/**
+ * Attachment.
+ */
+@SuppressWarnings({"PMD.ShortClassName", "PMD.ShortVariable",
+    "PMD.MethodNamingConventions", "PMD.VariableNamingConventions"})
+public class MessageAttachment extends RealmObject {
+  private String title;
+  private String title_url;
+  private String image_url;
+  private String image_type;
+  private String image_size;
+}

app/src/main/java/chat/rocket/android/model/ddp/MessageUrl.java

+package chat.rocket.android.model.ddp;
+
+import io.realm.RealmObject;
+import io.realm.annotations.PrimaryKey;
+
+/**
+ * Url.
+ */
+@SuppressWarnings({"PMD.ShortClassName", "PMD.ShortVariable",
+    "PMD.MethodNamingConventions", "PMD.VariableNamingConventions"})
+public class MessageUrl extends RealmObject {
+  @PrimaryKey private String url;
+  private String parsedUrl;
+}

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

@@ -11,6 +11,7 @@ import chat.rocket.android.helper.TextUtils;
 import chat.rocket.android.model.ServerConfig;
 import chat.rocket.android.service.ddp.ActiveUsersSubscriber;
 import chat.rocket.android.service.ddp.LoginServiceConfigurationSubscriber;
+import chat.rocket.android.service.observer.LoadMessageProcedureObserver;
 import chat.rocket.android.service.observer.MethodCallObserver;
 import chat.rocket.android.service.observer.SessionObserver;
 import chat.rocket.android.service.observer.TokenLoginObserver;
@@ -34,7 +35,8 @@ public class RocketChatWebSocketThread extends HandlerThread {
       ActiveUsersSubscriber.class,
       TokenLoginObserver.class,
       MethodCallObserver.class,
-      SessionObserver.class
+      SessionObserver.class,
+      LoadMessageProcedureObserver.class
   };
   private final Context appContext;
   private final String serverConfigId;

app/src/main/java/chat/rocket/android/service/observer/LoadMessageProcedureObserver.java

+package chat.rocket.android.service.observer;
+
+import android.content.Context;
+import bolts.Task;
+import chat.rocket.android.helper.MethodCallHelper;
+import chat.rocket.android.model.LoadMessageProcedure;
+import chat.rocket.android.model.SyncState;
+import chat.rocket.android.ws.RocketChatWebSocketAPI;
+import io.realm.Realm;
+import io.realm.RealmResults;
+import java.util.List;
+import jp.co.crowdworks.realm_java_helpers_bolts.RealmHelperBolts;
+import org.json.JSONObject;
+import timber.log.Timber;
+
+/**
+ * Background process for loading messages.
+ */
+public class LoadMessageProcedureObserver extends AbstractModelObserver<LoadMessageProcedure> {
+
+  private final MethodCallHelper methodCall;
+
+  public LoadMessageProcedureObserver(Context context, String serverConfigId,
+      RocketChatWebSocketAPI api) {
+    super(context, serverConfigId, api);
+    methodCall = new MethodCallHelper(serverConfigId, api);
+  }
+
+  @Override protected RealmResults<LoadMessageProcedure> queryItems(Realm realm) {
+    return realm.where(LoadMessageProcedure.class)
+        .equalTo("serverConfigId", serverConfigId)
+        .equalTo("syncstate", SyncState.NOT_SYNCED)
+        .findAll();
+  }
+
+  @Override protected void onCollectionChanged(List<LoadMessageProcedure> list) {
+    if (list == null || list.isEmpty()) {
+      return;
+    }
+
+    LoadMessageProcedure procedure = list.get(0);
+    final String roomId = procedure.getRoomId();
+    final boolean isReset = procedure.isReset();
+    final long timestamp = procedure.getTimestamp();
+    final int count = procedure.getCount();
+    final long lastSeen = 0; // TODO: Not implemented yet.
+
+    RealmHelperBolts.executeTransaction(realm ->
+        realm.createOrUpdateObjectFromJson(LoadMessageProcedure.class, new JSONObject()
+            .put("roomId", roomId)
+            .put("syncstate", SyncState.SYNCING))
+    ).onSuccessTask(task ->
+        methodCall.loadHistory(roomId, isReset ? 0 : timestamp, count, lastSeen)
+            .onSuccessTask(_task ->
+                RealmHelperBolts.executeTransaction(realm ->
+                    realm.createOrUpdateObjectFromJson(LoadMessageProcedure.class, new JSONObject()
+                        .put("roomId", roomId)
+                        .put("syncstate", SyncState.SYNCED))))
+    ).continueWithTask(task -> {
+      if (task.isFaulted()) {
+        Timber.w(task.getError());
+        return RealmHelperBolts.executeTransaction(realm ->
+            realm.createOrUpdateObjectFromJson(LoadMessageProcedure.class, new JSONObject()
+                .put("roomId", roomId)
+                .put("syncstate", SyncState.FAILED)));
+      } else {
+        return Task.forResult(null);
+      }
+    });
+  }
+}

app/src/main/java/chat/rocket/android/service/observer/MethodCallObserver.java

@@ -76,9 +76,9 @@ public class MethodCallObserver extends AbstractModelObserver<MethodCall> {
                   .put("resultJson", result == null ? null : result.toString()));
             })
         )
-    ).continueWith(task -> {
+    ).continueWithTask(task -> {
       if (task.isFaulted()) {
-        RealmHelperBolts.executeTransaction(realm -> {
+        return RealmHelperBolts.executeTransaction(realm -> {
           Exception exception = task.getError();
           final String errMessage = (exception instanceof DDPClientCallback.RPC.Error)
               ? ((DDPClientCallback.RPC.Error) exception).error.toString()
@@ -88,9 +88,9 @@ public class MethodCallObserver extends AbstractModelObserver<MethodCall> {
               .put("syncstate", SyncState.FAILED)
               .put("resultJson", errMessage));
           return null;
-        }).continueWith(new LogcatIfError());
+        });
       }
-      return null;
-    });
+      return task;
+    }).continueWith(new LogcatIfError());
   }
 }

app/src/main/java/chat/rocket/android/service/observer/SessionObserver.java

@@ -16,10 +16,15 @@ import jp.co.crowdworks.realm_java_helpers_bolts.RealmHelperBolts;
  * Observes user is logged into server.
  */
 public class SessionObserver extends AbstractModelObserver<ServerConfig> {
+  private final MethodCallHelper methodCall;
   private int count;
 
+  /**
+   * constructor.
+   */
   public SessionObserver(Context context, String serverConfigId, RocketChatWebSocketAPI api) {
     super(context, serverConfigId, api);
+    methodCall = new MethodCallHelper(serverConfigId, api);
     count = 0;
   }
 
@@ -54,11 +59,10 @@ public class SessionObserver extends AbstractModelObserver<ServerConfig> {
   }
 
   @DebugLog private void onLogin() {
-    final MethodCallHelper methodCallHelper = new MethodCallHelper(serverConfigId);
     RealmHelperBolts.executeTransaction(realm -> {
       realm.delete(RoomSubscription.class);
       return null;
-    }).onSuccessTask(_task -> methodCallHelper.getRooms())
+    }).onSuccessTask(_task -> methodCall.getRooms())
         .continueWith(new LogcatIfError());
 
   }

app/src/main/java/chat/rocket/android/service/observer/TokenLoginObserver.java

@@ -10,8 +10,11 @@ import java.util.List;
 
 public class TokenLoginObserver extends AbstractModelObserver<ServerConfig> {
 
+  private final MethodCallHelper methodCall;
+
   public TokenLoginObserver(Context context, String serverConfigId, RocketChatWebSocketAPI api) {
     super(context, serverConfigId, api);
+    methodCall = new MethodCallHelper(serverConfigId, api);
   }
 
   @Override protected RealmResults<ServerConfig> queryItems(Realm realm) {
@@ -28,7 +31,7 @@ public class TokenLoginObserver extends AbstractModelObserver<ServerConfig> {
     }
 
     ServerConfig config = list.get(0);
-    new MethodCallHelper(serverConfigId, webSocketAPI).loginWithToken(config.getToken())
+    methodCall.loginWithToken(config.getToken())
         .continueWith(task -> {
           if (task.isFaulted()) {
             ServerConfig.logConnectionError(serverConfigId, task.getError());