From bd9c3813fd1512ea51d7424e56d67cd156066128 Mon Sep 17 00:00:00 2001
From: Kevin Jahns <kevin.jahns@rwth-aachen.de>
Date: Thu, 26 Nov 2015 00:46:02 +0100
Subject: [PATCH] * starting flow integration * found a bug in EventHelper *
reduce wait() calls
---
.flowconfig | 12 ++++++
.vscode/settings.json | 2 +
declarations/Structs.js | 26 ++++++++++++
declarations/Y.js | 17 ++++++++
package.json | 1 +
src/Connector.js | 2 +-
src/Connectors/Test.js | 22 ++++------
src/Database.js | 94 ++++++++++++++++++++++++++++++-----------
src/SpecHelper.js | 7 +--
src/Types/Map.js | 4 +-
src/Utils.js | 84 +++++++++++++++++++++++-------------
src/y.js | 4 +-
12 files changed, 200 insertions(+), 75 deletions(-)
create mode 100644 .flowconfig
create mode 100644 declarations/Structs.js
create mode 100644 declarations/Y.js
diff --git a/.flowconfig b/.flowconfig
new file mode 100644
index 00000000..54306bd6
--- /dev/null
+++ b/.flowconfig
@@ -0,0 +1,12 @@
+[ignore]
+.*/node_modules/.*
+.*/dist/.*
+.*/build/.*
+
+[include]
+./src/
+
+[libs]
+./declarations/
+
+[options]
diff --git a/.vscode/settings.json b/.vscode/settings.json
index 20af2f68..503cfe00 100644
--- a/.vscode/settings.json
+++ b/.vscode/settings.json
@@ -1,3 +1,5 @@
// Place your settings in this file to overwrite default and user settings.
{
+ "standard.enable": true,
+ "javascript.validate.enable": false
}
\ No newline at end of file
diff --git a/declarations/Structs.js b/declarations/Structs.js
new file mode 100644
index 00000000..58001bbc
--- /dev/null
+++ b/declarations/Structs.js
@@ -0,0 +1,26 @@
+/* @flow */
+
+type UserId = string
+type Id = [UserId, number]
+
+/*
+type Struct = {
+ id: Id,
+ left?: Id,
+ right?: Id,
+ target?: Id,
+ struct: 'Insert' | 'Delete'
+}*/
+type Struct = Insertion | Deletion
+
+type Insertion = {
+ id: Id,
+ left: Id,
+ right: Id,
+ struct: 'Insert'
+}
+
+type Deletion = {
+ target: Id,
+ struct: 'Delete'
+}
\ No newline at end of file
diff --git a/declarations/Y.js b/declarations/Y.js
new file mode 100644
index 00000000..e287be99
--- /dev/null
+++ b/declarations/Y.js
@@ -0,0 +1,17 @@
+/* @flow */
+
+type YGlobal = {
+ utils: Object;
+ Struct: Object;
+ AbstractDatabase: any;
+}
+
+type YInstance = {
+ db: Object,
+ connector: Object,
+ root: Object
+}
+
+declare var YConcurrency_TestingMode : boolean
+
+type Transaction<A> = Generator<any, A, any>
\ No newline at end of file
diff --git a/package.json b/package.json
index 96f53cb8..a3ffc144 100644
--- a/package.json
+++ b/package.json
@@ -16,6 +16,7 @@
"ignore": [
"build/**",
"dist/**",
+ "declarations/**",
"./y.js",
"./y.js.map"
]
diff --git a/src/Connector.js b/src/Connector.js
index 9ff9db4f..66b6a9d8 100644
--- a/src/Connector.js
+++ b/src/Connector.js
@@ -190,7 +190,7 @@ module.exports = function (Y) {
this.broadcastedHB = true
var db = this.y.db
var defer = Promise.defer()
- this.syncStep2 = defer.promise
+ this.syncStep2 = defer.promise
db.requestTransaction(function * () {
yield* this.applyDeleteSet(m.deleteSet)
this.store.apply(m.os)
diff --git a/src/Connectors/Test.js b/src/Connectors/Test.js
index 97d7329a..2f95a0d7 100644
--- a/src/Connectors/Test.js
+++ b/src/Connectors/Test.js
@@ -24,16 +24,11 @@ module.exports = function (Y) {
}
},
whenTransactionsFinished: function () {
- var self = this
- return new Promise (function (resolve) {
- wait().then(function () {
- var ps = []
- for (var name in self.users) {
- ps.push(self.users[name].y.db.whenTransactionsFinished())
- }
- Promise.all(ps).then(resolve)
- })
- })
+ var ps = []
+ for (var name in this.users) {
+ ps.push(this.users[name].y.db.whenTransactionsFinished())
+ }
+ return Promise.all(ps)
},
flushOne: function flushOne () {
var bufs = []
@@ -59,15 +54,16 @@ module.exports = function (Y) {
function nextFlush () {
var c = globalRoom.flushOne()
if (c) {
- while (c = globalRoom.flushOne()) {
+ while (c) {
+ c = globalRoom.flushOne()
}
- globalRoom.whenTransactionsFinished().then(nextFlush)
+ globalRoom.whenTransactionsFinished().then(nextFlush)
} else {
setTimeout(function () {
var c = globalRoom.flushOne()
if (c) {
c.then(function () {
- globalRoom.whenTransactionsFinished().then(nextFlush)
+ globalRoom.whenTransactionsFinished().then(nextFlush)
})
} else {
resolve()
diff --git a/src/Database.js b/src/Database.js
index 9977b589..0dbd427c 100644
--- a/src/Database.js
+++ b/src/Database.js
@@ -1,6 +1,7 @@
+/* @flow */
'use strict'
-module.exports = function (Y) {
+module.exports = function (Y /* : YGlobal */) {
/*
Partial definition of an OperationStore.
TODO: name it Database, operation store only holds operations.
@@ -14,6 +15,28 @@ module.exports = function (Y) {
- destroy the database
*/
class AbstractDatabase {
+ /* ::
+ y: YInstance;
+ forwardAppliedOperations: boolean;
+ listenersById: Object;
+ listenersByIdExecuteNow: Array<Object>;
+ listenersByIdRequestPending: boolean;
+ initializedTypes: Object;
+ whenUserIdSetListener: ?Function;
+ waitingTransactions: Array<Transaction>;
+ transactionInProgress: boolean;
+ executeOrder: Array<Object>;
+ gc1: Array<Struct>;
+ gc2: Array<Struct>;
+ gcTimeout: number;
+ gcInterval: any;
+ garbageCollect: Function;
+ executeOrder: Array<any>; // for debugging only
+ userId: UserId;
+ opClock: number;
+ transactionsFinished: ?{promise: Promise, resolve: any};
+ transact: (x: ?Generator) => any;
+ */
constructor (y, opts) {
this.y = y
// whether to broadcast all applied operations (insert & delete hook)
@@ -51,7 +74,7 @@ module.exports = function (Y) {
return new Promise((resolve) => {
os.requestTransaction(function * () {
if (os.y.connector != null && os.y.connector.isSynced) {
- for (var i in os.gc2) {
+ for (var i = 0; i < os.gc2.length; i++) {
var oid = os.gc2[i]
yield* this.garbageCollectOperation(oid)
}
@@ -72,7 +95,7 @@ module.exports = function (Y) {
}
addToDebug () {
if (typeof YConcurrency_TestingMode !== 'undefined') {
- var command = Array.prototype.map.call(arguments, function (s) {
+ var command /* :string */ = Array.prototype.map.call(arguments, function (s) {
if (typeof s === 'string') {
return s
} else {
@@ -89,10 +112,10 @@ module.exports = function (Y) {
var self = this
return new Promise(function (resolve) {
self.requestTransaction(function * () {
- var ungc = self.gc1.concat(self.gc2)
+ var ungc /* :Array<Struct> */ = self.gc1.concat(self.gc2)
self.gc1 = []
self.gc2 = []
- for (var i in ungc) {
+ for (var i = 0; i < ungc.length; i++) {
var op = yield* this.getOperation(ungc[i])
delete op.gc
yield* this.setOperation(op)
@@ -145,7 +168,8 @@ module.exports = function (Y) {
return new Promise(function (resolve) {
self.requestTransaction(function * () {
self.userId = userId
- self.opClock = (yield* this.getState(userId)).clock
+ var state = yield* this.getState(userId)
+ self.opClock = state.clock
if (self.whenUserIdSetListener != null) {
self.whenUserIdSetListener()
self.whenUserIdSetListener = null
@@ -225,7 +249,7 @@ module.exports = function (Y) {
store.listenersByIdRequestPending = false
- for (let key in exeNow) {
+ for (let key = 0; key < exeNow.length; key++) {
let o = exeNow[key].op
yield* store.tryExecute.call(this, o)
}
@@ -233,7 +257,8 @@ module.exports = function (Y) {
for (var sid in ls) {
var l = ls[sid]
var id = JSON.parse(sid)
- if ((yield* this.getOperation(id)) == null) {
+ var op = yield* this.getOperation(id)
+ if (op == null) {
store.listenersById[sid] = l
} else {
for (let key in l) {
@@ -250,15 +275,28 @@ module.exports = function (Y) {
/*
Actually execute an operation, when all expected operations are available.
*/
+ /* :: // TODO: this belongs somehow to transaction
+ store: Object;
+ getOperation: any;
+ isGarbageCollected: any;
+ addOperation: any;
+ whenOperationsExist: any;
+ */
* tryExecute (op) {
this.store.addToDebug('yield* this.store.tryExecute.call(this, ', JSON.stringify(op), ')')
if (op.struct === 'Delete') {
yield* Y.Struct.Delete.execute.call(this, op)
yield* this.store.operationAdded(this, op)
- } else if ((yield* this.getOperation(op.id)) == null && !(yield* this.isGarbageCollected(op.id))) {
- yield* Y.Struct[op.struct].execute.call(this, op)
- yield* this.addOperation(op)
- yield* this.store.operationAdded(this, op)
+ } else {
+ var defined = yield* this.getOperation(op.id)
+ if (defined == null) {
+ var isGarbageCollected = yield* this.isGarbageCollected(op.id)
+ if (!isGarbageCollected) {
+ yield* Y.Struct[op.struct].execute.call(this, op)
+ yield* this.addOperation(op)
+ yield* this.store.operationAdded(this, op)
+ }
+ }
}
}
// called by a transaction when an operation is added
@@ -300,30 +338,38 @@ module.exports = function (Y) {
}
}
var t = this.initializedTypes[JSON.stringify(op.parent)]
- // notify parent, if it has been initialized as a custom type
- if (t != null) {
- yield* t._changed(transaction, Y.utils.copyObject(op))
- }
// Delete if DS says this is actually deleted
- if (!op.deleted && (yield* transaction.isDeleted(op.id))) {
+ var opIsDeleted = yield* transaction.isDeleted(op.id)
+ if (!op.deleted && opIsDeleted) {
var delop = {
struct: 'Delete',
target: op.id
}
yield* Y.Struct['Delete'].execute.call(transaction, delop)
- if (t != null) {
- yield* t._changed(transaction, delop)
- }
+ }
+
+ // notify parent, if it has been initialized as a custom type
+ if (t != null) {
+ yield* t._changed(transaction, Y.utils.copyObject(op))
}
}
}
whenTransactionsFinished () {
if (this.transactionInProgress) {
if (this.transactionsFinished == null) {
- this.transactionsFinished = Promise.defer()
+ var resolve
+ var promise = new Promise(function (r) {
+ resolve = r
+ })
+ this.transactionsFinished = {
+ resolve: resolve,
+ promise: promise
+ }
+ return promise
+ } else {
+ return this.transactionsFinished.promise
}
- return this.transactionsFinished.promise
} else {
return Promise.resolve()
}
@@ -340,8 +386,8 @@ module.exports = function (Y) {
return this.waitingTransactions.shift()
}
}
- requestTransaction (makeGen, callImmediately) {
- if (callImmediately || true) {
+ requestTransaction (makeGen/* :any */, callImmediately) {
+ if (callImmediately) {
this.waitingTransactions.push(makeGen)
if (!this.transactionInProgress) {
this.transactionInProgress = true
diff --git a/src/SpecHelper.js b/src/SpecHelper.js
index 5a80de99..da92a2c3 100644
--- a/src/SpecHelper.js
+++ b/src/SpecHelper.js
@@ -37,12 +37,12 @@ g.describeManyTimes = function describeManyTimes (times, name, f) {
*/
function wait (t) {
if (t == null) {
- t = 5
+ t = 0
}
return new Promise(function (resolve) {
setTimeout(function () {
resolve()
- }, t * 3)
+ }, t)
})
}
g.wait = wait
@@ -162,7 +162,8 @@ g.compareAllUsers = async(function * compareAllUsers (users) {
var buffer = Y.utils.globalRoom.buffers
for (var name in buffer) {
if (buffer[name].length > 0) {
- debugger // not all ops were transmitted..
+ // not all ops were transmitted..
+ debugger // eslint-disable-line
}
}
diff --git a/src/Types/Map.js b/src/Types/Map.js
index efca38d9..4cd4a4b1 100644
--- a/src/Types/Map.js
+++ b/src/Types/Map.js
@@ -157,7 +157,6 @@ module.exports = function (Y) {
insert.id = this.os.getNextOpId()
var eventHandler = this.eventHandler
eventHandler.awaitAndPrematurelyCall([insert])
-
this.os.requestTransaction(function *() {
yield* this.applyCreatedOperations([insert])
eventHandler.awaitedInserts(1)
@@ -258,7 +257,8 @@ module.exports = function (Y) {
}
* _changed (transaction, op) {
if (op.struct === 'Delete') {
- op.key = (yield* transaction.getOperation(op.target)).parentSub
+ var target = yield* transaction.getOperation(op.target)
+ op.key = target.parentSub
}
this.eventHandler.receivedOp(op)
}
diff --git a/src/Utils.js b/src/Utils.js
index 513fafc3..018876d7 100644
--- a/src/Utils.js
+++ b/src/Utils.js
@@ -1,3 +1,4 @@
+/* @flow */
'use strict'
/*
@@ -6,24 +7,32 @@
Why: When constructing custom types, you sometimes want your types to work
synchronous: E.g.
``` Synchronous
- mytype.setSomething("yay")
- mytype.getSomething() === "yay"
- ```
- ``` Asynchronous
- mytype.setSomething("yay")
- mytype.getSomething() === undefined
- mytype.waitForSomething().then(function(){
+ mytype.setSomething("yay")
mytype.getSomething() === "yay"
- })
+ ```
+ versus
+ ``` Asynchronous
+ mytype.setSomething("yay")
+ mytype.getSomething() === undefined
+ mytype.waitForSomething().then(function(){
+ mytype.getSomething() === "yay"
+ })
+ ```
The structures usually work asynchronously (you have to wait for the
database request to finish). EventHandler will help you to make your type
- synchronously.
+ synchronous.
*/
-module.exports = function (Y) {
+module.exports = function (Y /* : YGlobal*/) {
Y.utils = {}
class EventHandler {
+ /* ::
+ waiting: Array<Insertion | Deletion>;
+ awaiting: number;
+ onevent: Function;
+ eventListeners: Array<Function>;
+ */
/*
onevent: is called when the structure changes.
@@ -31,7 +40,7 @@ module.exports = function (Y) {
prematurely called. Events for received operations can not be executed until
all prematurely called operations were executed ("waiting operations")
*/
- constructor (onevent) {
+ constructor (onevent /* : Function */) {
this.waiting = []
this.awaiting = 0
this.onevent = onevent
@@ -73,7 +82,7 @@ module.exports = function (Y) {
this.eventListeners = []
}
callEventListeners (event) {
- for (var i in this.eventListeners) {
+ for (var i = 0; i < this.eventListeners.length; i++) {
try {
this.eventListeners[i](event)
} catch (e) {
@@ -88,18 +97,24 @@ module.exports = function (Y) {
var ops = this.waiting.splice(this.waiting.length - n)
for (var oid = 0; oid < ops.length; oid++) {
var op = ops[oid]
- for (var i = this.waiting.length - 1; i >= 0; i--) {
- let w = this.waiting[i]
- if (Y.utils.compareIds(op.left, w.id)) {
- // include the effect of op in w
- w.right = op.id
- // exclude the effect of w in op
- op.left = w.left
- } else if (Y.utils.compareIds(op.right, w.id)) {
- // similar..
- w.left = op.id
- op.right = w.right
+ if (op.struct === 'Insert') {
+ for (var i = this.waiting.length - 1; i >= 0; i--) {
+ let w = this.waiting[i]
+ if (w.struct === 'Insert') {
+ if (Y.utils.compareIds(op.left, w.id)) {
+ // include the effect of op in w
+ w.right = op.id
+ // exclude the effect of w in op
+ op.left = w.left
+ } else if (Y.utils.compareIds(op.right, w.id)) {
+ // similar..
+ w.left = op.id
+ op.right = w.right
+ }
+ }
}
+ } else {
+ throw new Error('Expected Insert Operation!')
}
}
this._tryCallEvents()
@@ -109,16 +124,20 @@ module.exports = function (Y) {
*/
awaitedDeletes (n, newLeft) {
var ops = this.waiting.splice(this.waiting.length - n)
- for (var j in ops) {
+ for (var j = 0; j < ops.length; j++) {
var del = ops[j]
- if (newLeft != null) {
- for (var i in this.waiting) {
- let w = this.waiting[i]
- // We will just care about w.left
- if (Y.utils.compareIds(del.target, w.left)) {
- del.left = newLeft
+ if (del.struct === 'Delete') {
+ if (newLeft != null) {
+ for (var i = 0; i < this.waiting.length; i++) {
+ let w = this.waiting[i]
+ // We will just care about w.left
+ if (w.struct === 'Insert' && Y.utils.compareIds(del.target, w.left)) {
+ w.left = newLeft
+ }
}
}
+ } else {
+ throw new Error('Expected Delete Operation!')
}
}
this._tryCallEvents()
@@ -149,6 +168,11 @@ module.exports = function (Y) {
- the constructor of the custom type (e.g. in order to inherit from a type)
*/
class CustomType { // eslint-disable-line
+ /* ::
+ createType: any;
+ initType: any;
+ class: Function;
+ */
constructor (def) {
if (def.createType == null ||
def.initType == null ||
diff --git a/src/y.js b/src/y.js
index 91927475..df34586c 100644
--- a/src/y.js
+++ b/src/y.js
@@ -1,4 +1,3 @@
-/* @flow */
'use strict'
require('./Connector.js')(Y)
@@ -20,7 +19,8 @@ Y.extend = function (name, value) {
}
}
-Y.requestModules = function (modules) {
+Y.requestModules = requestModules
+function requestModules (modules) {
var promises = []
for (var i = 0; i < modules.length; i++) {
var modulename = 'y-' + modules[i].toLowerCase()