|
|
@@ -9,229 +9,232 @@ const Chunk = require('./chunk')
|
|
|
const Operation = require('./operation')
|
|
|
|
|
|
/**
|
|
|
- * @class
|
|
|
- * @classdesc
|
|
|
* Operational Transformation client.
|
|
|
*
|
|
|
* See OT.md for explanation.
|
|
|
*/
|
|
|
-function OtClient(_projectId, _editor, _blobStore, _socket) {
|
|
|
- const STATE_DISCONNECTED = 0
|
|
|
- const STATE_LOADING = 1
|
|
|
- const STATE_READY = 2
|
|
|
- const STATE_WAITING = 3
|
|
|
-
|
|
|
- let _version = null
|
|
|
- let _state = STATE_DISCONNECTED
|
|
|
- const _buffer = []
|
|
|
- let _ackVersion = null
|
|
|
- let _outstanding = []
|
|
|
- let _pending = []
|
|
|
- const _waiting = []
|
|
|
-
|
|
|
- this.connect = function otClientConnect() {
|
|
|
- switch (_state) {
|
|
|
- case STATE_DISCONNECTED:
|
|
|
- _state = STATE_LOADING
|
|
|
- _socket.emit('authenticate', {
|
|
|
- projectId: _projectId,
|
|
|
- token: 'letmein',
|
|
|
- })
|
|
|
- break
|
|
|
- default:
|
|
|
- throw new Error('connect in state ' + _state)
|
|
|
+class OtClient {
|
|
|
+ constructor(_projectId, _editor, _blobStore, _socket) {
|
|
|
+ const STATE_DISCONNECTED = 0
|
|
|
+ const STATE_LOADING = 1
|
|
|
+ const STATE_READY = 2
|
|
|
+ const STATE_WAITING = 3
|
|
|
+
|
|
|
+ let _version = null
|
|
|
+ let _state = STATE_DISCONNECTED
|
|
|
+ const _buffer = []
|
|
|
+ let _ackVersion = null
|
|
|
+ let _outstanding = []
|
|
|
+ let _pending = []
|
|
|
+ const _waiting = []
|
|
|
+
|
|
|
+ this.connect = function otClientConnect() {
|
|
|
+ switch (_state) {
|
|
|
+ case STATE_DISCONNECTED:
|
|
|
+ _state = STATE_LOADING
|
|
|
+ _socket.emit('authenticate', {
|
|
|
+ projectId: _projectId,
|
|
|
+ token: 'letmein',
|
|
|
+ })
|
|
|
+ break
|
|
|
+ default:
|
|
|
+ throw new Error('connect in state ' + _state)
|
|
|
+ }
|
|
|
}
|
|
|
- }
|
|
|
-
|
|
|
- /**
|
|
|
- * The latest project version number for which the client can construct the
|
|
|
- * project content.
|
|
|
- *
|
|
|
- * @return {number} non-negative
|
|
|
- */
|
|
|
- this.getVersion = function () {
|
|
|
- return _version
|
|
|
- }
|
|
|
|
|
|
- _socket.on('load', function otClientOnLoad(data) {
|
|
|
- switch (_state) {
|
|
|
- case STATE_LOADING: {
|
|
|
- const chunk = Chunk.fromRaw(data)
|
|
|
- const snapshot = chunk.getSnapshot()
|
|
|
- snapshot.applyAll(chunk.getChanges(), { strict: true })
|
|
|
- _version = chunk.getEndVersion()
|
|
|
- // TODO: we can get remote changes here, so it's not correct to wait for
|
|
|
- // the editor to load before transitioning to the READY state
|
|
|
- _editor.load(snapshot).then(function () {
|
|
|
- _state = STATE_READY
|
|
|
- })
|
|
|
- break
|
|
|
- }
|
|
|
- default:
|
|
|
- throw new Error('loaded in state ' + _state)
|
|
|
+ /**
|
|
|
+ * The latest project version number for which the client can construct the
|
|
|
+ * project content.
|
|
|
+ *
|
|
|
+ * @return {number} non-negative
|
|
|
+ */
|
|
|
+ this.getVersion = function () {
|
|
|
+ return _version
|
|
|
}
|
|
|
- })
|
|
|
|
|
|
- //
|
|
|
- // Local Operations
|
|
|
- //
|
|
|
+ _socket.on('load', function otClientOnLoad(data) {
|
|
|
+ switch (_state) {
|
|
|
+ case STATE_LOADING: {
|
|
|
+ const chunk = Chunk.fromRaw(data)
|
|
|
+ const snapshot = chunk.getSnapshot()
|
|
|
+ snapshot.applyAll(chunk.getChanges(), { strict: true })
|
|
|
+ _version = chunk.getEndVersion()
|
|
|
+ // TODO: we can get remote changes here, so it's not correct to wait for
|
|
|
+ // the editor to load before transitioning to the READY state
|
|
|
+ _editor.load(snapshot).then(function () {
|
|
|
+ _state = STATE_READY
|
|
|
+ })
|
|
|
+ break
|
|
|
+ }
|
|
|
+ default:
|
|
|
+ throw new Error('loaded in state ' + _state)
|
|
|
+ }
|
|
|
+ })
|
|
|
|
|
|
- function sendOutstandingChange() {
|
|
|
- const changeRequest = new ChangeRequest(_version, _outstanding)
|
|
|
- _socket.emit('change', changeRequest.toRaw())
|
|
|
- _state = STATE_WAITING
|
|
|
- }
|
|
|
+ //
|
|
|
+ // Local Operations
|
|
|
+ //
|
|
|
|
|
|
- function sendLocalOperation(operation) {
|
|
|
- _outstanding.push(operation)
|
|
|
- sendOutstandingChange()
|
|
|
- }
|
|
|
+ function sendOutstandingChange() {
|
|
|
+ const changeRequest = new ChangeRequest(_version, _outstanding)
|
|
|
+ _socket.emit('change', changeRequest.toRaw())
|
|
|
+ _state = STATE_WAITING
|
|
|
+ }
|
|
|
|
|
|
- function queueLocalOperation(operation) {
|
|
|
- _pending.push(operation)
|
|
|
- }
|
|
|
+ function sendLocalOperation(operation) {
|
|
|
+ _outstanding.push(operation)
|
|
|
+ sendOutstandingChange()
|
|
|
+ }
|
|
|
|
|
|
- this.handleLocalOperation = function otClientHandleLocalOperation(operation) {
|
|
|
- switch (_state) {
|
|
|
- case STATE_READY:
|
|
|
- sendLocalOperation(operation)
|
|
|
- break
|
|
|
- case STATE_WAITING:
|
|
|
- queueLocalOperation(operation)
|
|
|
- break
|
|
|
- default:
|
|
|
- throw new Error('local operation in state ' + _state)
|
|
|
+ function queueLocalOperation(operation) {
|
|
|
+ _pending.push(operation)
|
|
|
}
|
|
|
- }
|
|
|
|
|
|
- /**
|
|
|
- * A promise that resolves when the project reaches the given version.
|
|
|
- *
|
|
|
- * @param {number} version non-negative
|
|
|
- * @return {Promise}
|
|
|
- */
|
|
|
- this.waitForVersion = function otClientWaitForVersion(version) {
|
|
|
- if (!_waiting[version]) _waiting[version] = []
|
|
|
- return new BPromise(function (resolve, reject) {
|
|
|
- _waiting[version].push(resolve)
|
|
|
- })
|
|
|
- }
|
|
|
+ this.handleLocalOperation = function otClientHandleLocalOperation(
|
|
|
+ operation
|
|
|
+ ) {
|
|
|
+ switch (_state) {
|
|
|
+ case STATE_READY:
|
|
|
+ sendLocalOperation(operation)
|
|
|
+ break
|
|
|
+ case STATE_WAITING:
|
|
|
+ queueLocalOperation(operation)
|
|
|
+ break
|
|
|
+ default:
|
|
|
+ throw new Error('local operation in state ' + _state)
|
|
|
+ }
|
|
|
+ }
|
|
|
|
|
|
- function resolveWaitingPromises() {
|
|
|
- for (const version in _waiting) {
|
|
|
- if (!Object.prototype.hasOwnProperty.call(_waiting, version)) continue
|
|
|
- if (version > _version) continue
|
|
|
- _waiting[version].forEach(function (resolve) {
|
|
|
- resolve()
|
|
|
+ /**
|
|
|
+ * A promise that resolves when the project reaches the given version.
|
|
|
+ *
|
|
|
+ * @param {number} version non-negative
|
|
|
+ * @return {Promise}
|
|
|
+ */
|
|
|
+ this.waitForVersion = function otClientWaitForVersion(version) {
|
|
|
+ if (!_waiting[version]) _waiting[version] = []
|
|
|
+ return new BPromise(function (resolve, reject) {
|
|
|
+ _waiting[version].push(resolve)
|
|
|
})
|
|
|
- delete _waiting[version]
|
|
|
}
|
|
|
- }
|
|
|
|
|
|
- //
|
|
|
- // Messages from Server
|
|
|
- //
|
|
|
-
|
|
|
- function advanceIfReady() {
|
|
|
- if (_ackVersion !== null && _version === _ackVersion) {
|
|
|
- _version += 1
|
|
|
- _ackVersion = null
|
|
|
- handleAckReady()
|
|
|
- advanceIfReady()
|
|
|
- return
|
|
|
- }
|
|
|
- const changeNotes = _.remove(_buffer, function (changeNote) {
|
|
|
- return changeNote.getBaseVersion() === _version
|
|
|
- })
|
|
|
- if (changeNotes.length === 1) {
|
|
|
- handleRemoteChangeReady(changeNotes[0].getChange())
|
|
|
- _version += 1
|
|
|
- advanceIfReady()
|
|
|
- return
|
|
|
- }
|
|
|
- if (changeNotes.length !== 0) {
|
|
|
- throw new Error('multiple remote changes in client version ' + _version)
|
|
|
+ function resolveWaitingPromises() {
|
|
|
+ for (const version in _waiting) {
|
|
|
+ if (!Object.prototype.hasOwnProperty.call(_waiting, version)) continue
|
|
|
+ if (version > _version) continue
|
|
|
+ _waiting[version].forEach(function (resolve) {
|
|
|
+ resolve()
|
|
|
+ })
|
|
|
+ delete _waiting[version]
|
|
|
+ }
|
|
|
}
|
|
|
- }
|
|
|
|
|
|
- function bufferRemoteChangeNote(changeNote) {
|
|
|
- const version = changeNote.getBaseVersion()
|
|
|
- if (_.find(_buffer, 'baseVersion', version)) {
|
|
|
- throw new Error('multiple changes in version ' + version)
|
|
|
- }
|
|
|
- if (version === _ackVersion) {
|
|
|
- throw new Error('received change that was acked in ' + _ackVersion)
|
|
|
- }
|
|
|
- _buffer.push(changeNote)
|
|
|
- }
|
|
|
+ //
|
|
|
+ // Messages from Server
|
|
|
+ //
|
|
|
|
|
|
- function handleAckReady() {
|
|
|
- // console.log('handleAckReady')
|
|
|
- if (_outstanding.length === 0) {
|
|
|
- throw new Error('ack complete without outstanding change')
|
|
|
- }
|
|
|
- if (_state !== STATE_WAITING) {
|
|
|
- throw new Error('ack complete in state ' + _state)
|
|
|
+ function advanceIfReady() {
|
|
|
+ if (_ackVersion !== null && _version === _ackVersion) {
|
|
|
+ _version += 1
|
|
|
+ _ackVersion = null
|
|
|
+ handleAckReady()
|
|
|
+ advanceIfReady()
|
|
|
+ return
|
|
|
+ }
|
|
|
+ const changeNotes = _.remove(_buffer, function (changeNote) {
|
|
|
+ return changeNote.getBaseVersion() === _version
|
|
|
+ })
|
|
|
+ if (changeNotes.length === 1) {
|
|
|
+ handleRemoteChangeReady(changeNotes[0].getChange())
|
|
|
+ _version += 1
|
|
|
+ advanceIfReady()
|
|
|
+ return
|
|
|
+ }
|
|
|
+ if (changeNotes.length !== 0) {
|
|
|
+ throw new Error('multiple remote changes in client version ' + _version)
|
|
|
+ }
|
|
|
}
|
|
|
- _editor.handleChangeAcknowledged()
|
|
|
- resolveWaitingPromises()
|
|
|
- if (_pending.length > 0) {
|
|
|
- _outstanding = _pending
|
|
|
- _pending = []
|
|
|
- sendOutstandingChange()
|
|
|
- } else {
|
|
|
- _outstanding = []
|
|
|
- _state = STATE_READY
|
|
|
+
|
|
|
+ function bufferRemoteChangeNote(changeNote) {
|
|
|
+ const version = changeNote.getBaseVersion()
|
|
|
+ if (_.find(_buffer, 'baseVersion', version)) {
|
|
|
+ throw new Error('multiple changes in version ' + version)
|
|
|
+ }
|
|
|
+ if (version === _ackVersion) {
|
|
|
+ throw new Error('received change that was acked in ' + _ackVersion)
|
|
|
+ }
|
|
|
+ _buffer.push(changeNote)
|
|
|
}
|
|
|
- }
|
|
|
|
|
|
- function handleRemoteChangeReady(change) {
|
|
|
- if (_pending.length > 0) {
|
|
|
+ function handleAckReady() {
|
|
|
+ // console.log('handleAckReady')
|
|
|
if (_outstanding.length === 0) {
|
|
|
- throw new Error('pending change without outstanding change')
|
|
|
+ throw new Error('ack complete without outstanding change')
|
|
|
+ }
|
|
|
+ if (_state !== STATE_WAITING) {
|
|
|
+ throw new Error('ack complete in state ' + _state)
|
|
|
+ }
|
|
|
+ _editor.handleChangeAcknowledged()
|
|
|
+ resolveWaitingPromises()
|
|
|
+ if (_pending.length > 0) {
|
|
|
+ _outstanding = _pending
|
|
|
+ _pending = []
|
|
|
+ sendOutstandingChange()
|
|
|
+ } else {
|
|
|
+ _outstanding = []
|
|
|
+ _state = STATE_READY
|
|
|
}
|
|
|
}
|
|
|
|
|
|
- Operation.transformMultiple(_outstanding, change.getOperations())
|
|
|
- Operation.transformMultiple(_pending, change.getOperations())
|
|
|
+ function handleRemoteChangeReady(change) {
|
|
|
+ if (_pending.length > 0) {
|
|
|
+ if (_outstanding.length === 0) {
|
|
|
+ throw new Error('pending change without outstanding change')
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ Operation.transformMultiple(_outstanding, change.getOperations())
|
|
|
+ Operation.transformMultiple(_pending, change.getOperations())
|
|
|
|
|
|
- _editor.applyRemoteChange(change)
|
|
|
- }
|
|
|
+ _editor.applyRemoteChange(change)
|
|
|
+ }
|
|
|
|
|
|
- _socket.on('ack', function otClientOnAck(data) {
|
|
|
- switch (_state) {
|
|
|
- case STATE_WAITING: {
|
|
|
- const changeNote = ChangeNote.fromRaw(data)
|
|
|
- _ackVersion = changeNote.getBaseVersion()
|
|
|
- advanceIfReady()
|
|
|
- break
|
|
|
+ _socket.on('ack', function otClientOnAck(data) {
|
|
|
+ switch (_state) {
|
|
|
+ case STATE_WAITING: {
|
|
|
+ const changeNote = ChangeNote.fromRaw(data)
|
|
|
+ _ackVersion = changeNote.getBaseVersion()
|
|
|
+ advanceIfReady()
|
|
|
+ break
|
|
|
+ }
|
|
|
+ default:
|
|
|
+ throw new Error('ack in state ' + _state)
|
|
|
}
|
|
|
- default:
|
|
|
- throw new Error('ack in state ' + _state)
|
|
|
- }
|
|
|
- })
|
|
|
+ })
|
|
|
|
|
|
- _socket.on('change', function otClientOnChange(data) {
|
|
|
- switch (_state) {
|
|
|
- case STATE_READY:
|
|
|
- case STATE_WAITING:
|
|
|
- bufferRemoteChangeNote(ChangeNote.fromRaw(data))
|
|
|
- advanceIfReady()
|
|
|
- break
|
|
|
- default:
|
|
|
- throw new Error('remote change in state ' + _state)
|
|
|
- }
|
|
|
- })
|
|
|
-
|
|
|
- //
|
|
|
- // Connection State
|
|
|
- // TODO: socket.io error handling
|
|
|
- //
|
|
|
-
|
|
|
- _socket.on('disconnect', function () {
|
|
|
- _state = STATE_DISCONNECTED
|
|
|
- // eslint-disable-next-line no-console
|
|
|
- console.log('disconnected') // TODO: how do we handle disconnect?
|
|
|
- })
|
|
|
+ _socket.on('change', function otClientOnChange(data) {
|
|
|
+ switch (_state) {
|
|
|
+ case STATE_READY:
|
|
|
+ case STATE_WAITING:
|
|
|
+ bufferRemoteChangeNote(ChangeNote.fromRaw(data))
|
|
|
+ advanceIfReady()
|
|
|
+ break
|
|
|
+ default:
|
|
|
+ throw new Error('remote change in state ' + _state)
|
|
|
+ }
|
|
|
+ })
|
|
|
+
|
|
|
+ //
|
|
|
+ // Connection State
|
|
|
+ // TODO: socket.io error handling
|
|
|
+ //
|
|
|
+
|
|
|
+ _socket.on('disconnect', function () {
|
|
|
+ _state = STATE_DISCONNECTED
|
|
|
+ // eslint-disable-next-line no-console
|
|
|
+ console.log('disconnected') // TODO: how do we handle disconnect?
|
|
|
+ })
|
|
|
+ }
|
|
|
}
|
|
|
+
|
|
|
module.exports = OtClient
|