Merge pull request #121 from matrix-org/rav/realtime_callbacks

Implement HTTP callbacks in realtime

lib/http-api.js

@@ -21,6 +21,11 @@ limitations under the License.
 var q = require("q");
 var utils = require("./utils");
 
+// we use our own implementation of setTimeout, so that if we get suspended in
+// the middle of a /sync, we cancel the sync as soon as we awake, rather than
+// waiting for the delay to elapse.
+var callbacks = require("./realtime-callbacks");
+
 /*
 TODO:
 - CS: complete register function (doing stages)
@@ -125,12 +130,12 @@ module.exports.MatrixHttpApi.prototype = {
                 cb(new Error('Timeout'));
             };
 
-            xhr.timeout_timer = setTimeout(timeout_fn, 30000);
+            xhr.timeout_timer = callbacks.setTimeout(timeout_fn, 30000);
 
             xhr.onreadystatechange = function() {
                 switch (xhr.readyState) {
                     case global.XMLHttpRequest.DONE:
-                        clearTimeout(xhr.timeout_timer);
+                        callbacks.clearTimeout(xhr.timeout_timer);
                         var err;
                         if (!xhr.responseText) {
                             err = new Error('No response body.');
@@ -152,10 +157,10 @@ module.exports.MatrixHttpApi.prototype = {
                 }
             };
             xhr.upload.addEventListener("progress", function(ev) {
-                clearTimeout(xhr.timeout_timer);
+                callbacks.clearTimeout(xhr.timeout_timer);
                 upload.loaded = ev.loaded;
                 upload.total = ev.total;
-                xhr.timeout_timer = setTimeout(timeout_fn, 30000);
+                xhr.timeout_timer = callbacks.setTimeout(timeout_fn, 30000);
                 defer.notify(ev);
             });
             url += "?access_token=" + encodeURIComponent(this.opts.accessToken);
@@ -450,9 +455,13 @@ module.exports.MatrixHttpApi.prototype = {
 
         var timeoutId;
         var timedOut = false;
+        var req;
         if (localTimeoutMs) {
-            timeoutId = setTimeout(function() {
+            timeoutId = callbacks.setTimeout(function() {
                 timedOut = true;
+                if (req && req.abort) {
+                    req.abort();
+                }
                 defer.reject(new module.exports.MatrixError({
                     error: "Locally timed out waiting for a response",
                     errcode: "ORG.MATRIX.JSSDK_TIMEOUT",
@@ -464,7 +473,7 @@ module.exports.MatrixHttpApi.prototype = {
         var reqPromise = defer.promise;
 
         try {
-            var req = this.opts.request(
+            req = this.opts.request(
                 {
                     uri: uri,
                     method: method,
@@ -477,7 +486,7 @@ module.exports.MatrixHttpApi.prototype = {
                 },
                 function(err, response, body) {
                     if (localTimeoutMs) {
-                        clearTimeout(timeoutId);
+                        callbacks.clearTimeout(timeoutId);
                         if (timedOut) {
                             return; // already rejected promise
                         }

lib/realtime-callbacks.js

@@ -0,0 +1,203 @@
+/*
+Copyright 2016 OpenMarket Ltd
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+/* A re-implementation of the javascript callback functions (setTimeout,
+ * clearTimeout; setInterval and clearInterval are not yet implemented) which
+ * try to improve handling of large clock jumps (as seen when
+ * suspending/resuming the system).
+ *
+ * In particular, if a timeout would have fired while the system was suspended,
+ * it will instead fire as soon as possible after resume.
+ */
+
+"use strict";
+
+// we schedule a callback at least this often, to check if we've missed out on
+// some wall-clock time due to being suspended.
+var TIMER_CHECK_PERIOD_MS = 1000;
+
+// counter, for making up ids to return from setTimeout
+var _count = 0;
+
+// the key for our callback with the real global.setTimeout
+var _realCallbackKey;
+
+// a sorted list of the callbacks to be run.
+// each is an object with keys [runAt, func, params, key].
+var _callbackList = [];
+
+// var debuglog = console.log.bind(console);
+var debuglog = function() {};
+
+/**
+ * Replace the function used by this module to get the current time.
+ *
+ * Intended for use by the unit tests.
+ *
+ * @param {function} f function which should return a millisecond counter
+ *
+ * @internal
+ */
+module.exports.setNow = function(f) {
+    _now = f || Date.now;
+};
+var _now = Date.now;
+
+/**
+ * reimplementation of window.setTimeout, which will call the callback if
+ * the wallclock time goes past the deadline.
+ *
+ * @param {function} func   callback to be called after a delay
+ * @param {Number} delayMs  number of milliseconds to delay by
+ *
+ * @return {Number} an identifier for this callback, which may be passed into
+ *                   clearTimeout later.
+ */
+module.exports.setTimeout = function(func, delayMs) {
+    delayMs = delayMs || 0;
+    if (delayMs < 0) {
+        delayMs = 0;
+    }
+
+    var params = Array.prototype.slice.call(arguments, 2);
+    var runAt = _now() + delayMs;
+    var key = _count++;
+    debuglog("setTimeout: scheduling cb", key, "at", runAt,
+             "(delay", delayMs, ")");
+    var data = {
+        runAt: runAt,
+        func: func,
+        params: params,
+        key: key,
+    };
+
+    // figure out where it goes in the list
+    var idx = binarySearch(
+        _callbackList, function(el) {
+            return el.runAt - runAt;
+        }
+    );
+
+    _callbackList.splice(idx, 0, data);
+    _scheduleRealCallback();
+
+    return key;
+};
+
+/**
+ * reimplementation of window.clearTimeout, which mirrors setTimeout
+ *
+ * @param {Number} key   result from an earlier setTimeout call
+ */
+module.exports.clearTimeout = function(key) {
+    if (_callbackList.length === 0) {
+        return;
+    }
+
+    // remove the element from the list
+    var i;
+    for (i = 0; i < _callbackList.length; i++) {
+        var cb = _callbackList[i];
+        if (cb.key == key) {
+            _callbackList.splice(i, 1);
+            break;
+        }
+    }
+
+    // iff it was the first one in the list, reschedule our callback.
+    if (i === 0) {
+        _scheduleRealCallback();
+    }
+};
+
+// use the real global.setTimeout to schedule a callback to _runCallbacks.
+function _scheduleRealCallback() {
+    if (_realCallbackKey) {
+        global.clearTimeout(_realCallbackKey);
+    }
+
+    var first = _callbackList[0];
+
+    if (!first) {
+        debuglog("_scheduleRealCallback: no more callbacks, not rescheduling");
+        return;
+    }
+
+    var now = _now();
+    var delayMs = Math.min(first.runAt - now, TIMER_CHECK_PERIOD_MS);
+
+    debuglog("_scheduleRealCallback: now:", now, "delay:", delayMs);
+    _realCallbackKey = global.setTimeout(_runCallbacks, delayMs);
+}
+
+function _runCallbacks() {
+    var cb;
+    var now = _now();
+    debuglog("_runCallbacks: now:", now);
+
+    // get the list of things to call
+    var callbacksToRun = [];
+    while (true) {
+        var first = _callbackList[0];
+        if (!first || first.runAt > now) {
+            break;
+        }
+        cb = _callbackList.shift();
+        debuglog("_runCallbacks: popping", cb.key);
+        callbacksToRun.push(cb);
+    }
+
+    // reschedule the real callback before running our functions, to
+    // keep the codepaths the same whether or not our functions
+    // register their own setTimeouts.
+    _scheduleRealCallback();
+
+    for (var i = 0; i < callbacksToRun.length; i++) {
+        cb = callbacksToRun[i];
+        try {
+            cb.func.apply(null, cb.params);
+        } catch (e) {
+            console.error("Uncaught exception in callback function",
+                          e.stack || e);
+        }
+    }
+}
+
+
+/* search in a sorted array.
+ *
+ * returns the index of the last element for which func returns
+ * greater than zero, or array.length if no such element exists.
+ */
+function binarySearch(array, func) {
+    // min is inclusive, max exclusive.
+    var min = 0,
+        max = array.length;
+
+    while (min < max) {
+        var mid = (min + max) >> 1;
+        var res = func(array[mid]);
+        if (res > 0) {
+            // the element at 'mid' is too big; set it as the new max.
+            max = mid;
+        } else {
+            // the element at 'mid' is too small. 'min' is inclusive, so +1.
+            min = mid + 1;
+        }
+    }
+    // presumably, min==max now.
+    return min;
+}

