add progress notifications api

lib/extensions.js

@@ -43,26 +43,53 @@ module.exports = function(realmConstructor) {
     //Add async open API
     Object.defineProperties(realmConstructor, getOwnPropertyDescriptors({
         open(config) {
-            return new Promise((resolve, reject) => {
+            let syncSession;
-                realmConstructor._waitForDownload(config, (error) => {
+            let promise = new Promise((resolve, reject) => {
-                    if (error) {
+                realmConstructor._waitForDownload(config,
-                        reject(error);
+                    (session) => {
-                    }
+                        syncSession = session;
-                    else {
+                    },
-                        try {
+                    (error) => {
-                            let syncedRealm = new this(config);
+                        if (error) {
-                            //FIXME: RN hangs here. Remove when node's makeCallback alternative is implemented
+                            reject(error);
-                            setTimeout(() => { resolve(syncedRealm); }, 1);
-                        } catch (e) {
-                            reject(e);
                         }
-                    }
+                        else {
-                });
+                            try {
+                                let syncedRealm = new this(config);
+                                //FIXME: RN hangs here. Remove when node's makeCallback alternative is implemented
+                                setTimeout(() => { resolve(syncedRealm); }, 1);
+                            } catch (e) {
+                                reject(e);
+                            }
+                        }
+                    });
             });
+
+            promise.progress = (callback) => {
+                if (syncSession) {
+                    syncSession.addProgressNotification('download', 'forCurrentlyOutstandingWork', callback);
+                }
+
+                return promise;
+            };
+
+            return promise;
         },
 
-        openAsync(config, callback) {
+        openAsync(config, progressCallback, callback) {
-                realmConstructor._waitForDownload(config, (error) => {
+
+            if (!callback) {
+                callback = progressCallback;
+                progressCallback = null;
+            }
+
+            realmConstructor._waitForDownload(config,
+                (syncSession) => {
+                    if (progressCallback) {
+                        syncSession.addProgressNotification('download', 'forCurrentlyOutstandingWork', progressCallback);;
+                    }
+                },
+                (error) => {
                     if (error) {
                         callback(error);
                     }

src/js_realm.hpp

@@ -619,9 +619,14 @@ void RealmClass<T>::get_sync_session(ContextType ctx, ObjectType object, ReturnV
 
 template<typename T>
 void RealmClass<T>::wait_for_download_completion(ContextType ctx, FunctionType, ObjectType this_object, size_t argc, const ValueType arguments[], ReturnValue &return_value) {
-    validate_argument_count(argc, 2);
+    validate_argument_count(argc, 2, 3);
     auto config_object = Value::validated_to_object(ctx, arguments[0]);
-    auto callback_function = Value::validated_to_function(ctx, arguments[1]);
+    auto callback_function = Value::validated_to_function(ctx, arguments[argc - 1]);
+
+    FunctionType session_callback;
+    if (argc == 3) {
+        session_callback = Value::validated_to_function(ctx, arguments[1]);
+    }
 
 #if REALM_ENABLE_SYNC
     ValueType sync_config_value = Object::get_property(ctx, config_object, "sync");
@@ -694,6 +699,13 @@ void RealmClass<T>::wait_for_download_completion(ContextType ctx, FunctionType,
             std::shared_ptr<SyncUser> user = sync_config->user;
             if (user && user->state() != SyncUser::State::Error) {
                 if (auto session = user->session_for_on_disk_path(config.path)) {
+                    if (Value::is_valid(session_callback)) {
+                        auto syncSession = create_object<T, SessionClass<T>>(ctx, new WeakSession(session));
+                        ValueType callback_arguments[1];
+                        callback_arguments[0] = syncSession;
+                        Function<T>::callback(protected_ctx, session_callback, protected_this, 1, callback_arguments);
+                    }
+
                     if (progressFuncDefined) {
                         session->register_progress_notifier(std::move(progressFunc), SyncSession::NotifierType::download, false);
                     } 

src/js_sync.hpp

@@ -147,6 +147,31 @@ void UserClass<T>::logout(ContextType ctx, FunctionType, ObjectType this_object,
     get_internal<T, UserClass<T>>(this_object)->get()->log_out();
 }
 
+template<typename T>
+class ProgressNotificationTokenClass : public ClassDefinition<T, uint64_t> {
+    using GlobalContextType = typename T::GlobalContext;
+    using ContextType = typename T::Context;
+    using FunctionType = typename T::Function;
+    using ObjectType = typename T::Object;
+    using ValueType = typename T::Value;
+    using String = js::String<T>;
+    using Object = js::Object<T>;
+    using Value = js::Value<T>;
+    using Function = js::Function<T>;
+    using ReturnValue = js::ReturnValue<T>;
+
+public:
+    std::string const name = "ProgressNotificationToken";
+    
+    static ObjectType create_instance(ContextType, uint64_t);
+
+    static void stop(ContextType, FunctionType, ObjectType, size_t, const ValueType[], ReturnValue &);
+
+    MethodMap<T> const methods = {
+        {"stop", wrap<stop>},
+    };
+};
+
 template<typename T>
 class SessionClass : public ClassDefinition<T, WeakSession> {
     using ContextType = typename T::Context;
@@ -160,6 +185,7 @@ class SessionClass : public ClassDefinition<T, WeakSession> {
 
 public:
     std::string const name = "Session";
+    using ProgressHandler = void(uint64_t transferred_bytes, uint64_t transferrable_bytes);
 
     static FunctionType create_constructor(ContextType);
 
@@ -170,6 +196,10 @@ public:
 
     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 &);
+#if REALM_ENABLE_SYNC
+    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 &);
+#endif
 
     PropertyMap<T> const properties = {
         {"config", {wrap<get_config>, nullptr}},
@@ -180,7 +210,11 @@ public:
 
     MethodMap<T> const methods = {
         {"_simulateError", wrap<simulate_error>},
-        {"_refreshAccessToken", wrap<refresh_access_token>}
+        {"_refreshAccessToken", wrap<refresh_access_token>},
+#if REALM_ENABLE_SYNC
+        {"addProgressNotification", wrap<add_progress_notification>},
+        {"removeProgressNotification", wrap<remove_progress_notification>},
+#endif
     };
 };
 
@@ -307,8 +341,82 @@ void SessionClass<T>::refresh_access_token(ContextType ctx, FunctionType, Object
     }
 }
 
+#if REALM_ENABLE_SYNC
 template<typename T>
-class SyncClass : public ClassDefinition<T, void *> {
+void SessionClass<T>::add_progress_notification(ContextType ctx, FunctionType, ObjectType this_object, size_t argc, const ValueType arguments[], ReturnValue &return_value) {
+    validate_argument_count(argc, 3);
+
+    if (auto session = get_internal<T, SessionClass<T>>(this_object)->lock()) {
+        
+        std::string direction = Value::validated_to_string(ctx, arguments[0], "direction");
+        std::string mode = Value::validated_to_string(ctx, arguments[1], "mode");
+        SyncSession::NotifierType notifierType;
+        if (direction == "download") {
+            notifierType = SyncSession::NotifierType::download;
+        }
+        else if (direction == "upload") {
+            notifierType = SyncSession::NotifierType::upload;
+        }
+        else {
+            throw std::invalid_argument("Invalid argument 'direction'. Only 'download' and 'upload' progress notification directions are supported");
+        }
+
+        bool is_streaming = false;
+        if (mode == "reportIndefinitely") {
+            is_streaming = true;
+        }
+        else if (mode == "forCurrentlyOutstandingWork") {
+            is_streaming = false;
+        }
+        else {
+            throw std::invalid_argument("Invalid argument 'mode'. Only 'reportIndefinitely' and 'forCurrentlyOutstandingWork' progress notification modes are supported");
+        }
+
+        auto callback_function = Value::validated_to_function(ctx, arguments[2], "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<ProgressHandler> progressFunc; 
+
+        EventLoopDispatcher<ProgressHandler> progress_handler([=](uint64_t transferred_bytes, uint64_t transferrable_bytes) {
+            HANDLESCOPE
+            ValueType callback_arguments[2];
+            callback_arguments[0] = Value::from_number(protected_ctx, transferred_bytes);
+            callback_arguments[1] = Value::from_number(protected_ctx, transferrable_bytes);
+
+            Function<T>::callback(protected_ctx, protected_callback, protected_this, 2, callback_arguments);
+        });
+
+        progressFunc = std::move(progress_handler);
+
+        
+        auto registrationToken = session->register_progress_notifier(std::move(progressFunc), notifierType, false);
+
+        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, "registrationToken", Value::from_number(protected_ctx, registrationToken), attributes);
+    }
+}
+
+template<typename T>
+void SessionClass<T>::remove_progress_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");
+    auto syncSession = Value::validated_to_object(ctx, syncSessionProp);
+    auto registrationToken = Object::get_property(ctx, callback_function, "registrationToken");
+
+    if (auto session = get_internal<T, SessionClass<T>>(syncSession)->lock()) {
+        auto reg = Value::validated_to_number(ctx, registrationToken);
+        session->unregister_progress_notifier(reg);
+    }
+}
+#endif
+
+template<typename T>
+class SyncClass : public ClassDefinition<T, void*> {
     using GlobalContextType = typename T::GlobalContext;
     using ContextType = typename T::Context;
     using FunctionType = typename T::Function;
@@ -447,5 +555,13 @@ void SyncClass<T>::populate_sync_config(ContextType ctx, ObjectType realm_constr
     }
 }
 
+template<typename T>
+void ProgressNotificationTokenClass<T>::stop(ContextType ctx, FunctionType, ObjectType this_object, size_t argc, const ValueType arguments[], ReturnValue &return_value) {
+}
+
+template<typename T>
+typename T::Object ProgressNotificationTokenClass<T>::create_instance(ContextType ctx, uint64_t progressToken) {
+    return create_object<T, ProgressNotificationTokenClass<T>>(ctx, new uint64_t(progressToken));
+}
 } // js
 } // realm

tests/js/session-tests.js

@@ -26,7 +26,7 @@ const Realm = require('realm');
 const TestCase = require('./asserts');
 
 const isNodeProccess = (typeof process === 'object' && process + '' === '[object process]');
-console.log("isnode " + isNodeProccess + " typeof " + (typeof(process) === 'object'));
+
 function node_require(module) {
     return require(module);
 }  
@@ -84,13 +84,23 @@ function runOutOfProcess(nodeJsFilePath) {
     fs.appendFileSync(tmpFile.fd, content, { encoding: 'utf8' });
     nodeArgs[0] = tmpFile.name;
     return new Promise((resolve, reject) => {
-        const child = execFile('node', nodeArgs, { cwd: tmpDir.name }, (error, stdout, stderr) => {
+        try {
-            if (error) {
+            console.log('runOutOfProcess command\n node ' + nodeArgs.join(" "));
-                reject(new Error(`Error executing ${nodeJsFilePath} Error: ${error}`));
+            const child = execFile('node', nodeArgs, { cwd: tmpDir.name }, (error, stdout, stderr) => {
+                if (error) {
+                    console.error("runOutOfProcess failed\n" + error);
+                    reject(new Error(`Running ${nodeJsFilePath} failed. error: ${error}`));
+                    return;
+                }
+
+                console.log('runOutOfProcess success\n' + stdout);
+                resolve();
+            });
             }
-            resolve();
+        catch (e) {
-        });
+            reject(e);
-    })
+        };
+    });
 }
 
 module.exports = {
@@ -150,7 +160,7 @@ module.exports = {
         const realmName = uuid();
         const expectedObjectsCount = 3;
 
-        runOutOfProcess(__dirname + '/download-api-helper.js', username, realmName, REALM_MODULE_PATH)
+        return runOutOfProcess(__dirname + '/download-api-helper.js', username, realmName, REALM_MODULE_PATH)
             .then(() => {
                 return promisifiedLogin('http://localhost:9080', username, 'password').then(user => {
                     const accessTokenRefreshed = this;
@@ -186,7 +196,7 @@ module.exports = {
         const realmName = uuid();
         const expectedObjectsCount = 3;
 
-        runOutOfProcess(__dirname + '/download-api-helper.js', username, realmName, REALM_MODULE_PATH)
+        return runOutOfProcess(__dirname + '/download-api-helper.js', username, realmName, REALM_MODULE_PATH)
             .then(() => {
                 return promisifiedLogin('http://localhost:9080', username, 'password').then(user => {
                     return new Promise((resolve, reject) => {
@@ -239,7 +249,7 @@ module.exports = {
         const realmName = uuid();
         const expectedObjectsCount = 3;
 
-        runOutOfProcess(__dirname + '/download-api-helper.js', username, realmName, REALM_MODULE_PATH)
+        return runOutOfProcess(__dirname + '/download-api-helper.js', username, realmName, REALM_MODULE_PATH)
             .then(() => {
                 return promisifiedLogin('http://localhost:9080', username, 'password').then(user => {
                     const accessTokenRefreshed = this;
@@ -275,7 +285,7 @@ module.exports = {
         const realmName = uuid();
         const expectedObjectsCount = 3;
 
-        runOutOfProcess(__dirname + '/download-api-helper.js', username, realmName, REALM_MODULE_PATH)
+        return runOutOfProcess(__dirname + '/download-api-helper.js', username, realmName, REALM_MODULE_PATH)
             .then(() => {
                 return promisifiedLogin('http://localhost:9080', username, 'password').then(user => {
                     return new Promise((resolve, reject) => {
@@ -322,7 +332,7 @@ module.exports = {
         const realmName = uuid();
         const expectedObjectsCount = 3;
 
-       runOutOfProcess(__dirname + '/download-api-helper.js', username, realmName, REALM_MODULE_PATH)
+       return runOutOfProcess(__dirname + '/download-api-helper.js', username, realmName, REALM_MODULE_PATH)
             .then(() => {
                 return promisifiedLogin('http://localhost:9080', username, 'password').then(user => {
                     return new Promise((resolve, reject) => {
@@ -449,5 +459,191 @@ module.exports = {
                 session._simulateError(123, 'simulated error');
             });
         });
-    }
+    },
+
+
+    testProgressNotificationsForRealmConstructor() {
+        if (!isNodeProccess) {
+            return Promise.resolve();
+        }
+
+        const username = uuid();
+        const realmName = uuid();
+
+        return runOutOfProcess(__dirname + '/download-api-helper.js', username, realmName, REALM_MODULE_PATH)
+            .then(() => {
+                return promisifiedLogin('http://localhost:9080', username, 'password').then(user => {
+                    return new Promise((resolve, reject) => {
+                        let config = {
+                            sync: {
+                                user,
+                                url: `realm://localhost:9080/~/${realmName}`
+                            },
+                            schema: [{ name: 'Dog', properties: { name: 'string' } }],
+                        };
+
+                        let realm = new Realm(config);
+                        const progressCallback = (transferred, total) => {
+                            resolve();
+                        };
+                        
+                        realm.syncSession.addProgressNotification('download', 'reportIndefinitely', progressCallback);
+
+                        setTimeout(function() {
+                            reject("Progress Notifications API failed to call progress callback for Realm constructor");
+                        }, 5000);
+                    });
+                });
+            });
+    },  
+
+    testProgressNotificationsUnregisterForRealmConstructor() {
+        if (!isNodeProccess) {
+            return Promise.resolve();
+        }
+
+        const username = uuid();
+        const realmName = uuid();
+
+        return runOutOfProcess(__dirname + '/download-api-helper.js', username, realmName, REALM_MODULE_PATH)
+            .then(() => {
+                return promisifiedLogin('http://localhost:9080', username, 'password').then(user => {
+                    return new Promise((resolve, reject) => {
+                        let config = {
+                            sync: {
+                                user,
+                                url: `realm://localhost:9080/~/${realmName}`
+                            },
+                            schema: [{ name: 'Dog', properties: { name: 'string' } }],
+                        };
+
+                        let realm = new Realm(config);
+                        let unregisterFunc;
+                       
+                        let writeDataFunc = () => {
+                            realm.write(() => {
+                                for (let i = 1; i <= 3; i++) {
+                                    realm.create('Dog', { name: `Lassy ${i}` });
+                                }
+                            });
+                        }
+
+                        let syncFinished = false;
+                        let failOnCall = false;
+                        const progressCallback = (transferred, total) => {
+                            if (failOnCall) {
+                                reject(new Error("Progress callback should not be called after removeProgressNotification"));
+                            }
+                            
+                            syncFinished = transferred === total;
+                            
+                            //unregister and write some new data.
+                            if (syncFinished) { 
+                                failOnCall = true;
+                                unregisterFunc();
+                                
+                                //use second callback to wait for sync finished
+                                realm.syncSession.addProgressNotification('upload', 'reportIndefinitely', (x, y) => {
+                                    if (x === y) {
+                                        resolve();
+                                    }
+                                });
+                                writeDataFunc();
+                            }
+                        };
+                        
+                        realm.syncSession.addProgressNotification('upload', 'reportIndefinitely', progressCallback);
+                        
+                        unregisterFunc = () => {
+                            realm.syncSession.removeProgressNotification(progressCallback);
+                        };
+
+                        writeDataFunc();
+                    });
+                });
+            });
+    },
+
+    testProgressNotificationsForRealmOpen() {
+        if (!isNodeProccess) {
+            return Promise.resolve();
+        }
+
+        const username = uuid();
+        const realmName = uuid();
+
+        return runOutOfProcess(__dirname + '/download-api-helper.js', username, realmName, REALM_MODULE_PATH)
+            .then(() => {
+                return promisifiedLogin('http://localhost:9080', username, 'password').then(user => {
+                    return new Promise((resolve, reject) => {
+                        let config = {
+                            sync: {
+                                user,
+                                url: `realm://localhost:9080/~/${realmName}`
+                            },
+                            schema: [{ name: 'Dog', properties: { name: 'string' } }],
+                        };
+
+                        let progressCalled = false;
+                        Realm.open(config)
+                            .progress((transferred, total) => {
+                                progressCalled = true;
+                            })
+                            .then(() => {
+                                TestCase.assertTrue(progressCalled);
+                                resolve();
+                            })
+                            .catch((e) => reject(e));
+
+                        setTimeout(function() {
+                            reject("Progress Notifications API failed to call progress callback for Realm constructor");
+                        }, 5000);
+                    });
+                });
+            });
+    },  
+
+    testProgressNotificationsForRealmOpenAsync() {
+        if (!isNodeProccess) {
+            return Promise.resolve();
+        }
+
+        const username = uuid();
+        const realmName = uuid();
+
+        return runOutOfProcess(__dirname + '/download-api-helper.js', username, realmName, REALM_MODULE_PATH)
+            .then(() => {
+                return promisifiedLogin('http://localhost:9080', username, 'password').then(user => {
+                    return new Promise((resolve, reject) => {
+                        let config = {
+                            sync: {
+                                user,
+                                url: `realm://localhost:9080/~/${realmName}`
+                            },
+                            schema: [{ name: 'Dog', properties: { name: 'string' } }],
+                        };
+
+                        let progressCalled = false;
+                        
+                        Realm.openAsync(config, 
+                            (transferred, total) => {
+                                progressCalled = true;
+                            }, 
+                            (error) => {
+                                if (error) {
+                                    reject(error);
+                                    return;
+                                }
+                                
+                                TestCase.assertTrue(progressCalled);
+                                resolve();
+                            });
+
+                        setTimeout(function() {
+                            reject("Progress Notifications API failed to call progress callback for Realm constructor");
+                        }, 5000);
+                    });
+                });
+            });
+    },  
 }