checkpoint 2: Send/Receive meteor_login_service_configuration

app/build.gradle

@@ -48,7 +48,7 @@ dependencies {
     compile 'com.facebook.stetho:stetho-okhttp3:1.4.1'
     compile 'com.uphyca:stetho_realm:2.0.0'
 
-    compile 'chat.rocket:android-ddp:0.0.4'
+    compile 'chat.rocket:android-ddp:0.0.5'
 
     compile 'com.jakewharton.timber:timber:4.3.1'
     compile 'com.jakewharton.rxbinding:rxbinding:0.4.0'

app/src/main/java/chat/rocket/android/model/doc/MeteorLoginServiceConfiguration.java

+package chat.rocket.android.model.doc;
+
+import io.realm.RealmObject;
+import io.realm.annotations.PrimaryKey;
+
+public class MeteorLoginServiceConfiguration extends RealmObject {
+    @PrimaryKey
+    private String id;
+    private String service;
+    private String consumerKey; //for Twitter
+    private String appId;//for Facebook
+    private String clientId;//for other auth providers
+}

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

@@ -12,7 +12,9 @@ import bolts.Task;
 import bolts.TaskCompletionSource;
 import chat.rocket.android.helper.TextUtils;
 import chat.rocket.android.model.ServerConfig;
+import chat.rocket.android.service.ddp_subscription.LoginServiceConfigurationSubscriber;
 import chat.rocket.android.ws.RocketChatWebSocketAPI;
+import chat.rocket.android_ddp.DDPClient;
 import hugo.weaving.DebugLog;
 import jp.co.crowdworks.realm_java_helpers.RealmHelper;
 import timber.log.Timber;
@@ -98,13 +100,16 @@ public class RocketChatWebSocketThread extends HandlerThread {
     }
 
     private void scheduleUnregisterListeners() {
+        if (isAlive()) {
             new Handler(getLooper()).post(() -> {
                 Timber.d("thread %s: quit()", Thread.currentThread().getId());
                 unregisterListeners();
             });
         }
+    }
 
     private static final Class[] REGISTERABLE_CLASSES = {
+            LoginServiceConfigurationSubscriber.class
     };
 
     private final ArrayList<Registerable> mListeners = new ArrayList<>();
@@ -125,11 +130,19 @@ public class RocketChatWebSocketThread extends HandlerThread {
         return mWebSocketAPI.connect().onSuccess(task -> {
             registerListenersActually();
 
+            DDPClient client = task.getResult().client;
+
+            // handling WebSocket#onClose() callback.
+            client.getOnCloseCallback().onSuccess(_task -> {
+                quit();
+                return null;
+            });
 
             // just for debugging.
-            task.getResult().client.getSubscriptionCallback().subscribe(event -> {
+            client.getSubscriptionCallback().subscribe(event -> {
                 Timber.d(TAG, "Callback [DEBUG] < " + event);
             });
+
             return null;
         }).continueWith(task -> {
             if (task.isFaulted()) {

app/src/main/java/chat/rocket/android/service/ddp_subscription/AbstractDDPDocEventSubscriber.java

+package chat.rocket.android.service.ddp_subscription;
+
+import android.content.Context;
+import android.text.TextUtils;
+
+import org.json.JSONException;
+import org.json.JSONObject;
+
+import java.util.Iterator;
+
+import chat.rocket.android.helper.LogcatIfError;
+import chat.rocket.android.service.Registerable;
+import chat.rocket.android.ws.RocketChatWebSocketAPI;
+import chat.rocket.android_ddp.DDPSubscription;
+import io.realm.Realm;
+import io.realm.RealmObject;
+import jp.co.crowdworks.realm_java_helpers_bolts.RealmHelperBolts;
+import rx.Subscription;
+import timber.log.Timber;
+
+abstract class AbstractDDPDocEventSubscriber implements Registerable {
+    protected final Context mContext;
+    protected final RocketChatWebSocketAPI mAPI;
+    private String mID;
+    private Subscription mSubscription;
+
+    public AbstractDDPDocEventSubscriber(Context context, RocketChatWebSocketAPI api) {
+        mContext = context;
+        mAPI = api;
+    }
+
+    protected abstract String getSubscriptionName();
+    protected abstract String getSubscriptionCallbackName();
+    protected abstract Class<? extends RealmObject> getModelClass();
+    protected JSONObject customizeFieldJSON(JSONObject json) { return json; }
+
+    @Override
+    public void register() {
+        mAPI.subscribe(getSubscriptionName(), null).onSuccess(task -> {
+            mID = task.getResult().id;
+            return null;
+        }).continueWith(task -> {
+            if (task.isFaulted()) {
+                Timber.w(task.getError());
+            }
+            return null;
+        });
+
+        RealmHelperBolts.executeTransaction(realm -> {
+            realm.delete(getModelClass());
+            return null;
+        }).onSuccess(task -> {
+            registerSubscriptionCallback();
+            return null;
+        }).continueWith(new LogcatIfError());
+    }
+
+    private void registerSubscriptionCallback() {
+        mSubscription = mAPI.getSubscriptionCallback()
+                .filter(event -> event instanceof DDPSubscription.DocEvent
+                        && getSubscriptionCallbackName().equals(((DDPSubscription.DocEvent) event).collection))
+                .cast(DDPSubscription.DocEvent.class)
+                .subscribe(docEvent -> {
+                    try {
+                        if (docEvent instanceof DDPSubscription.Added.Before) {
+                            onDocumentAdded((DDPSubscription.Added) docEvent); //ignore Before
+                        } else if (docEvent instanceof DDPSubscription.Added) {
+                            onDocumentAdded((DDPSubscription.Added) docEvent);
+                        } else if (docEvent instanceof DDPSubscription.Removed) {
+                            onDocumentRemoved((DDPSubscription.Removed) docEvent);
+                        } else if (docEvent instanceof DDPSubscription.Changed) {
+                            onDocumentChanged((DDPSubscription.Changed) docEvent);
+                        } else if (docEvent instanceof DDPSubscription.MovedBefore) {
+                            //ignore movedBefore
+                        }
+                    } catch (Exception e) {
+                        Timber.w(e);
+                    }
+                });
+    }
+
+    protected void onDocumentAdded(DDPSubscription.Added docEvent) {
+        RealmHelperBolts.executeTransaction(realm -> {
+            onDocumentAdded(realm, docEvent);
+            return null;
+        }).continueWith(new LogcatIfError());
+    }
+
+    protected void onDocumentChanged(DDPSubscription.Changed docEvent) {
+        RealmHelperBolts.executeTransaction(realm -> {
+            onDocumentChanged(realm, docEvent);
+            return null;
+        }).continueWith(new LogcatIfError());
+    }
+
+    protected void onDocumentRemoved(DDPSubscription.Removed docEvent) {
+        RealmHelperBolts.executeTransaction(realm -> {
+            onDocumentRemoved(realm, docEvent);
+            return null;
+        }).continueWith(new LogcatIfError());
+    }
+
+    private void mergeJSON(JSONObject target, JSONObject src) throws JSONException {
+        Iterator<String> it = src.keys();
+        while(it.hasNext()) {
+            String key = it.next();
+            target.put(key, src.get(key));
+        }
+    }
+
+    private void onDocumentAdded(Realm realm, DDPSubscription.Added docEvent) throws JSONException {
+        //executed in RealmTransaction
+        JSONObject json = new JSONObject().put("id", docEvent.docID);
+        mergeJSON(json, docEvent.fields);
+        realm.createOrUpdateObjectFromJson(getModelClass(), customizeFieldJSON(json));
+    }
+
+    private void onDocumentChanged(Realm realm, DDPSubscription.Changed docEvent) throws JSONException {
+        //executed in RealmTransaction
+        JSONObject json = new JSONObject().put("id", docEvent.docID);
+        for (int i=0; i<docEvent.cleared.length(); i++) {
+            String fieldToDelete = docEvent.cleared.getString(i);
+            json.remove(fieldToDelete);
+        }
+        mergeJSON(json, docEvent.fields);
+        realm.createOrUpdateObjectFromJson(getModelClass(), customizeFieldJSON(json));
+    }
+
+    private void onDocumentRemoved(Realm realm, DDPSubscription.Removed docEvent) throws JSONException {
+        //executed in RealmTransaction
+        realm.where(getModelClass()).equalTo("id", docEvent.docID).findAll().deleteAllFromRealm();
+    }
+
+    @Override
+    public void keepalive() {
+
+    }
+
+    @Override
+    public void unregister() {
+        if (mSubscription != null) mSubscription.unsubscribe();
+        if (!TextUtils.isEmpty(mID)) {
+            mAPI.unsubscribe(mID).continueWith(new LogcatIfError());
+        }
+    }
+}

app/src/main/java/chat/rocket/android/service/ddp_subscription/LoginServiceConfigurationSubscriber.java

+package chat.rocket.android.service.ddp_subscription;
+
+import android.content.Context;
+
+import chat.rocket.android.model.doc.MeteorLoginServiceConfiguration;
+import chat.rocket.android.ws.RocketChatWebSocketAPI;
+import io.realm.RealmObject;
+
+public class LoginServiceConfigurationSubscriber extends AbstractDDPDocEventSubscriber {
+    public LoginServiceConfigurationSubscriber(Context context, RocketChatWebSocketAPI api) {
+        super(context, api);
+    }
+
+    @Override
+    protected String getSubscriptionName() {
+        return "meteor.loginServiceConfiguration";
+    }
+
+    @Override
+    protected String getSubscriptionCallbackName() {
+        return "meteor_accounts_loginServiceConfiguration";
+    }
+
+    @Override
+    protected Class<? extends RealmObject> getModelClass() {
+        return MeteorLoginServiceConfiguration.class;
+    }
+}

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

+package chat.rocket.android.service.observer;
+
+import android.content.Context;
+
+import chat.rocket.android.service.Registerable;
+import chat.rocket.android.ws.RocketChatWebSocketAPI;
+import io.realm.RealmObject;
+import jp.co.crowdworks.realm_java_helpers.RealmListObserver;
+
+abstract class AbstractModelObserver<T extends RealmObject> extends RealmListObserver<T> implements Registerable {
+
+    protected final Context mContext;
+    protected final RocketChatWebSocketAPI mAPI;
+
+    public AbstractModelObserver(Context context, RocketChatWebSocketAPI api) {
+        mContext = context;
+        mAPI = api;
+    }
+
+
+    @Override
+    public void register() {
+        sub();
+    }
+
+    @Override
+    public void unregister() {
+        unsub();
+    }
+}

app/src/main/java/chat/rocket/android/ws/RocketChatWebSocketAPI.java

 package chat.rocket.android.ws;
 
+import org.json.JSONArray;
+
+import java.util.UUID;
+
 import bolts.Task;
 import chat.rocket.android.helper.OkHttpHelper;
 import chat.rocket.android_ddp.DDPClient;
 import chat.rocket.android_ddp.DDPClientCallback;
+import chat.rocket.android_ddp.DDPSubscription;
+import rx.Observable;
 
 public class RocketChatWebSocketAPI {
     private final DDPClient mDDPClient;
@@ -29,4 +35,18 @@ public class RocketChatWebSocketAPI {
     public void close() {
         mDDPClient.close();
     }
+
+
+    public Task<DDPSubscription.Ready> subscribe(final String name, JSONArray param) {
+        return mDDPClient.sub(UUID.randomUUID().toString(), name, param);
+    }
+
+    public Task<DDPSubscription.NoSub> unsubscribe(final String id) {
+        return mDDPClient.unsub(id);
+    }
+
+    public Observable<DDPSubscription.Event> getSubscriptionCallback() {
+        return mDDPClient.getSubscriptionCallback();
+    }
+
 }