egroupware/api/js/webodf/collab/backend/pullbox/OperationRouter.js

/**
 * Copyright (C) 2013 KO GmbH <copyright@kogmbh.com>
 *
 * @licstart
 * This file is part of WebODF.
 *
 * WebODF is free software: you can redistribute it and/or modify it
 * under the terms of the GNU Affero General Public License (GNU AGPL)
 * as published by the Free Software Foundation, either version 3 of
 * the License, or (at your option) any later version.
 *
 * WebODF is distributed in the hope that it will be useful, but
 * WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with WebODF.  If not, see <http://www.gnu.org/licenses/>.
 * @licend
 *
 * @source: http://www.webodf.org/
 * @source: https://github.com/kogmbh/WebODF/
 */

/*global define, runtime, core, ops*/

webodfModule.define("webodf/editor/backend/pullbox/OperationRouter", [], function () {
    "use strict";

    // TODO: these eventid strings should be defined at OperationRouter interface
    var /**@const @type {!string}*/
        EVENT_BEFORESAVETOFILE =                  "beforeSaveToFile",
        /**@const @type {!string}*/
        EVENT_SAVEDTOFILE =                       "savedToFile",
        /**@const @type {!string}*/
        EVENT_HASLOCALUNSYNCEDOPERATIONSCHANGED = "hasLocalUnsyncedOperationsChanged",
        /**@const @type {!string}*/
        EVENT_HASSESSIONHOSTCONNECTIONCHANGED =   "hasSessionHostConnectionChanged";

    runtime.loadClass("ops.OperationTransformer");
    runtime.loadClass("core.EventNotifier");

    /**
     * route operations in a networked collaborative manner.
     *
     * incoming operations (from controller) are sent to a server,
     * who will distribute them.
     *
     * incoming operations (from the server are played on the DOM.
     */

    /**
     * @constructor
     * @implements ops.OperationRouter
     */
    return function PullBoxOperationRouter(sessionId, memberId, server, odfContainer, errorCallback) {

        var operationFactory,
            /**@type{function(!ops.Operation):boolean}*/
            playbackFunction,
            idleTimeout = null,
            syncOpsTimeout = null,
            /**@type{!boolean}*/
            isInstantSyncRequested = false,
            /**@type{!boolean}*/
            isPlayingUnplayedServerOpSpecs = false,
            /**@type{!boolean}*/
            isSyncCallRunning = false,
            /**@type{!boolean}*/
            hasError = false,
            /**@type{!boolean}*/
            syncingBlocked = false,
            /** @type {!string} id of latest op stack state known on the server */
            lastServerSeq = "",
            /** @type {!Array.<!Function>} sync request callbacks created since the last sync call to the server */
            syncRequestCallbacksQueue = [],
            /** @type {!Array.<!Object>} ops created since the last sync call to the server */
            unsyncedClientOpspecQueue = [],
            /** @type {!Array.<!Object>} ops already received from the server but not yet applied */
            unplayedServerOpspecQueue = [],
            /** @type {!Array.<!Function>} sync request callbacks which should be called after the received ops have been applied server */
            uncalledSyncRequestCallbacksQueue = [],
            /**@type{!boolean}*/
            hasLocalUnsyncedOps = false,
            /**@type{!boolean}*/
            hasSessionHostConnection = true,
            eventNotifier = new core.EventNotifier([
                EVENT_BEFORESAVETOFILE,
                EVENT_SAVEDTOFILE,
                EVENT_HASLOCALUNSYNCEDOPERATIONSCHANGED,
                EVENT_HASSESSIONHOSTCONNECTIONCHANGED,
                ops.OperationRouter.signalProcessingBatchStart,
                ops.OperationRouter.signalProcessingBatchEnd
            ]),
            /**@type{!boolean} tells if any local ops have been modifying ops */
            hasPushedModificationOps = false,
            operationTransformer = new ops.OperationTransformer(),
            /**@const*/replayTime = 500,
            /**@const*/syncOpsDelay = 3000,
            /**@const*/idleDelay = 5000;

        /**
         * @return {undefined}
         */
        function updateHasLocalUnsyncedOpsState() {
            var hasLocalUnsyncedOpsNow = (unsyncedClientOpspecQueue.length > 0);

            // no change?
            if (hasLocalUnsyncedOps === hasLocalUnsyncedOpsNow) {
                return;
            }

            hasLocalUnsyncedOps = hasLocalUnsyncedOpsNow;
            eventNotifier.emit(EVENT_HASLOCALUNSYNCEDOPERATIONSCHANGED, hasLocalUnsyncedOps);
        }

        /**
         * @param {!boolean} hasConnection
         * @return {undefined}
         */
        function updateHasSessionHostConnectionState(hasConnection) {
            // no change?
            if (hasSessionHostConnection === hasConnection) {
                return;
            }

            hasSessionHostConnection = hasConnection;
            eventNotifier.emit(EVENT_HASSESSIONHOSTCONNECTIONCHANGED, hasSessionHostConnection);
        }

        /**
         * @return {undefined}
         */
        function playUnplayedServerOpSpecs() {
            /**
             * @return {undefined}
             */
            function doPlayUnplayedServerOpSpecs() {
                var opspec, op, startTime, i;

                isPlayingUnplayedServerOpSpecs = false;

                // take start time
                startTime = Date.now();

                eventNotifier.emit(ops.OperationRouter.signalProcessingBatchStart, {});

                // apply as much as possible in the given time
                while (unplayedServerOpspecQueue.length > 0) {
                    // time over?
                    if (Date.now() - startTime > replayTime) {
                        break;
                    }

                    opspec = unplayedServerOpspecQueue.shift();

                    // use factory to create an instance, and playback!
                    op = operationFactory.create(opspec);
                    runtime.log(" op in: "+runtime.toJson(opspec));
                    if (op !== null) {
                        if (!playbackFunction(op)) {
                            eventNotifier.emit(ops.OperationRouter.signalProcessingBatchEnd, {});
                            hasError = true;
                            errorCallback("opExecutionFailure");
                            return;
                        }
                    } else {
                        eventNotifier.emit(ops.OperationRouter.signalProcessingBatchEnd, {});
                        hasError = true;
                        runtime.log("ignoring invalid incoming opspec: " + opspec);
                        errorCallback("unknownOpReceived");
                        return;
                    }
                }

                eventNotifier.emit(ops.OperationRouter.signalProcessingBatchEnd, {});

                // still unplayed opspecs?
                if (unplayedServerOpspecQueue.length > 0) {
                    // let other events be handled. then continue
                    isPlayingUnplayedServerOpSpecs = true;
                    runtime.getWindow().setTimeout(doPlayUnplayedServerOpSpecs, 1);
                } else {
                    // finally call all the callbacks waiting for that sync!
                    for (i = 0; i < uncalledSyncRequestCallbacksQueue.length; i += 1) {
                        uncalledSyncRequestCallbacksQueue[i]();
                    }

                    uncalledSyncRequestCallbacksQueue = [];
                }
            }

            if (isPlayingUnplayedServerOpSpecs) {
                return;
            }
            doPlayUnplayedServerOpSpecs();
        }

        /**
         * @param {Array.<!Object>} opspecs
         * @param {Array.<!Function>} callbacks
         * @return {undefined}
         */
        function receiveOpSpecsFromNetwork(opspecs, callbacks) {
            // append to existing unplayed
            unplayedServerOpspecQueue = unplayedServerOpspecQueue.concat(opspecs);
            uncalledSyncRequestCallbacksQueue = uncalledSyncRequestCallbacksQueue.concat(callbacks);
        }

        /**
         * Transforms the unsynced client ops and the server ops,
         * applies the server ops after transformation
         * @param {Array.<!Object>} serverOpspecs
         * @return {!boolean}
         */
        function handleOpsSyncConflict(serverOpspecs) {
            var i,
                transformResult;

            if (! serverOpspecs) {
                // TODO: proper error message, stop working
                runtime.BrowserRuntime$assert(false, "no opspecs received!");
                return false;
            } // TODO: more checking of proper content in serverOpspecs

            transformResult = operationTransformer.transform(unsyncedClientOpspecQueue, /**@type{!Array.<!Object>}*/(serverOpspecs));

            if (!transformResult) {
                return false;
            }

            // store transformed server ops
            for (i = 0; i < transformResult.opSpecsB.length; i += 1) {
                unplayedServerOpspecQueue.push(transformResult.opSpecsB[i]);
            }

            // store opspecs of all transformed client opspecs
            unsyncedClientOpspecQueue = [];
            for (i = 0; i < transformResult.opSpecsA.length; i += 1) {
                unsyncedClientOpspecQueue.push(transformResult.opSpecsA[i]);
            }

            return true;
        }

        /**
         * @return {undefined}
         */
        function syncOps() {
            var syncedClientOpspecs,
                syncRequestCallbacksArray;

            /**
             * @return {undefined}
             */
            function startSyncOpsTimeout() {
                idleTimeout = null;
                syncOpsTimeout = runtime.getWindow().setTimeout(function() {
                    syncOpsTimeout = null;
                    syncOps();
                }, syncOpsDelay);
            }

            if (isSyncCallRunning || hasError) {
                return;
            }
            // TODO: hack, remove
            if (syncingBlocked) {
                return;
            }

runtime.log("OperationRouter: sending sync_ops call");
            // no more instant pull request in any case
            isInstantSyncRequested = false;
            // set lock
            isSyncCallRunning = true;

            // take specs from queue, if any
            syncedClientOpspecs = unsyncedClientOpspecQueue;
            unsyncedClientOpspecQueue = [];
            syncRequestCallbacksArray = syncRequestCallbacksQueue;
            syncRequestCallbacksQueue = [];

            server.call({
                command: 'sync_ops',
                args: {
                    es_id: sessionId,
                    member_id: memberId,
                    seq_head: String(lastServerSeq),
                    client_ops: syncedClientOpspecs
                }
            }, function(responseData) {
                var response,
                    /**@type{!boolean}*/
                    hasUnresolvableConflict = false;

                updateHasSessionHostConnectionState(true);

                if (syncingBlocked) {
                    return;
                }

                try {
                    response = /** @type{{result:string, head_seq:string, ops:Array.<!Object>}} */(runtime.fromJson(responseData));
                } catch (e) {
                    hasError = true;
                    runtime.log("Could not parse reply: "+responseData);
                    errorCallback("unknownServerReply");
                    return;
                }
                // TODO: hack, remove
                runtime.log("sync_ops reply: " + responseData);

                // just new ops?
                if (response.result === "new_ops") {
                    if (response.ops.length > 0) {
                        // no new locally in the meantime?
                        if (unsyncedClientOpspecQueue.length === 0) {
                            receiveOpSpecsFromNetwork(response.ops, syncRequestCallbacksArray);
                        } else {
                            // transform server ops against new local ones and apply,
                            // transform and send new local ops to server
                            runtime.log("meh, have new ops locally meanwhile, have to do transformations.");
                            hasUnresolvableConflict = !handleOpsSyncConflict(response.ops);
                            syncRequestCallbacksQueue = syncRequestCallbacksArray.concat(syncRequestCallbacksQueue);
                       }
                        // and note server state
                        lastServerSeq = response.head_seq;
                    } else {
                        receiveOpSpecsFromNetwork([], syncRequestCallbacksArray);
                    }
                } else if (response.result === "added") {
                    runtime.log("All added to server");
                    receiveOpSpecsFromNetwork([], syncRequestCallbacksArray);
                    // note server state
                    lastServerSeq = response.head_seq;
                    updateHasLocalUnsyncedOpsState();
                } else if (response.result === "conflict") {
                    // put the send ops back into the outgoing queue
                    unsyncedClientOpspecQueue = syncedClientOpspecs.concat(unsyncedClientOpspecQueue);
                    syncRequestCallbacksQueue = syncRequestCallbacksArray.concat(syncRequestCallbacksQueue);
                    // transform server ops against new local ones and apply,
                    // transform and request new send new local ops to server
                    runtime.log("meh, server has new ops meanwhile, have to do transformations.");
                    hasUnresolvableConflict = !handleOpsSyncConflict(response.ops);
                    // and note server state
                    lastServerSeq = response.head_seq;
                    // try again instantly
                    if (!hasUnresolvableConflict) {
                        isInstantSyncRequested = true;
                    }
                } else if (response.result === "error") {
                    runtime.log("server reports an error: "+response.error);
                    hasError = true;
                    errorCallback(
                        response.error === "ENOSESSION" ? "sessionDoesNotExist":
                        response.error === "ENOMEMBER" ?  "notMemberOfSession":
                                                          "unknownServerReply"
                    );
                } else {
                    hasError = true;
                    runtime.log("Unexpected result on sync-ops call: "+response.result);
                    errorCallback("unknownServerReply");
                }
                if (hasError) {
                    return;
                }

                // unlock
                isSyncCallRunning = false;

                if (hasUnresolvableConflict) {
                    hasError = true;
                    errorCallback("unresolvableConflictingOps");
                } else {
                    // prepare next sync
                    if (isInstantSyncRequested) {
                        syncOps();
                    } else {
                        // nothing on client to sync?
                        if (unsyncedClientOpspecQueue.length === 0) {
                            idleTimeout = runtime.getWindow().setTimeout(startSyncOpsTimeout, idleDelay);
                        } else {
                            startSyncOpsTimeout();
                        }
                    }
                    playUnplayedServerOpSpecs();
                }
            }, function() {
                runtime.log("meh, server cannot be reached ATM.");
                // signal connection problem, but do not give up for now
                updateHasSessionHostConnectionState(false);
                // put the (not) send ops back into the outgoing queue
                unsyncedClientOpspecQueue = syncedClientOpspecs.concat(unsyncedClientOpspecQueue);
                syncRequestCallbacksQueue = syncRequestCallbacksArray.concat(syncRequestCallbacksQueue);
                // unlock
                isSyncCallRunning = false;
                // nothing on client to sync?
                if (unsyncedClientOpspecQueue.length === 0) {
                    idleTimeout = runtime.getWindow().setTimeout(startSyncOpsTimeout, idleDelay);
                } else {
                    startSyncOpsTimeout();
                }
                playUnplayedServerOpSpecs();
            });
        }

        function triggerPushingOps() {
            // disable any idle timeout
            if (idleTimeout) {
                runtime.clearTimeout(idleTimeout);
                idleTimeout = null;
            }

            // enable syncOps timeout, if needed
            if (!syncOpsTimeout && !isSyncCallRunning) {
runtime.log("OperationRouter: opsSync requested for pushing");
                syncOpsTimeout = runtime.getWindow().setTimeout(function() {
                    syncOpsTimeout = null;
                    syncOps();
                }, syncOpsDelay);
            }
        }

        /**
         * @param {!Funtion} cb
         * @return {undefined}
         */
        function requestInstantOpsSync(cb) {
            // register callback
            syncRequestCallbacksQueue.push(cb);

            // disable any idle timeout
            if (idleTimeout) {
                runtime.clearTimeout(idleTimeout);
                idleTimeout = null;
            }

            // disable any syncOps timeout
            if (syncOpsTimeout) {
                runtime.clearTimeout(syncOpsTimeout);
                syncOpsTimeout = null;
            }

runtime.log("OperationRouter: instant opsSync requested");
            isInstantSyncRequested = true;
            syncOps();
        }

        this.requestReplay = function (done_cb) {
            requestInstantOpsSync(done_cb);
        };

        /**
         * Sets the factory to use to create operation instances from operation specs.
         *
         * @param {!ops.OperationFactory} f
         * @return {undefined}
         */
        this.setOperationFactory = function (f) {
            operationFactory = f;
        };

        /**
         * Sets the method which should be called to apply operations.
         *
         * @param {!function(!ops.Operation):boolean} playback_func
         * @return {undefined}
         */
        this.setPlaybackFunction = function (playback_func) {
            playbackFunction = playback_func;
        };

        /**
         * Brings the locally created operations into the game.
         *
         * @param {!Array.<!ops.Operation>} operations
         * @return {undefined}
         */
        this.push = function (operations) {
            var i, op, opspec,
                timestamp = Date.now();

            if (hasError) {
                return;
            }
            // TODO: should be an assert in the future
            // there needs to be a flag telling that processing is happening,
            // and thus any input should be dropped in the sessioncontroller
            // ideally also have some UI element showing the processing state
            if (unplayedServerOpspecQueue.length > 0) {
                return;
            }

            eventNotifier.emit(ops.OperationRouter.signalProcessingBatchStart, {});

            for (i = 0; i < operations.length; i += 1) {
                op = operations[i];
                opspec = op.spec();

                // note if any local ops modified
                hasPushedModificationOps = hasPushedModificationOps || op.isEdit;

                // add timestamp TODO: improve the useless recreation of the op
                opspec.timestamp = timestamp;
                op = operationFactory.create(opspec);

                // apply locally
                if (!playbackFunction(op)) {
                    hasError = true;
                    errorCallback("opExecutionFailure");
                    return;
                }

                // send to server
                unsyncedClientOpspecQueue.push(opspec);
            }

            triggerPushingOps();

            updateHasLocalUnsyncedOpsState();

            eventNotifier.emit(ops.OperationRouter.signalProcessingBatchEnd, {});
        };

        /**
         * Requests a gracefull shutdown of the Operation Router.
         * Buffered operations shall be sent to the server.
         * A callback is called on success.
         */
        this.close = function (cb) {
            function cbDoneSaving(err) {
                eventNotifier.emit(EVENT_SAVEDTOFILE, null);
                cb(err);
            }

            function cbSuccess(fileData) {
                server.writeSessionStateToFile(sessionId, memberId, lastServerSeq, fileData, cbDoneSaving);
            }

            function doClose() {
                syncingBlocked = true;
                if (hasPushedModificationOps) {
                    eventNotifier.emit(EVENT_BEFORESAVETOFILE, null);

                    odfContainer.createByteArray(cbSuccess, cbDoneSaving);
                } else {
                    cb();
                }
            }

            if (hasError) {
                cb();
            } else if (hasLocalUnsyncedOps) {
                requestInstantOpsSync(doClose);
            } else {
                doClose();
            }
        };

        /**
         * @param {!string} eventId
         * @param {!Function} cb
         * @return {undefined}
         */
        this.subscribe = function (eventId, cb) {
            eventNotifier.subscribe(eventId, cb);
        };

        /**
         * @param {!string} eventId
         * @param {!Function} cb
         * @return {undefined}
         */
        this.unsubscribe = function (eventId, cb) {
            eventNotifier.unsubscribe(eventId, cb);
        };

        /**
         * @return {!boolean}
         */
        this.hasLocalUnsyncedOps = function () {
            return hasLocalUnsyncedOps;
        };

        /**
         * @return {!boolean}
         */
        this.hasSessionHostConnection = function () {
            return hasSessionHostConnection;
        };
    };
});