spec/unit/realtime-callbacks.spec.js

@@ -0,0 +1,179 @@
+"use strict";
+
+var callbacks = require("../../lib/realtime-callbacks");
+var test_utils = require("../test-utils.js");
+
+describe("realtime-callbacks", function() {
+    var clock = jasmine.Clock;
+    var fakeDate;
+
+    function tick(millis) {
+        // make sure we tick the fakedate first, otherwise nothing will happen!
+        fakeDate += millis;
+        clock.tick(millis);
+    }
+
+    beforeEach(function() {
+        test_utils.beforeEach(this);
+        clock.useMock();
+        fakeDate = Date.now();
+        callbacks.setNow(function() { return fakeDate; });
+    });
+
+    afterEach(function() {
+        callbacks.setNow();
+    });
+
+    describe("setTimeout", function() {
+        it("should call the callback after the timeout", function() {
+            var callback = jasmine.createSpy();
+            callbacks.setTimeout(callback, 100);
+
+            expect(callback).not.toHaveBeenCalled();
+            tick(100);
+            expect(callback).toHaveBeenCalled();
+        });
+
+
+        it("should default to a zero timeout", function() {
+            var callback = jasmine.createSpy();
+            callbacks.setTimeout(callback);
+
+            expect(callback).not.toHaveBeenCalled();
+            tick(0);
+            expect(callback).toHaveBeenCalled();
+        });
+
+        it("should pass any parameters to the callback", function() {
+            var callback = jasmine.createSpy();
+            callbacks.setTimeout(callback, 0, "a", "b", "c");
+            tick(0);
+            expect(callback).toHaveBeenCalledWith("a", "b", "c");
+        });
+
+        it("should set 'this' to the global object", function() {
+            var callback = jasmine.createSpy();
+            callback.andCallFake(function() {
+                expect(this).toBe(global);
+                expect(this.console).toBeDefined();
+            });
+            callbacks.setTimeout(callback);
+            tick(0);
+            expect(callback).toHaveBeenCalled();
+        });
+
+        it("should handle timeouts of several seconds", function() {
+            var callback = jasmine.createSpy();
+            callbacks.setTimeout(callback, 2000);
+
+            expect(callback).not.toHaveBeenCalled();
+            for (var i = 0; i < 4; i++) {
+                tick(500);
+            }
+            expect(callback).toHaveBeenCalled();
+        });
+
+        it("should call multiple callbacks in the right order", function() {
+            var callback1 = jasmine.createSpy("callback1");
+            var callback2 = jasmine.createSpy("callback2");
+            var callback3 = jasmine.createSpy("callback3");
+            callbacks.setTimeout(callback2, 200);
+            callbacks.setTimeout(callback1, 100);
+            callbacks.setTimeout(callback3, 300);
+
+            expect(callback1).not.toHaveBeenCalled();
+            expect(callback2).not.toHaveBeenCalled();
+            expect(callback3).not.toHaveBeenCalled();
+            tick(100);
+            expect(callback1).toHaveBeenCalled();
+            expect(callback2).not.toHaveBeenCalled();
+            expect(callback3).not.toHaveBeenCalled();
+            tick(100);
+            expect(callback1).toHaveBeenCalled();
+            expect(callback2).toHaveBeenCalled();
+            expect(callback3).not.toHaveBeenCalled();
+            tick(100);
+            expect(callback1).toHaveBeenCalled();
+            expect(callback2).toHaveBeenCalled();
+            expect(callback3).toHaveBeenCalled();
+        });
+
+        it("should treat -ve timeouts the same as a zero timeout", function() {
+            var callback1 = jasmine.createSpy("callback1");
+            var callback2 = jasmine.createSpy("callback2");
+
+            // check that cb1 is called before cb2
+            callback1.andCallFake(function() {
+                expect(callback2).not.toHaveBeenCalled();
+            });
+
+            callbacks.setTimeout(callback1);
+            callbacks.setTimeout(callback2, -100);
+
+            expect(callback1).not.toHaveBeenCalled();
+            expect(callback2).not.toHaveBeenCalled();
+            tick(0);
+            expect(callback1).toHaveBeenCalled();
+            expect(callback2).toHaveBeenCalled();
+        });
+
+        it("should not get confused by chained calls", function() {
+            var callback2 = jasmine.createSpy("callback2");
+            var callback1 = jasmine.createSpy("callback1");
+            callback1.andCallFake(function() {
+                callbacks.setTimeout(callback2, 0);
+                expect(callback2).not.toHaveBeenCalled();
+            });
+
+            callbacks.setTimeout(callback1);
+            expect(callback1).not.toHaveBeenCalled();
+            expect(callback2).not.toHaveBeenCalled();
+            tick(0);
+            expect(callback1).toHaveBeenCalled();
+            expect(callback2).toHaveBeenCalled();
+        });
+
+        it("should be immune to exceptions", function() {
+            var callback1 = jasmine.createSpy("callback1");
+            callback1.andCallFake(function() {
+                throw new Error("prepare to die");
+            });
+            var callback2 = jasmine.createSpy("callback2");
+            callbacks.setTimeout(callback1, 0);
+            callbacks.setTimeout(callback2, 0);
+
+            expect(callback1).not.toHaveBeenCalled();
+            expect(callback2).not.toHaveBeenCalled();
+            tick(0);
+            expect(callback1).toHaveBeenCalled();
+            expect(callback2).toHaveBeenCalled();
+        });
+
+    });
+
+    describe("cancelTimeout", function() {
+        it("should cancel a pending timeout", function() {
+            var callback = jasmine.createSpy();
+            var k = callbacks.setTimeout(callback);
+            callbacks.clearTimeout(k);
+            tick(0);
+            expect(callback).not.toHaveBeenCalled();
+        });
+
+        it("should not affect sooner timeouts", function() {
+            var callback1 = jasmine.createSpy("callback1");
+            var callback2 = jasmine.createSpy("callback2");
+
+            callbacks.setTimeout(callback1, 100);
+            var k = callbacks.setTimeout(callback2, 200);
+            callbacks.clearTimeout(k);
+
+            tick(100);
+            expect(callback1).toHaveBeenCalled();
+            expect(callback2).not.toHaveBeenCalled();
+
+            tick(150);
+            expect(callback2).not.toHaveBeenCalled();
+        });
+    });
+});