Support for Connection notifications. (#1950)

CHANGELOG.md

@@ -1,3 +1,27 @@
+2.15.0 Release notes (YYYY-MM-DD)
+=============================================================
+### Compatibility
+* Sync protocol: 24
+* Server-side history format: 4
+* File format: 7
+* Realm Object Server: 3.0.0 or later
+
+### Breaking changes
+* None.
+
+### Enhancements
+* [Sync] Added `Realm.Sync.ConnectionState` representing the states a connection to the server can be in.
+* [Sync] Added support for `Session.addConnectionNotification()` and `Session.removeConnectionNotification`.
+* [Sync] Added `Session.connectionState`.
+* [Sync] Added `Session.isConnected()`.
+
+### Bug fixes
+* None
+
+### Internals
+* Updated to Object Store commit: 97fd03819f398b3c81c8b007feaca8636629050b
+
+
 2.14.2 Release notes (2018-8-8)
 =============================================================
 ### Compatibility

docs/sync.js

@@ -595,6 +595,44 @@ class Session {
     * @param {callback(transferred, transferable)} callback - a previously registered progress callback
     */
     removeProgressNotification(progressCallback) {}
+
+    /**
+     *
+     * @param connectionCallback
+     */
+    addConnectionNotification(connectionCallback) {}
+
+    /**
+     * Unregister a state notification callback that was previously registered with addStateNotification.
+     * Calling the function multiple times with the same callback is ignored.
+     *
+     * @param {callback(oldState, newState)} callback - a previously registered state callback.
+     */
+    removeConnectionNotification(connectionCallback) {}
+
+    /**
+     * Gets the current state of the connection to the server. Multiple sessions might share the same underlying
+     * connection. In that case, any connection change is sent to all sessions.
+     *
+     * Can be either:
+     *  - Realm.Sync.ConnectionState.Disconnected: No connection to the server is available.
+     *  - Realm.Sync.ConnectionState.Connecting: An attempt to connect to the server is in progress.
+     *  - Realm.Sync.ConnectionState.Connected: The connection to the server is active and data can be synchronized.
+     *
+     * Data will only be synchronized with the Realm ObjectServer if this method returns `Connected` and `state()`
+     * returns `Active` or `Dying`.
+     *
+     * @type {string}
+     */
+    connectionState() {}
+
+    /**
+     * Returns `true` if the session is currently active and connected to the server, `false` if not.
+     *
+     * @type {boolean}
+     */
+    isConnected() {}
+
 }
 
 /**

lib/browser/session.js

@@ -35,7 +35,10 @@ createMethods(Session.prototype, objectTypes.SESSION, [
     '_refreshAccessToken',
     '_simulateError',
     'addProgressNotification',
-    'removeProgressNotification'
+    'removeProgressNotification',
+    'addConnectionNotification',
+    'removeConnectionNotification',
+    'isConnected',
 ]);
 
 export function createSession(realmId, info) {

lib/extensions.js

@@ -242,6 +242,12 @@ module.exports = function(realmConstructor) {
             Invalidated: 3, // The subscription has been removed.
         };
 
+        realmConstructor.Sync.ConnectionState = {
+            Disconnected: "disconnected",
+            Connecting: "connecting",
+            Connected: "connected",
+        }
+
         // Define the permission schemas as constructors so that they can be
         // passed into directly to functions which want object type names
         const permissionsSchema = Object.freeze({

lib/index.d.ts

@@ -430,10 +430,18 @@ declare namespace Realm.Sync {
         custom_http_headers?: { [header: string]: string };
     }
 
+    enum ConnectionState {
+        Disconnected = "disconnected",
+        Connecting = "connecting",
+        Connected = "connected",
+    }
+
     type ProgressNotificationCallback = (transferred: number, transferable: number) => void;
     type ProgressDirection = 'download' | 'upload';
     type ProgressMode = 'reportIndefinitely' | 'forCurrentlyOutstandingWork';
 
+    type ConnectionNotificationCallback = (oldState: ConnectionState, newState: ConnectionState) => void;
+
     /**
     * Session
     * @see { @link https://realm.io/docs/javascript/latest/api/Realm.Sync.Session.html }
@@ -443,9 +451,15 @@ declare namespace Realm.Sync {
         readonly state: 'invalid' | 'active' | 'inactive';
         readonly url: string;
         readonly user: User;
+        readonly connectionState: ConnectionState;
 
         addProgressNotification(direction: ProgressDirection, mode: ProgressMode, progressCallback: ProgressNotificationCallback): void;
         removeProgressNotification(progressCallback: ProgressNotificationCallback): void;
+
+        addConnectionNotification(callback): void;
+        removeConnectionNotification(callback): void;
+
+        isConnected(): boolean;
     }
 
     type SubscriptionNotificationCallback = (subscription: Subscription, state: number) => void;

src/js_sync.hpp

@@ -213,6 +213,8 @@ class SessionClass : public ClassDefinition<T, WeakSession> {
 public:
     std::string const name = "Session";
     using ProgressHandler = void(uint64_t transferred_bytes, uint64_t transferrable_bytes);
+    using StateHandler = void(SyncSession::PublicState old_state, SyncSession::PublicState new_state);
+    using ConnectionHandler = void(SyncSession::ConnectionState old_state, SyncSession::ConnectionState new_state);
 
     static FunctionType create_constructor(ContextType);
 
@@ -220,11 +222,18 @@ public:
     static void get_user(ContextType, ObjectType, ReturnValue &);
     static void get_url(ContextType, ObjectType, ReturnValue &);
     static void get_state(ContextType, ObjectType, ReturnValue &);
+    static void get_connection_state(ContextType, ObjectType, ReturnValue &);
 
     static void simulate_error(ContextType, FunctionType, ObjectType, size_t, const ValueType[], ReturnValue &);
     static void refresh_access_token(ContextType, FunctionType, ObjectType, size_t, const ValueType[], ReturnValue &);
     static void add_progress_notification(ContextType ctx, FunctionType, ObjectType this_object, size_t argc, const ValueType arguments[], ReturnValue &);
     static void remove_progress_notification(ContextType ctx, FunctionType, ObjectType this_object, size_t argc, const ValueType arguments[], ReturnValue &);
+    static void add_state_notification(ContextType ctx, FunctionType, ObjectType this_object, size_t argc, const ValueType arguments[], ReturnValue &);
+    static void remove_state_notification(ContextType ctx, FunctionType, ObjectType this_object, size_t argc, const ValueType arguments[], ReturnValue &);
+    static void add_connection_notification(ContextType ctx, FunctionType, ObjectType this_object, size_t argc, const ValueType arguments[], ReturnValue &);
+    static void remove_connection_notification(ContextType ctx, FunctionType, ObjectType this_object, size_t argc, const ValueType arguments[], ReturnValue &);
+    static void is_connected(ContextType ctx, FunctionType, ObjectType this_object, size_t argc, const ValueType arguments[], ReturnValue &);
+
 
     static void override_server(ContextType ctx, ObjectType this_object, Arguments args, ReturnValue&);
 
@@ -232,7 +241,8 @@ public:
         {"config", {wrap<get_config>, nullptr}},
         {"user", {wrap<get_user>, nullptr}},
         {"url", {wrap<get_url>, nullptr}},
-        {"state", {wrap<get_state>, nullptr}}
+        {"state", {wrap<get_state>, nullptr}},
+        {"connectionState", {wrap<get_connection_state>, nullptr}},
     };
 
     MethodMap<T> const methods = {
@@ -241,7 +251,13 @@ public:
         {"_overrideServer", wrap<override_server>},
         {"addProgressNotification", wrap<add_progress_notification>},
         {"removeProgressNotification", wrap<remove_progress_notification>},
+        {"addConnectionNotification", wrap<add_connection_notification>},
+        {"removeConnectionNotification", wrap<remove_connection_notification>},
+        {"isConnected", wrap<is_connected>},
     };
+
+private:
+    static std::string get_connection_state_value(SyncSession::ConnectionState state);
 };
 
 template<typename T>
@@ -448,12 +464,29 @@ void SessionClass<T>::get_state(ContextType ctx, ObjectType object, ReturnValue
     if (auto session = get_internal<T, SessionClass<T>>(object)->lock()) {
         if (session->state() == SyncSession::PublicState::Inactive) {
             return_value.set(inactive);
-        } else if (session->state() != SyncSession::PublicState::Error) {
+        } else {
             return_value.set(active);
         }
     }
 }
 
+template<typename T>
+std::string SessionClass<T>::get_connection_state_value(SyncSession::ConnectionState state) {
+    switch(state) {
+        case SyncSession::ConnectionState::Disconnected: return "disconnected";
+        case SyncSession::ConnectionState::Connecting: return "connecting";
+        case SyncSession::ConnectionState::Connected: return "connected";
+    }
+}
+
+template<typename T>
+void SessionClass<T>::get_connection_state(ContextType ctx, ObjectType object, ReturnValue &return_value) {
+    return_value.set(get_connection_state_value(SyncSession::ConnectionState::Disconnected));
+    if (auto session = get_internal<T, SessionClass<T>>(object)->lock()) {
+        return_value.set(get_connection_state_value(session->connection_state()));
+    }
+}
+
 template<typename T>
 void SessionClass<T>::simulate_error(ContextType ctx, FunctionType, ObjectType this_object, size_t argc, const ValueType arguments[], ReturnValue &) {
     validate_argument_count(argc, 2);
@@ -553,6 +586,66 @@ void SessionClass<T>::remove_progress_notification(ContextType ctx, FunctionType
     }
 }
 
+template<typename T>
+void SessionClass<T>::add_connection_notification(ContextType ctx, FunctionType, ObjectType this_object, size_t argc, const ValueType arguments[], ReturnValue &return_value) {
+    validate_argument_count(argc, 1);
+    if (auto session = get_internal<T, SessionClass<T>>(this_object)->lock()) {
+        auto callback_function = Value::validated_to_function(ctx, arguments[0], "callback");
+        Protected<FunctionType> protected_callback(ctx, callback_function);
+        Protected<ObjectType> protected_this(ctx, this_object);
+        Protected<typename T::GlobalContext> protected_ctx(Context<T>::get_global_context(ctx));
+
+        std::function<ConnectionHandler> connectionFunc;
+
+        EventLoopDispatcher<ConnectionHandler> connection_handler([=](SyncSession::ConnectionState old_state, SyncSession::ConnectionState new_state) {
+            HANDLESCOPE
+            ValueType callback_arguments[2];
+            callback_arguments[0] = Value::from_string(protected_ctx, get_connection_state_value(old_state));
+            callback_arguments[1] = Value::from_string(protected_ctx, get_connection_state_value(new_state));
+            Function<T>::callback(protected_ctx, protected_callback, typename T::Object(), 2, callback_arguments);
+        });
+
+        connectionFunc = std::move(connection_handler);
+
+        auto notificationToken = session->register_connection_change_callback(std::move(connectionFunc));
+        auto syncSession = create_object<T, SessionClass<T>>(ctx, new WeakSession(session));
+        PropertyAttributes attributes = ReadOnly | DontEnum | DontDelete;
+        Object::set_property(ctx, callback_function, "_syncSession", syncSession, attributes);
+        Object::set_property(ctx, callback_function, "_connectionNotificationToken", Value::from_number(protected_ctx, notificationToken), attributes);
+    }
+}
+
+template<typename T>
+void SessionClass<T>::remove_connection_notification(ContextType ctx, FunctionType, ObjectType this_object, size_t argc, const ValueType arguments[], ReturnValue &return_value) {
+    validate_argument_count(argc, 1);
+    auto callback_function = Value::validated_to_function(ctx, arguments[0], "callback");
+    auto syncSessionProp = Object::get_property(ctx, callback_function, "_syncSession");
+    if (Value::is_undefined(ctx, syncSessionProp) || Value::is_null(ctx, syncSessionProp)) {
+        return;
+    }
+    auto syncSession = Value::validated_to_object(ctx, syncSessionProp);
+    auto registrationToken = Object::get_property(ctx, callback_function, "_connectionNotificationToken");
+
+    if (auto session = get_internal<T, SessionClass<T>>(syncSession)->lock()) {
+        auto reg = Value::validated_to_number(ctx, registrationToken);
+        session->unregister_connection_change_callback(reg);
+    }
+}
+
+template<typename T>
+void SessionClass<T>::is_connected(ContextType ctx, FunctionType, ObjectType this_object, size_t argc, const ValueType arguments[], ReturnValue &return_value) {
+    validate_argument_count(argc, 0);
+    return_value.set(false);
+    if (auto session = get_internal<T, SessionClass<T>>(this_object)->lock()) {
+        auto state = session->state();
+        auto connection_state = session->connection_state();
+        if (connection_state == SyncSession::ConnectionState::Connected
+            && (state == SyncSession::PublicState::Active || state == SyncSession::PublicState::Dying)) {
+            return_value.set(true);
+        }
+    }
+}
+
 template<typename T>
 void SessionClass<T>::override_server(ContextType ctx, ObjectType this_object, Arguments args, ReturnValue&) {
     args.validate_count(2);

src/object-store

@@ -1 +1 @@
-Subproject commit bec15175287165322088b682032d14e6dbaa813c
+Subproject commit 97fd03819f398b3c81c8b007feaca8636629050b

tests/js/session-tests.js

@@ -927,5 +927,113 @@ module.exports = {
                 session._simulateError(211, 'ClientReset'); // 211 -> divering histories
             });
         });
-    }
-}
+    },
+
+    testAddConnectionNotification() {
+        if (!isNodeProccess) {
+            return;
+        }
+
+        return Realm.Sync.User.register('http://localhost:9080', uuid(), 'password').then((u) => {
+            return new Promise((resolve, reject) => {
+                let config = {
+                    sync: {
+                        user: u,
+                        url: `realm://localhost:9080/~/${uuid()}`,
+                        fullSynchronization: true,
+                    }
+                };
+                
+                Realm.open(config).then(realm => {
+                    realm.syncSession.addConnectionNotification((oldState, newState) => {
+                        if (oldState === Realm.Sync.ConnectionState.Connected && newState === Realm.Sync.ConnectionState.Disconnected) {
+                            resolve('Done');
+                        }
+                    });
+                    realm.close()
+                }).catch(error => reject(error));
+            });
+        });
+    },
+
+    testRemoveConnectionNotification() {
+        if (!isNodeProccess) {
+            return;
+        }
+
+        return Realm.Sync.User.register('http://localhost:9080', uuid(), 'password').then((u) => {
+            return new Promise((resolve, reject) => {
+                let config = {
+                    sync: {
+                        user: u,
+                        url: `realm://localhost:9080/~/${uuid()}`,
+                        fullSynchronization: true,
+                    }
+                };
+                
+                Realm.open(config).then(realm => {
+                    let callback1 = (oldState, newState) => {
+                        reject("Should not be called");
+                    };
+                    let callback2 = (oldState, newState) => {
+                        if (oldState === Realm.Sync.ConnectionState.Connected && newState === Realm.Sync.ConnectionState.Disconnected) {
+                            resolve('Done');
+                        }
+                    };
+                    let session = realm.syncSession;
+                    session.addConnectionNotification(callback1);
+                    session.addConnectionNotification(callback2);
+                    session.removeConnectionNotification(callback1);
+                    realm.close()
+                }).catch(error => reject(error));
+            });
+        });
+    },
+
+    testConnectionState() {
+        if (!isNodeProccess) {
+            return;
+        }
+
+        return Realm.Sync.User.register('http://localhost:9080', uuid(), 'password').then((u) => {
+            return new Promise((resolve, reject) => {
+                let config = {
+                    sync: {
+                        user: u,
+                        url: `realm://localhost:9080/~/${uuid()}`,
+                        fullSynchronization: true,
+                    }
+                };
+
+                Realm.open(config).then(realm => {
+                    let session = realm.syncSession;
+                    TestCase.assertEqual(session.connectionState, Realm.Sync.ConnectionState.Disconnected);
+                    TestCase.assertFalse(session.isConnected());
+                    session.addConnectionNotification((oldState, newState) => {
+                        switch (newState) {
+                            case Realm.Sync.ConnectionState.Disconnected:
+                                TestCase.assertEqual(session.connectionState, Realm.Sync.ConnectionState.Disconnected);
+                                TestCase.assertFalse(session.isConnected());
+                                break;
+                            case Realm.Sync.ConnectionState.Connecting:
+                                TestCase.assertEqual(session.connectionState, Realm.Sync.ConnectionState.Connecting);
+                                TestCase.assertFalse(session.isConnected());
+                                break;
+                            case Realm.Sync.ConnectionState.Connected:
+                                TestCase.assertEqual(session.connectionState, Realm.Sync.ConnectionState.Connected);
+                                TestCase.assertTrue(session.isConnected());
+                                break;
+                            default:
+                                reject(`unknown connection value: ${newState}`);
+                        }
+
+                        if (oldState === Realm.Sync.ConnectionState.Connecting && newState === Realm.Sync.ConnectionState.Connected) {
+                            resolve('Done');
+                        }
+                    });
+                }).catch(error => reject(error));
+            });
+        });
+    },
+
+};