diff --git a/rollup.test.js b/rollup.test.js index 8b2b604d..dca3e47d 100644 --- a/rollup.test.js +++ b/rollup.test.js @@ -3,7 +3,7 @@ import commonjs from 'rollup-plugin-commonjs' import multiEntry from 'rollup-plugin-multi-entry' export default { - entry: 'test/{encode-decode,red-black-tree}.js', + entry: 'test/y-array.tests.js', moduleName: 'y-tests', format: 'umd', plugins: [ diff --git a/src/Binary/Decoder.js b/src/Binary/Decoder.js index 72c4453b..07d9de37 100644 --- a/src/Binary/Decoder.js +++ b/src/Binary/Decoder.js @@ -1,4 +1,6 @@ import utf8 from 'utf-8' +import ID from '../Util/ID.js' +import { default as RootID, RootFakeUserID } from '../Util/RootID.js' export default class BinaryDecoder { constructor (buffer) { @@ -107,9 +109,11 @@ export default class BinaryDecoder { */ readID () { let user = this.readVarUint() - if (user === 0xFFFFFF) { + if (user === RootFakeUserID) { // read property name and type id - return new RootID(this.readVarString(), this.readVarUint()) + const rid = new RootID(this.readVarString(), null) + rid.type = this.readVarUint() + return rid } return new ID(user, this.readVarUint()) } diff --git a/src/Binary/Encoder.js b/src/Binary/Encoder.js index dfa62a5f..f080ccaf 100644 --- a/src/Binary/Encoder.js +++ b/src/Binary/Encoder.js @@ -1,4 +1,5 @@ import utf8 from 'utf-8' +import { RootFakeUserID } from '../Util/RootID.js' const bits7 = 0b1111111 const bits8 = 0b11111111 @@ -68,13 +69,14 @@ export default class BinaryEncoder { } } - writeOpID (id) { - let user = id[0] + writeID (id) { + const user = id.user this.writeVarUint(user) - if (user !== 0xFFFFFF) { - this.writeVarUint(id[1]) + if (user !== RootFakeUserID) { + this.writeVarUint(id.clock) } else { - this.writeVarString(id[1]) + this.writeVarString(id.name) + this.writeVarUint(id.type) } } } diff --git a/src/Connector.js b/src/Connector.js index c22cea6f..d7ac8f4b 100644 --- a/src/Connector.js +++ b/src/Connector.js @@ -41,7 +41,6 @@ export default class AbstractConnector { reconnect () { this.log('reconnecting..') - return this.y.db.startGarbageCollector() } disconnect () { @@ -63,7 +62,7 @@ export default class AbstractConnector { userLeft (user) { if (this.connections.has(user)) { - this.log('%s: User left %s', this.userId, user) + this.log('%s: User left %s', this.y.userID, user) this.connections.delete(user) // check if isSynced event can be sent now this._setSyncedWith(null) @@ -83,7 +82,7 @@ export default class AbstractConnector { if (this.connections.has(user)) { throw new Error('This user already joined!') } - this.log('%s: User joined %s', this.userId, user) + this.log('%s: User joined %s', this.y.userID, user) this.connections.set(user, { uid: user, isSynced: false, @@ -115,33 +114,32 @@ export default class AbstractConnector { } } - _syncWithUser (userid) { + _syncWithUser (userID) { if (this.role === 'slave') { return // "The current sync has not finished or this is controlled by a master!" } - sendSyncStep1(this, userid) + sendSyncStep1(this, userID) } _fireIsSyncedListeners () { - new Promise().then(() => { + setTimeout(() => { if (!this.isSynced) { this.isSynced = true // It is safer to remove this! - // TODO: remove: this.garbageCollectAfterSync() // call whensynced listeners for (var f of this.whenSyncedListeners) { f() } this.whenSyncedListeners = [] } - }) + }, 0) } send (uid, buffer) { if (!(buffer instanceof ArrayBuffer || buffer instanceof Uint8Array)) { throw new Error('Expected Message to be an ArrayBuffer or Uint8Array - don\'t use this method to send custom messages') } - this.log('%s: Send \'%y\' to %s', this.userId, buffer, uid) + this.log('%s: Send \'%y\' to %s', this.y.userID, buffer, uid) this.logMessage('Message: %Y', buffer) } @@ -149,7 +147,7 @@ export default class AbstractConnector { if (!(buffer instanceof ArrayBuffer || buffer instanceof Uint8Array)) { throw new Error('Expected Message to be an ArrayBuffer or Uint8Array - don\'t use this method to send custom messages') } - this.log('%s: Broadcast \'%y\'', this.userId, buffer) + this.log('%s: Broadcast \'%y\'', this.y.userID, buffer) this.logMessage('Message: %Y', buffer) } @@ -157,7 +155,11 @@ export default class AbstractConnector { Buffer operations, and broadcast them when ready. */ broadcastStruct (struct) { - let firstContent = this.broadcastBuffer.length === 0 + const firstContent = this.broadcastBuffer.length === 0 + if (firstContent) { + this.broadcastBuffer.writeVarString(this.y.room) + this.broadcastBuffer.writeVarString('update') + } struct._toBinary(this.broadcastBuffer) if (this.maxBufferLength > 0 && this.broadcastBuffer.length > this.maxBufferLength) { // it is necessary to send the buffer now @@ -172,7 +174,7 @@ export default class AbstractConnector { // (or buffer exceeds maxBufferLength) setTimeout(() => { if (this.broadcastBuffer.length > 0) { - this.broadcast(this.broadcastBuffer) + this.broadcast(this.broadcastBuffer.createBuffer()) this.broadcastBuffer = new BinaryEncoder() } }) @@ -201,7 +203,7 @@ export default class AbstractConnector { if (!(buffer instanceof ArrayBuffer || buffer instanceof Uint8Array)) { return Promise.reject(new Error('Expected Message to be an ArrayBuffer or Uint8Array!')) } - if (sender === this.userId) { + if (sender === this.y.userID) { return Promise.resolve() } let decoder = new BinaryDecoder(buffer) @@ -210,7 +212,7 @@ export default class AbstractConnector { encoder.writeVarString(roomname) let messageType = decoder.readVarString() let senderConn = this.connections.get(sender) - this.log('%s: Receive \'%s\' from %s', this.userId, messageType, sender) + this.log('%s: Receive \'%s\' from %s', this.y.userID, messageType, sender) this.logMessage('Message: %Y', buffer) if (senderConn == null && !skipAuth) { throw new Error('Received message from unknown peer!') @@ -247,7 +249,7 @@ export default class AbstractConnector { computeMessage (messageType, senderConn, decoder, encoder, sender, skipAuth) { if (messageType === 'sync step 1' && (senderConn.auth === 'write' || senderConn.auth === 'read')) { // cannot wait for sync step 1 to finish, because we may wait for sync step 2 in sync step 1 (->lock) - readSyncStep1()(decoder, encoder, this.y, senderConn, sender) + readSyncStep1(decoder, encoder, this.y, senderConn, sender) } else if (messageType === 'sync step 2' && senderConn.auth === 'write') { readSyncStep2(decoder, encoder, this.y, senderConn, sender) } else if (messageType === 'update' && (skipAuth || senderConn.auth === 'write')) { diff --git a/src/MessageHandler/deleteSet.js b/src/MessageHandler/deleteSet.js index 555183ae..e261639c 100644 --- a/src/MessageHandler/deleteSet.js +++ b/src/MessageHandler/deleteSet.js @@ -1,4 +1,4 @@ -import deleteItemRange from 'deleteItemRange' +import { deleteItemRange } from '../Struct/Delete.js' export function stringifyDeleteSet (y, decoder, strBuilder) { let dsLength = decoder.readUint32() @@ -48,7 +48,7 @@ export function writeDeleteSet (y, encoder) { if (currentUser !== null) { // happens on first iteration encoder.setUint32(lastLenPos, currentLength) } - encoder.writeUint32(laterDSLenPus, numberOfUsers) + encoder.setUint32(laterDSLenPus, numberOfUsers) } export function readDeleteSet (y, decoder) { diff --git a/src/MessageHandler/integrateRemoteStructs.js b/src/MessageHandler/integrateRemoteStructs.js index b969c737..05bd393a 100644 --- a/src/MessageHandler/integrateRemoteStructs.js +++ b/src/MessageHandler/integrateRemoteStructs.js @@ -45,7 +45,7 @@ export function integrateRemoteStructs (decoder, encoder, y) { let reference = decoder.readVarUint() let Constr = getStruct(reference) let struct = new Constr() - let missing = struct._fromBinary(decoder) + let missing = struct._fromBinary(y, decoder) if (missing.length === 0) { while (struct != null) { _integrateRemoteStructHelper(y, struct) diff --git a/src/MessageHandler/stateSet.js b/src/MessageHandler/stateSet.js index 88b5a577..3ec21c99 100644 --- a/src/MessageHandler/stateSet.js +++ b/src/MessageHandler/stateSet.js @@ -10,15 +10,14 @@ export function readStateSet (decoder) { return ss } -export function writeStateSet (encoder) { +export function writeStateSet (y, encoder) { let lenPosition = encoder.pos let len = 0 encoder.writeUint32(0) - this.ss.iterate(null, null, function (n) { - encoder.writeVarUint(n.id[0]) - encoder.writeVarUint(n.clock) + for (let [user, clock] of y.ss.state) { + encoder.writeVarUint(user) + encoder.writeVarUint(clock) len++ - }) + } encoder.setUint32(lenPosition, len) - return len === 0 } diff --git a/src/MessageHandler/syncStep1.js b/src/MessageHandler/syncStep1.js index a4f53680..632a61ff 100644 --- a/src/MessageHandler/syncStep1.js +++ b/src/MessageHandler/syncStep1.js @@ -1,4 +1,7 @@ import BinaryEncoder from '../Binary/Encoder.js' +import { readStateSet, writeStateSet } from './stateSet.js' +import { writeDeleteSet } from './deleteSet.js' +import ID from '../Util/ID.js' export function stringifySyncStep1 (decoder, strBuilder) { let auth = decoder.readVarString() @@ -17,14 +20,22 @@ export function stringifySyncStep1 (decoder, strBuilder) { } } -export function sendSyncStep1 (y, syncUser) { +export function sendSyncStep1 (connector, syncUser) { let encoder = new BinaryEncoder() - encoder.writeVarString(y.room) + encoder.writeVarString(connector.y.room) encoder.writeVarString('sync step 1') - encoder.writeVarString(y.connector.authInfo || '') - encoder.writeVarUint(y.connector.protocolVersion) - y.ss.writeStateSet(encoder) - y.connector.send(syncUser, encoder.createBuffer()) + encoder.writeVarString(connector.authInfo || '') + encoder.writeVarUint(connector.protocolVersion) + writeStateSet(connector.y, encoder) + connector.send(syncUser, encoder.createBuffer()) +} + +export default function writeStructs (encoder, decoder, y, ss) { + for (let [user, clock] of ss) { + y.os.iterate(new ID(user, clock), null, function (struct) { + struct._toBinary(encoder) + }) + } } export function readSyncStep1 (decoder, encoder, y, senderConn, sender) { @@ -32,22 +43,20 @@ export function readSyncStep1 (decoder, encoder, y, senderConn, sender) { // check protocol version if (protocolVersion !== y.connector.protocolVersion) { console.warn( - `You tried to sync with a yjs instance that has a different protocol version + `You tried to sync with a Yjs instance that has a different protocol version (You: ${protocolVersion}, Client: ${protocolVersion}). - The sync was stopped. You need to upgrade your dependencies (especially Yjs & the Connector)! `) y.destroy() } - - // send sync step 2 + // write sync step 2 encoder.writeVarString('sync step 2') encoder.writeVarString(y.connector.authInfo || '') - writeDeleteSet(encoder) - // reads ss and writes os - writeOperations(encoder, decoder) + writeDeleteSet(y, encoder) + const ss = readStateSet(decoder) + writeStructs(encoder, decoder, y, ss) y.connector.send(senderConn.uid, encoder.createBuffer()) senderConn.receivedSyncStep2 = true if (y.connector.role === 'slave') { - sendSyncStep1(y, sender) + sendSyncStep1(y.connector, sender) } } diff --git a/src/MessageHandler/syncStep2.js b/src/MessageHandler/syncStep2.js index 3b0c6b70..13c1642e 100644 --- a/src/MessageHandler/syncStep2.js +++ b/src/MessageHandler/syncStep2.js @@ -1,6 +1,6 @@ import { integrateRemoteStructs } from './integrateRemoteStructs.js' import { stringifyUpdate } from './update.js' -import ID from '../Util/ID.js' +import { readDeleteSet } from './deleteSet.js' export function stringifySyncStep2 (decoder, strBuilder) { strBuilder.push(' - auth: ' + decoder.readVarString() + '\n') @@ -22,27 +22,8 @@ export function stringifySyncStep2 (decoder, strBuilder) { } } -export function writeSyncStep2 () { - // TODO -} - -export default function writeStructs (encoder, decoder, y, ss) { - let lenPos = encoder.pos - let len = 0 - encoder.writeUint32(0) - for (let [user, clock] of ss) { - y.os.iterate(new ID(user, clock), null, function (struct) { - struct._toBinary(y, encoder) - len++ - }) - } - encoder.setUint32(lenPos, len) -} - export function readSyncStep2 (decoder, encoder, y, senderConn, sender) { - // apply operations first - applyDeleteSet(decoder) + readDeleteSet(y, decoder) integrateRemoteStructs(decoder, encoder, y) - // then apply ds y.connector._setSyncedWith(sender) } diff --git a/src/Persistence.js b/src/Persistence.js index 1cdb40ce..8582b735 100644 --- a/src/Persistence.js +++ b/src/Persistence.js @@ -35,7 +35,6 @@ export default function extendPersistence (Y) { } if (this.saveOperationsBuffer.length === 0) { this.saveOperationsBuffer = ops - this.y.db.whenTransactionsFinished().then(saveOperations) } else { this.saveOperationsBuffer = this.saveOperationsBuffer.concat(ops) } diff --git a/src/Store/DeleteStore.js b/src/Store/DeleteStore.js index 7477ab54..873126be 100644 --- a/src/Store/DeleteStore.js +++ b/src/Store/DeleteStore.js @@ -17,6 +17,18 @@ export default class DeleteStore extends Tree { var n = this.ds.findWithUpperBound(id) return n != null && n.id[0] === id[0] && id[1] < n.id[1] + n.len } + applyMissingDeletesOnStruct (struct) { + const strID = struct._id + // find most right delete + let n = this.findWithUpperBound(new ID(strID.user, strID.clock + struct.length - 1)) + if (n === null || n.id.user !== strID.user || n.id.clock + n.length <= strID.clock) { + // struct is not deleted + return null + } + // TODO: + // * iterate to the right and apply new Delete's + throw new Error('Not implemented!') + } /* * Mark an operation as deleted. returns the deleted node */ diff --git a/src/Store/OperationStore.js b/src/Store/OperationStore.js index a3e03e8a..f6be01f8 100644 --- a/src/Store/OperationStore.js +++ b/src/Store/OperationStore.js @@ -3,83 +3,61 @@ import RootID from '../Util/ID.js' import { getStruct } from '../Util/structReferences.js' export default class OperationStore extends Tree { + constructor (y) { + super() + this.y = y + } get (id) { let struct = this.find(id) if (struct === null && id instanceof RootID) { let Constr = getStruct(id.type) struct = new Constr() struct._id = id + struct._parent = this.y this.put(struct) } return struct } + // Use getItem for structs with _length > 1 getItem (id) { var item = this.findWithUpperBound(id) - if (item == null) { + if (item === null) { return null } - var len = item.content != null ? item.content.length : 1 // in case of opContent - if (id[0] === item.id[0] && id[1] < item.id[1] + len) { + const itemID = item._id + if (id.user === itemID.user && id.clock < itemID.clock + item._length) { return item } else { return null } - } // Return an insertion such that id is the first element of content - // This function manipulates an operation, if necessary - getInsertionCleanStart (id) { - var ins = this.getInsertion(id) - if (ins != null) { - if (ins.id[1] === id[1]) { - return ins - } else { - var left = Y.utils.copyObject(ins) - ins.content = left.content.splice(id[1] - ins.id[1]) - ins.id = id - var leftLid = Y.utils.getLastId(left) - ins.origin = leftLid - left.originOf = [ins.id] - left.right = ins.id - ins.left = leftLid - // debugger // check - this.setOperation(left) - this.setOperation(ins) - if (left.gc) { - this.store.queueGarbageCollector(ins.id) - } - return ins - } + // This function manipulates an item, if necessary + getItemCleanStart (id) { + var ins = this.getItem(id) + if (ins === null || ins._length === 1) { + return ins + } + const insID = ins._id + if (insID.clock === id.clock) { + return ins } else { - return null + return ins._splitAt(this.y, id.clock - insID.clock) } } // Return an insertion such that id is the last element of content // This function manipulates an operation, if necessary - getInsertionCleanEnd (id) { - var ins = this.getInsertion(id) - if (ins != null) { - if (ins.content == null || (ins.id[1] + ins.content.length - 1 === id[1])) { - return ins - } else { - var right = Y.utils.copyObject(ins) - right.content = ins.content.splice(id[1] - ins.id[1] + 1) // cut off remainder - right.id = [id[0], id[1] + 1] - var insLid = Y.utils.getLastId(ins) - right.origin = insLid - ins.originOf = [right.id] - ins.right = right.id - right.left = insLid - // debugger // check - this.setOperation(right) - this.setOperation(ins) - if (ins.gc) { - this.store.queueGarbageCollector(right.id) - } - return ins - } + getItemCleanEnd (id) { + var ins = this.getItem(id) + if (ins === null || ins._length === 1) { + return ins + } + const insID = ins._id + if (insID.clock + ins._length - 1 === id.clock) { + return ins } else { - return null + ins._splitAt(this.y, id.clock - insID.clock + 1) + return ins } } } diff --git a/src/Struct/Delete.js b/src/Struct/Delete.js index 4b9bd850..e195b1c2 100644 --- a/src/Struct/Delete.js +++ b/src/Struct/Delete.js @@ -1,26 +1,44 @@ -import StructManager from '../Util/StructManager.js' +import { getReference } from '../Util/structReferences.js' +export function deleteItemRange (y, user, clock, range) { + let items = y.os.getItems(this._target, this._length) + for (let i = items.length - 1; i >= 0; i--) { + items[i]._delete(y, false) + } +} + +/** + * Delete is not a real struct. It will not be saved in OS + */ export default class Delete { constructor () { - this._target = null + this._targetID = null this._length = null } _fromBinary (y, decoder) { - this._targetID = decoder.readOpID() + this._targetID = decoder.readID() this._length = decoder.readVarUint() } - _toBinary (y, encoder) { - encoder.writeUint8(StructManager.getReference(this.constructor)) - encoder.writeOpID(this._targetID) + _toBinary (encoder) { + encoder.writeUint8(getReference(this.constructor)) + encoder.writeID(this._targetID) encoder.writeVarUint(this._length) } - _integrate (y) { - let items = y.os.getItems(this._target, this._length) - for (let i = items.length - 1; i >= 0; i--) { - items[i]._delete() + /** + * - If created remotely (a remote user deleted something), + * this Delete is applied to all structs in id-range. + * - If created lokally (e.g. when y-array deletes a range of elements), + * this struct is broadcasted only (it is already executed) + */ + _integrate (y, locallyCreated = false) { + if (!locallyCreated) { + // from remote + const id = this._targetID + deleteItemRange(y, id.user, id.clock, this._length) + } else { + // from local + y.connector.broadcastStruct(this) } - // TODO: only broadcast if created by local user or if y.connector._forwardAppliedStructs.. - y.connector.broadcastStruct(this) if (y.persistence !== null) { y.persistence.saveOperations(this) } diff --git a/src/Struct/Item.js b/src/Struct/Item.js index 2b1ab0eb..4f7413e4 100644 --- a/src/Struct/Item.js +++ b/src/Struct/Item.js @@ -1,3 +1,27 @@ +import { getReference } from '../Util/structReferences.js' +import ID from '../Util/ID.js' +import { RootFakeUserID } from '../Util/RootID.js' +import Delete from './Delete.js' + +/** + * Helper utility to split an Item (see _splitAt) + * - copy all properties from a to b + * - connect a to b + * - assigns the correct _id + * - save b to os + */ +export function splitHelper (y, a, b, diff) { + const aID = a._id + b._id = new ID(aID.user, aID.clock + diff) + b._origin = a + b._left = a + a._right = b + a._right_origin = b + b._parent = a._parent + b._parentSub = a._parentSub + b._deleted = a._deleted + y.os.put(b) +} export default class Item { constructor () { @@ -13,22 +37,30 @@ export default class Item { get _length () { return 1 } - _getDistanceToOrigin () { - if (this.left == null) { - return 0 - } else { - var d = 0 - var o = this.left - while (o !== null && !this.origin.equals(o.id)) { - d++ - o = o.left - } - return d + /** + * Splits this struct so that another struct can be inserted in-between. + * This must be overwritten if _length > 1 + * Returns right part after split + * - diff === 0 => this + * - diff === length => this._right + * - otherwise => split _content and return right part of split + * (see ItemJSON/ItemString for implementation) + */ + _splitAt (y, diff) { + if (diff === 0) { + return this } + return this._right } - _delete (y) { + _delete (y, createDelete = true) { this._deleted = true y.ds.markDeleted(this._id, this._length) + if (createDelete) { + let del = new Delete() + del._targetID = this._id + del._length = this._length + del._integrate(y, true) + } } /* * - Integrate the struct so that other types/structs can see it @@ -36,8 +68,12 @@ export default class Item { * - Check if this is struct deleted */ _integrate (y) { - if (this._id === null) { + const selfID = this._id + if (selfID === null) { this._id = y.ss.getNextID(this._length) + } else if (selfID.clock < y.ss.getState(selfID.user)) { + // already applied.. + return [] } /* # $this has to find a unique position between origin and the next known character @@ -71,86 +107,145 @@ export default class Item { // Note that conflictingItems is a subset of itemsBeforeOrigin while (o !== null && o !== this._right) { itemsBeforeOrigin.add(o) - if (this.origin === o.origin) { + if (this._origin === o._origin) { // case 1 if (o._id.user < this._id.user) { - this.left = o + this._left = o conflictingItems = new Set() } } else if (itemsBeforeOrigin.has(o)) { // case 2 if (conflictingItems.has(o)) { - this.left = o + this._left = o conflictingItems = new Set() } } else { break } - o = o.right + o = o._right } - y.os.set(this) - y.ds.checkIfDeleted(this) - if (y.connector._forwardAppliedStructs || this._id.user === y.userID) { - y.connector.broadcastStruct(this) + if (this._left === null) { + if (this._parentSub !== null) { + this._parent._map.set(this._parentSub, this) + } else { + this._parent._start = this + } } - if (y.persistence !== null) { - y.persistence.saveOperations(this) + y.os.put(this) + if (this._id.user !== RootFakeUserID) { + if (y.connector._forwardAppliedStructs || this._id.user === y.userID) { + y.connector.broadcastStruct(this) + } + if (y.persistence !== null) { + y.persistence.saveOperations(this) + } + y.ds.applyMissingDeletesOnStruct(this) } } - _toBinary (y, encoder) { - encoder.writeUint8(StructManager.getReference(this.constructor)) - encoder.writeOpID(this._id) - encoder.writeOpID(this._parent._id) - encoder.writeVarString(this.parentSub === null ? '' : JSON.stringify(this.parentSub)) - encoder.writeOpID(this._left === null ? null : this._left._id) - encoder.writeOpID(this._right_origin === null ? null : this._right_origin._id) - encoder.writeOpID(this._origin === null ? null : this._origin._id) + _toBinary (encoder) { + encoder.writeUint8(getReference(this.constructor)) + let info = 0 + if (this._origin !== null) { + info += 0b1 // origin is defined + } + if (this._left !== this._origin) { + info += 0b10 // do not copy origin to left + } + if (this._right_origin !== null) { + info += 0b100 + } + if (this._parentSub !== null) { + info += 0b1000 + } + encoder.writeUint8(info) + encoder.writeID(this._id) + if (info & 0b1) { + encoder.writeID(this._origin._id) + } + if (info & 0b10) { + encoder.writeID(this._left._id) + } + if (info & 0b100) { + encoder.writeID(this._right_origin._id) + } + if (~info & 0b101) { + // neither origin nor right is defined + encoder.writeID(this._parent._id) + } + if (info & 0b1000) { + encoder.writeVarString(JSON.stringify(this._parentSub)) + } } _fromBinary (y, decoder) { let missing = [] - this._id = decoder.readOpID() - let parent = decoder.readOpID() - let parentSub = decoder.readVarString() - if (parentSub.length > 0) { - this._parentSub = JSON.parse(parentSub) - } - let left = decoder.readOpID() - let right = decoder.readOpId() - let origin = decoder.readOpID() - if (parent !== null && this._parent === null) { - let _parent = y.os.get(parent) - if (_parent === null) { - missing.push(parent) - } else { - this._parent = _parent + const info = decoder.readUint8() + this._id = decoder.readID() + // read origin + if (info & 0b1) { + // origin != null + const originID = decoder.readID() + if (this._origin === null) { + const origin = y.os.getItemCleanEnd(originID) + if (origin === null) { + missing.push(originID) + } else { + this._origin = origin + } } } - if (origin !== null && this._origin === null) { - let _origin = y.os.getCleanStart(origin) - if (_origin === null) { - missing.push(origin) - } else { - this._origin = _origin + // read left + if (info & 0b10) { + // left !== origin + const leftID = decoder.readID() + if (this._left === null) { + const left = y.os.getItemCleanEnd(leftID) + if (left === null) { + // use origin instead + this._left = this._origin + } else { + this._left = left + } + } + } else { + this._left = this._origin + } + // read right + if (info & 0b100) { + // right != null + const rightID = decoder.readID() + if (this._right_origin === null) { + const right = y.os.getCleanStart(rightID) + if (right === null) { + missing.push(right) + } else { + this._right = right + this._right_origin = right + } } } - if (left !== null && this._left === null) { - let _left = y.os.getCleanEnd(left) - if (_left === null) { - // use origin instead - this._left = this._origin - } else { - this._left = _left + // read parent + if (~info & 0b101) { + // neither origin nor right is defined + const parentID = decoder.readID() + if (this._parent === null) { + const parent = y.os.get(parentID) + if (parent === null) { + missing.push(parent) + } else { + this._parent = parent + } + } + } else if (this._parent === null) { + if (this._origin !== null) { + this._parent = this._origin._parent + } else if (this._right_origin !== null) { + this._parent = this._origin._parent } } - if (right !== null && this._right_origin === null) { - let _right = y.os.getCleanStart(right) - if (_right === null) { - missing.push(right) - } else { - this._right = _right - this._right_origin = _right - } + if (info & 0b1000) { + this._parentSub = decoder.readVarString() } + return missing } _logString () { return `left: ${this._left}, origin: ${this._origin}, right: ${this._right}, parent: ${this._parent}, parentSub: ${this._parentSub}` diff --git a/src/Struct/ItemJSON.js b/src/Struct/ItemJSON.js index 91fe1f86..25f00cf6 100644 --- a/src/Struct/ItemJSON.js +++ b/src/Struct/ItemJSON.js @@ -1,4 +1,4 @@ -import Item from './Item.js' +import { splitHelper, default as Item } from './Item.js' export default class ItemJSON extends Item { constructor () { @@ -17,8 +17,8 @@ export default class ItemJSON extends Item { } return missing } - _toBinary (y, encoder) { - super._toBinary(y, encoder) + _toBinary (encoder) { + super._toBinary(encoder) let len = this._content.length encoder.writeVarUint(len) for (let i = 0; i < len; i++) { @@ -29,4 +29,15 @@ export default class ItemJSON extends Item { let s = super._logString() return 'ItemJSON: ' + s } + _splitAt (y, diff) { + if (diff === 0) { + return this + } else if (diff >= this._length) { + return this._right + } + let item = new ItemJSON() + item._content = this._content.splice(diff) + splitHelper(y, this, item, diff) + return item + } } diff --git a/src/Struct/ItemString.js b/src/Struct/ItemString.js index cfe5c2a5..bef915b6 100644 --- a/src/Struct/ItemString.js +++ b/src/Struct/ItemString.js @@ -1,4 +1,4 @@ -import Item from './Item.js' +import { splitHelper, default as Item } from './Item.js' export default class ItemString extends Item { constructor () { @@ -13,12 +13,24 @@ export default class ItemString extends Item { this._content = decoder.readVarString() return missing } - _toBinary (y, encoder) { - super._toBinary(y, encoder) + _toBinary (encoder) { + super._toBinary(encoder) encoder.writeVarString(this._content) } _logString () { let s = super._logString() return 'ItemString: ' + s } + _splitAt (y, diff) { + if (diff === 0) { + return this + } else if (diff >= this._length) { + return this._right + } + let item = new ItemString() + item._content = this._content.slice(diff) + this._content = this._content.slice(0, diff) + splitHelper(y, this, item, diff) + return item + } } diff --git a/src/Struct/Type.js b/src/Struct/Type.js index bd498bf1..e595d0ae 100644 --- a/src/Struct/Type.js +++ b/src/Struct/Type.js @@ -5,6 +5,11 @@ export default class Type extends Item { super() this._map = new Map() this._start = null + this._y = null + } + _integrate (y) { + super._integrate(y) + this._y = y } _delete (y) { super._delete(y) diff --git a/src/Type/YArray.js b/src/Type/YArray.js index 11f07d25..3c24ba25 100644 --- a/src/Type/YArray.js +++ b/src/Type/YArray.js @@ -2,21 +2,50 @@ import Type from '../Struct/Type.js' import ItemJSON from '../Struct/ItemJSON.js' export default class YArray extends Type { + toJSON () { + return this.map(c => { + if (c instanceof Type) { + if (c.toJSON !== null) { + return c.toJSON() + } else { + return c.toString() + } + } + }) + } + map (f) { + const res = [] + this.forEach((c, i) => { + res.push(f(c, i, this)) + }) + return res + } forEach (f) { let pos = 0 let n = this._start while (n !== null) { - let content = n._getContent() - for (let i = 0; i < content.length; i++) { - pos++ - let c = content[i] - if (!c._deleted) { + if (!n._deleted) { + const content = n._content + const contentLen = content.length + for (let i = 0; i < contentLen; i++) { + pos++ f(content[i], pos, this) } } n = n._right } } + get length () { + let length = 0 + let n = this._start + while (n !== null) { + if (!n._deleted) { + length += n._length + } + n = n._next + } + return length + } [Symbol.iterator] () { return { next: function () { @@ -41,15 +70,37 @@ export default class YArray extends Type { _count: 0 } } + delete (pos, length = 1) { + let item = this._start + let count = 0 + while (item !== null && length > 0) { + if (count < pos && pos < count + item._length) { + const diffDel = pos - count + item = item + ._splitAt(this._y, diffDel) + ._splitAt(this._y, length) + length -= item._length + item._delete(this._y) + } + if (!item._deleted) { + count += item._length + } + item = item._right + } + if (length > 0) { + throw new Error('Delete exceeds the range of the YArray') + } + } insert (pos, content) { let left = this._start - let right + let right = null let count = 0 - while (left !== null && !left._deleted) { - if (pos < count + left._content.length) { - [left, right] = left._splitAt(pos - count) + while (left !== null) { + if (count <= pos && pos < count + left._content.length) { + right = left._splitAt(this.y, pos - count) break } + count += left._length left = left.right } if (pos > count) { @@ -79,6 +130,9 @@ export default class YArray extends Type { prevJsonIns._content.push(c) } } + if (prevJsonIns !== null) { + prevJsonIns._integrate(this._y) + } } _logString () { let s = super._logString() diff --git a/src/Type/YMap.js b/src/Type/YMap.js index 819e9d98..58189352 100644 --- a/src/Type/YMap.js +++ b/src/Type/YMap.js @@ -3,6 +3,25 @@ import Item from '../Struct/Item.js' import ItemJSON from '../Struct/ItemJSON.js' export default class YMap extends Type { + toJSON () { + const map = {} + for (let [key, item] of this._map) { + if (!item._deleted) { + let res + if (item instanceof Type) { + if (item.toJSON !== undefined) { + res = item.toJSON() + } else { + res = item.toString() + } + } else { + res = item._content[0] + } + map[key] = res + } + } + return map + } set (key, value) { let old = this._map.get(key) let v diff --git a/src/Type/YXml.js b/src/Type/YXml.js index 3bf9ac0a..4e0bd83e 100644 --- a/src/Type/YXml.js +++ b/src/Type/YXml.js @@ -1,4 +1,10 @@ import YArray from './YArray.js' export default class YXml extends YArray { + setDomFilter () { + // TODO + } + toString () { + return '' + } } diff --git a/src/Util/ID.js b/src/Util/ID.js index bfc1b666..3c3e8235 100644 --- a/src/Util/ID.js +++ b/src/Util/ID.js @@ -1,6 +1,4 @@ -import { getReference } from './structReferences.js' - export default class ID { constructor (user, clock) { this.user = user diff --git a/src/Util/RootID.js b/src/Util/RootID.js index e3498e56..0968dced 100644 --- a/src/Util/RootID.js +++ b/src/Util/RootID.js @@ -1,9 +1,12 @@ +import { getReference } from './structReferences.js' + +export const RootFakeUserID = 0xFFFFFF export default class RootID { constructor (name, typeConstructor) { - this.user = -1 + this.user = RootFakeUserID this.name = name - this.type = StructManager.getReference(typeConstructor) + this.type = getReference(typeConstructor) } equals (id) { return id !== null && id.user === this.user && id.name === this.name && id.type === this.type diff --git a/src/Util/Tree.js b/src/Util/Tree.js index 0680bf77..9960443e 100644 --- a/src/Util/Tree.js +++ b/src/Util/Tree.js @@ -140,11 +140,11 @@ export default class Tree { return null } else { while (true) { - if (from === null || (from.lessThan(o.val.id) && o.left !== null)) { + if (from === null || (from.lessThan(o.val._id) && o.left !== null)) { // o is included in the bound // try to find an element that is closer to the bound o = o.left - } else if (from !== null && o.val.id.lessThan(from)) { + } else if (from !== null && o.val._id.lessThan(from)) { // o is not within the bound, maybe one of the right elements is.. if (o.right !== null) { o = o.right @@ -168,11 +168,11 @@ export default class Tree { return null } else { while (true) { - if ((to === null || o.val.id.lessThan(to)) && o.right !== null) { + if ((to === null || o.val._id.lessThan(to)) && o.right !== null) { // o is included in the bound // try to find an element that is closer to the bound o = o.right - } else if (to !== null && to.lessThan(o.val.id)) { + } else if (to !== null && to.lessThan(o.val._id)) { // o is not within the bound, maybe one of the left elements is.. if (o.left !== null) { o = o.left @@ -213,8 +213,8 @@ export default class Tree { o !== null && ( to === null || // eslint-disable-line no-unmodified-loop-condition - o.val.id.lessThan(to) || - o.val.id.equals(to) + o.val._id.lessThan(to) || + o.val._id.equals(to) ) ) { f(o.val) @@ -238,9 +238,9 @@ export default class Tree { if (o === null) { return null } - if (id.lessThan(o.val.id)) { + if (id.lessThan(o.val._id)) { o = o.left - } else if (o.val.id.lessThan(id)) { + } else if (o.val._id.lessThan(id)) { o = o.right } else { return o @@ -393,14 +393,14 @@ export default class Tree { if (this.root !== null) { var p = this.root // p abbrev. parent while (true) { - if (node.val.id.lessThan(p.val.id)) { + if (node.val._id.lessThan(p.val._id)) { if (p.left === null) { p.left = node break } else { p = p.left } - } else if (p.val.id.lessThan(node.val.id)) { + } else if (p.val._id.lessThan(node.val._id)) { if (p.right === null) { p.right = node break diff --git a/src/Y.js b/src/Y.js index 9c33de87..fd013beb 100644 --- a/src/Y.js +++ b/src/Y.js @@ -3,6 +3,7 @@ import OperationStore from './Store/OperationStore.js' import StateStore from './Store/StateStore.js' import { generateUserID } from './Util/generateUserID.js' import RootID from './Util/RootID.js' +import NamedEventHandler from './Util/NamedEventHandler.js' import { messageToString, messageToRoomname } from './MessageHandler/messageToString.js' @@ -15,8 +16,10 @@ import YXml from './Type/YXml.js' import debug from 'debug' -export default class Y { +export default class Y extends NamedEventHandler { constructor (opts) { + super() + this._opts = opts this.userID = generateUserID() this.ds = new DeleteStore(this) this.os = new OperationStore(this) @@ -30,10 +33,17 @@ export default class Y { } this.connected = true this._missingStructs = new Map() - this._readyToIntegrate = new Map() + this._readyToIntegrate = [] + } + // fake _start for root properties (y.set('name', type)) + get _start () { + return null + } + set _start (start) { + return null } get room () { - return this.connector.opts.room + return this._opts.connector.room } get (name, TypeConstructor) { let id = new RootID(name, TypeConstructor) @@ -41,6 +51,7 @@ export default class Y { if (type === null) { type = new TypeConstructor() type._id = id + type._parent = this type._integrate(this) } return type @@ -68,9 +79,6 @@ export default class Y { } else { this.connector.disconnect() } - this.os.iterate(null, null, function (struct) { - struct.destroy() - }) this.os = null this.ds = null this.ss = null @@ -88,7 +96,8 @@ Y.extend = function extendYjs () { } } -Y.Connector = Connector +// TODO: The following assignments should be moved to yjs-dist +Y.AbstractConnector = Connector Y.Persisence = Persistence Y.Array = YArray Y.Map = YMap diff --git a/test/red-black-tree.js b/test/red-black-tree.js index 299e85f0..63cb6f08 100644 --- a/test/red-black-tree.js +++ b/test/red-black-tree.js @@ -55,15 +55,15 @@ function checkRootNodeIsBlack (t, tree) { test('RedBlack Tree', async function redBlackTree (t) { let tree = new RedBlackTree() - tree.put({id: new ID(8433, 0)}) - tree.put({id: new ID(12844, 0)}) - tree.put({id: new ID(1795, 0)}) - tree.put({id: new ID(30302, 0)}) - tree.put({id: new ID(64287)}) + tree.put({_id: new ID(8433, 0)}) + tree.put({_id: new ID(12844, 0)}) + tree.put({_id: new ID(1795, 0)}) + tree.put({_id: new ID(30302, 0)}) + tree.put({_id: new ID(64287)}) tree.delete(new ID(8433, 0)) - tree.put({id: new ID(28996)}) + tree.put({_id: new ID(28996)}) tree.delete(new ID(64287)) - tree.put({id: new ID(22721)}) + tree.put({_id: new ID(22721)}) checkRootNodeIsBlack(t, tree) checkBlackHeightOfSubTreesAreEqual(t, tree) checkRedNodesDoNotHaveBlackChildren(t, tree) @@ -83,7 +83,7 @@ test(`random tests (${numberOfRBTreeTests})`, async function random (t) { t.assert(false, 'tree and elements contain different results') } elements.push(obj) - tree.put({id: obj}) + tree.put({_id: obj}) } } else if (elements.length > 0) { // ~20% chance to delete an element @@ -101,7 +101,7 @@ test(`random tests (${numberOfRBTreeTests})`, async function random (t) { let allNodesExist = true for (let id of elements) { let node = tree.find(id) - if (!node.id.equals(id)) { + if (!node._id.equals(id)) { allNodesExist = false } } @@ -110,7 +110,7 @@ test(`random tests (${numberOfRBTreeTests})`, async function random (t) { let findAllNodesWithLowerBoundSerach = true for (let id of elements) { let node = tree.findWithLowerBound(id) - if (!node.id.equals(id)) { + if (!node._id.equals(id)) { findAllNodesWithLowerBoundSerach = false } } diff --git a/test/y-array.tests.js b/test/y-array.tests.js index e5bf0561..69c15a11 100644 --- a/test/y-array.tests.js +++ b/test/y-array.tests.js @@ -11,11 +11,11 @@ test('basic spec', async function array0 (t) { let throwInvalidPosition = false try { - array0.delete(1, 0) + array0.delete(1, 1) } catch (e) { throwInvalidPosition = true } - t.assert(throwInvalidPosition, 'Throws when deleting zero elements with an invalid position') + t.assert(throwInvalidPosition, 'Throws when deleting with an invalid position') array0.insert(0, ['A']) array0.delete(1, 0) diff --git a/tests-lib/helper.js b/tests-lib/helper.js index ca4ceb73..f6c87145 100644 --- a/tests-lib/helper.js +++ b/tests-lib/helper.js @@ -87,7 +87,7 @@ export async function compareUsers (t, users) { var userArrayValues = users.map(u => u.get('array', Y.Array).toJSON()) var userMapValues = users.map(u => u.get('map', Y.Map).toJSON()) - var userXmlValues = users.map(u => u.get('xml', Y.Xml).getDom().toString()) + var userXmlValues = users.map(u => u.get('xml', Y.Xml).toString()) // disconnect all except user 0 await Promise.all(users.slice(1).map(async u => @@ -107,28 +107,22 @@ export async function compareUsers (t, users) { u.connector.whenSynced(resolve) }) )) - var data = users.forEach(u => { + var data = users.map(u => { var data = {} let ops = [] u.os.iterate(null, null, function (op) { if (!op._deleted) { ops.push({ + id: op._id, left: op._left, right: op._right, deleted: op._deleted }) } }) - - data.os = {} - for (let i = 0; i < ops.length; i++) { - let op = ops[i] - op = Y.Struct[op.struct].encode(op) - delete op.origin - data.os[JSON.stringify(op.id)] = op - } - data.ds = getDeleteSet.apply(this) - data.ss = getStateSet.apply(this) + data.os = ops + data.ds = getDeleteSet(u) + data.ss = getStateSet(u) return data }) for (var i = 0; i < data.length - 1; i++) { @@ -141,7 +135,7 @@ export async function compareUsers (t, users) { t.compare(data[i].ss, data[i + 1].ss, 'ss') }, `Compare user${i} with user${i + 1}`) } - users.map(u => u.close()) + users.map(u => u.destroy()) } export async function initArrays (t, opts) { @@ -161,6 +155,7 @@ export async function initArrays (t, opts) { connector: connOpts }) result.users.push(y) + result['array' + i] = y.get('array', Y.Array) y.get('xml', Y.Xml).setDomFilter(function (d, attrs) { if (d.nodeName === 'HIDDEN') { return null @@ -193,9 +188,6 @@ export async function flushAll (t, users) { // use flushAll method specified in Test Connector await users[0].connector.testRoom.flushAll(users) } else { - // flush for any connector - await Promise.all(users.map(u => { return u.db.whenTransactionsFinished() })) - var flushCounter = users[0].get('flushHelper', Y.Map).get('0') || 0 flushCounter++ await Promise.all(users.map(async (u, i) => { diff --git a/tests-lib/test-connector.js b/tests-lib/test-connector.js index 07a7f92b..f20d3654 100644 --- a/tests-lib/test-connector.js +++ b/tests-lib/test-connector.js @@ -8,24 +8,21 @@ export class TestRoom { constructor (roomname) { this.room = roomname this.users = new Map() - this.nextUserId = 0 } join (connector) { - if (connector.userId == null) { - connector.setUserId(this.nextUserId++) - } - this.users.forEach((user, uid) => { - if (user.role === 'master' || connector.role === 'master') { - this.users.get(uid).userJoined(connector.userId, connector.role) + const userID = connector.y.userID + this.users.set(userID, connector) + for (let [uid, user] of this.users) { + if (uid !== userID && (user.role === 'master' || connector.role === 'master')) { connector.userJoined(uid, this.users.get(uid).role) + this.users.get(uid).userJoined(userID, connector.role) } - }) - this.users.set(connector.userId, connector) + } } leave (connector) { - this.users.delete(connector.userId) + this.users.delete(connector.y.userID) this.users.forEach(user => { - user.userLeft(connector.userId) + user.userLeft(connector.y.userID) }) } send (sender, receiver, m) { @@ -82,7 +79,7 @@ export default function extendTestConnector (Y) { return super.disconnect() } logBufferParsed () { - console.log(' === Logging buffer of user ' + this.userId + ' === ') + console.log(' === Logging buffer of user ' + this.y.userID + ' === ') for (let [user, conn] of this.connections) { console.log(` ${user}:`) for (let i = 0; i < conn.buffer.length; i++) { @@ -96,11 +93,11 @@ export default function extendTestConnector (Y) { } send (uid, message) { super.send(uid, message) - this.testRoom.send(this.userId, uid, message) + this.testRoom.send(this.y.userID, uid, message) } broadcast (message) { super.broadcast(message) - this.testRoom.broadcast(this.userId, message) + this.testRoom.broadcast(this.y.userID, message) } async whenSynced (f) { var synced = false @@ -119,7 +116,7 @@ export default function extendTestConnector (Y) { }) } receiveMessage (sender, m) { - if (this.userId !== sender && this.connections.has(sender)) { + if (this.y.userID !== sender && this.connections.has(sender)) { var buffer = this.connections.get(sender).buffer if (buffer == null) { buffer = this.connections.get(sender).buffer = [] @@ -135,30 +132,30 @@ export default function extendTestConnector (Y) { } } async _flushAll (flushUsers) { - if (flushUsers.some(u => u.connector.userId === this.userId)) { + if (flushUsers.some(u => u.connector.y.userID === this.y.userID)) { // this one needs to sync with every other user flushUsers = Array.from(this.connections.keys()).map(uid => this.testRoom.users.get(uid).y) } var finished = [] for (let i = 0; i < flushUsers.length; i++) { - let userId = flushUsers[i].connector.userId - if (userId !== this.userId && this.connections.has(userId)) { - let buffer = this.connections.get(userId).buffer + let userID = flushUsers[i].connector.y.userID + if (userID !== this.y.userID && this.connections.has(userID)) { + let buffer = this.connections.get(userID).buffer if (buffer != null) { var messages = buffer.splice(0) for (let j = 0; j < messages.length; j++) { - let p = super.receiveMessage(userId, messages[j]) + let p = super.receiveMessage(userID, messages[j]) finished.push(p) } } } } await Promise.all(finished) - await this.y.db.whenTransactionsFinished() return finished.length > 0 ? 'flushing' : 'done' } } - Y.extend('test', TestConnector) + // TODO: this should be moved to a separate module (dont work on Y) + Y.test = TestConnector } if (typeof Y !== 'undefined') {