github/meteor
1
1
Fork 0
mirror of https://github.com/meteor/meteor.git synced 2026-05-02 03:01:46 -04:00
Code Issues Packages Projects Releases Wiki Activity

radical radical Deps (needs more tests & docs)

Browse Source
This commit is contained in:
David Greenspan
2013-03-07 17:44:07 -08:00
parent b38ec5513d
commit 334f172cac
9 changed files with 125 additions and 189 deletions
Download Patch File Download Diff File Expand all files Collapse all files

193
packages/deps/deps.js
View File

@@ -3,8 +3,13 @@
Deps.active = false;
Deps.currentComputation = null;
var setCurrentComputation = function (c) {
Deps.currentComputation = c;
Deps.active = !! c;
};
var _debugFunc = function () {
// evaluate this lazily in order to not constrain load order
// lazy evaluation because `Meteor` does not exist right away
return (typeof Meteor !== "undefined" ? Meteor._debug :
((typeof console !== "undefined") && console.log ? console.log :
function () {}));
@@ -18,6 +23,8 @@
// `true` if we are in Deps.flush now
var inFlush = false;
var afterFlushCallbacks = [];
var requireFlush = function () {
if (! willFlush) {
setTimeout(Deps.flush, 0);
@@ -38,18 +45,15 @@
var self = this;
self.stopped = false;
self.invalidated = false;
self.active = false;
self.firstRun = true;
self._id = nextId++;
self._callbacks = {
onInvalidate: [],
afterInvalidate: []
};
self._onInvalidateCallbacks = [];
// the plan is at some point to use the parent relation
// to constrain the order that computations are processed
self._parent = parent;
self._func = (f || function () {});
self._processing = false;
try {
self._run();
@@ -61,104 +65,74 @@
_.extend(Deps.Computation.prototype, {
onInvalidate: function (f) {
if (! this.active)
throw new Error(
"Can only register callbacks on an active Computation");
var self = this;
this._callbacks.onInvalidate.push(f);
},
if (typeof f !== 'function')
throw new Error("onInvalidate requires a function");
afterInvalidate: function (f) {
if (! this.active)
throw new Error(
"Can only register callbacks on an active Computation");
var g = function () {
Deps.nonreactive(function () {
f(self);
});
};
this._callbacks.afterInvalidate.push(f);
if (self.invalidated)
g();
else
self._onInvalidateCallbacks.push(g);
},
invalidate: function () {
if (! this.invalidated) {
if (! this.active)
// an active computation is either already being
// processed or, on first run, is enqueued at
// the end of _run
this._enqueue();
this.invalidated = true;
var self = this;
if (! self.invalidated) {
// If this computation is currently being rerun,
// don't enqueue it. It will be rerun again
// immediately.
if (! self._processing) {
requireFlush();
pendingComputations.push(this);
}
self.invalidated = true;
// callbacks can't add callbacks, because
// self.invalidated === true.
for(var i = 0, f; f = self._onInvalidateCallbacks[i]; i++)
f(); // already bound with self as argument
self._onInvalidateCallbacks = [];
}
},
stop: function () {
if (! this.stopped) {
this.invalidate();
this.stopped = true;
this.invalidate();
}
},
_enqueue: function () {
requireFlush();
pendingComputations.push(this);
},
_run: function () {
var self = this;
self.invalidated = false;
var previous = Deps.currentComputation;
Deps.currentComputation = self;
Deps.active = true;
self.active = true;
setCurrentComputation(self);
try {
self._func(self);
} finally {
self.active = false;
Deps.currentComputation = previous;
Deps.active = !! Deps.currentComputation;
setCurrentComputation(previous);
}
if (self.firstRun && self.invalidated)
self._enqueue();
},
_process: function () {
var self = this;
while (self.invalidated) {
var onInvalidateCallbacks = self._callbacks.onInvalidate;
self._callbacks.onInvalidate = [];
var afterInvalidateCallbacks = self._callbacks.afterInvalidate;
self._callbacks.afterInvalidate = [];
for(var i = 0, f; f = onInvalidateCallbacks[i]; i++) {
try {
f(self);
} catch (e) {
_debugFunc()("Exception from Deps invalidation callback:",
e.stack);
}
self._processing = true;
while (self.invalidated && ! self.stopped) {
try {
self._run();
} catch (e) {
_debugFunc()("Exception from Deps rerun:", e.stack);
}
var wasStopped = self.stopped;
if (! wasStopped) {
try {
self._run();
} catch (e) {
_debugFunc()("Exception from Deps rerun:", e.stack);
}
}
for(var i = 0, f; f = afterInvalidateCallbacks[i]; i++) {
try {
f(self);
} catch (e) {
_debugFunc()("Exception from Deps invalidation callback:",
e.stack);
}
}
if (wasStopped)
break;
// If _run() stopped us, we loop around to handle callbacks.
// If _run() invalidated us, we run again immediately.
// A computation that invalidates itself indefinitely is an
// infinite loop, of course.
@@ -166,6 +140,7 @@
// We could put an iteration counter here and catch run-away
// loops.
}
self._processing = false;
}
});
@@ -220,32 +195,37 @@
// any useful notion of a nested flush.
//
// https://app.asana.com/0/159908330244/385138233856
if (inFlush) {
// note: consider removing this warning if it comes up
// in legit uses of flush and is annoying.
_debugFunc()("Warning: Ignored nested Deps.flush:",
(new Error).stack);
return;
}
if (inFlush)
throw new Error("Can't call Deps.flush while flushing");
if (Deps.active)
throw new Error("Can't flush inside Deps.run");
inFlush = true;
willFlush = true;
// It's possible for Computations to be active,
// if we are in an enclosing Deps.run in its
// first run (i.e. not called from flush).
// Keep one from being currentComputation.
Deps.nonreactive(function () {
while (pendingComputations.length ||
afterFlushCallbacks.length) {
while (pendingComputations.length) {
var comps = pendingComputations;
pendingComputations = [];
// rerun all pending computations
var comps = pendingComputations;
pendingComputations = [];
for (var i = 0, comp; comp = comps[i]; i++)
comp._process();
for (var i = 0, comp; comp = comps[i]; i++)
comp._process();
if (afterFlushCallbacks.length) {
// call one afterFlush callback, which may
// invalidate more computations
var func = afterFlushCallbacks.shift();
try {
func();
} catch (e) {
_debugFunc()("Exception from Deps afterFlush function:",
e.stack);
}
}
});
}
inFlush = false;
willFlush = false;
@@ -276,30 +256,21 @@
// computations being invalidated.
nonreactive: function (f) {
var previous = Deps.currentComputation;
Deps.currentComputation = null;
Deps.active = false;
setCurrentComputation(null);
try {
return f();
} finally {
Deps.currentComputation = previous;
Deps.active = !! Deps.currentComputation;
setCurrentComputation(previous);
}
},
onInvalidate: function (f) {
if (! Deps.active)
throw new Error("Deps.onInvalidate needs a currentComputation");
throw new Error("Deps.onInvalidate requires a currentComputation");
Deps.currentComputation.onInvalidate(f);
},
afterInvalidate: function (f) {
if (! Deps.active)
throw new Error("Deps.afterInvalidate needs a currentComputation");
Deps.currentComputation.afterInvalidate(f);
},
depend: function (v) {
if (! Deps.active)
return false;
@@ -307,16 +278,8 @@
return v.addDependent();
},
atFlush: function (f) {
Deps.nonreactive(function () {
Deps.run(function (c) {
c.onInvalidate(function () {
// (wrap f so as not to pass c)
f();
});
c.stop();
});
});
afterFlush: function (f) {
afterFlushCallbacks.push(f);
}
});

82
packages/deps/deps_tests.js
View File

@@ -1,3 +1,10 @@
// TO TEST:
// - Subscriptions use afterFlush inside onInvalidate
// - No current computation from onInvalidate
// - Subscription de-duping with nested Deps.run
// - Deps.flush throws
// - onInvalidate behavior
Tinytest.add('deps - run', function (test) {
var d = new Deps.Variable;
var x = 0;
@@ -127,10 +134,6 @@ Tinytest.add("deps - nested run", function (test) {
changeAndExpect(f, 'f');
// kill A
a.changed();
// This flush would be unnecessary if outstanding callbacks
// were processed in the containment order of their contexts
// (i.e. parents before children)
Deps.flush();
changeAndExpect(f, '');
changeAndExpect(e, '');
changeAndExpect(d, '');
@@ -166,8 +169,7 @@ Tinytest.add("deps - flush", function (test) {
Deps.flush();
test.equal(buf, 'aa');
/////
// Can't cause rerun nested in run
//////
buf = "";
@@ -178,26 +180,21 @@ Tinytest.add("deps - flush", function (test) {
c.invalidate();
Deps.onInvalidate(function () {
buf += "<";
buf += "*";
});
Deps.afterInvalidate(function () {
buf += ">";
});
if (c.firstRun)
Meteor.flush();
});
test.equal(buf, 'a');
test.equal(buf, 'a*');
Deps.flush();
test.equal(buf, 'a<a>');
test.equal(buf, 'a*a');
c2.stop();
test.equal(buf, 'a*a*');
Deps.flush();
test.equal(buf, 'a<a><>');
test.equal(buf, 'a*a*');
/////
// Can flush a diferent run from a run;
// no current computation in onInvalidate
// no current computation in afterFlush
buf = "";
@@ -206,28 +203,26 @@ Tinytest.add("deps - flush", function (test) {
// invalidate first time
if (c.firstRun)
c.invalidate();
Deps.onInvalidate(function () {
Deps.afterFlush(function () {
buf += (Deps.active ? "1" : "0");
});
});
Deps.atFlush(function () {
Deps.afterFlush(function () {
buf += 'c';
});
var c4 = Deps.run(function (c) {
c4 = c;
buf += 'b';
Meteor.flush();
buf += 'b';
});
test.equal(buf, 'ab0acb');
Deps.flush();
test.equal(buf, 'aba0c0');
c3.stop();
c4.stop();
Deps.flush();
});
Tinytest.add("deps - lifecycle", function (test) {
@@ -253,21 +248,17 @@ Tinytest.add("deps - lifecycle", function (test) {
test.equal(c, Deps.currentComputation);
test.equal(c.stopped, false);
test.equal(c.invalidated, false);
test.equal(c.active, true);
test.equal(c.firstRun, firstRun)
test.equal(c.firstRun, firstRun);
Deps.onInvalidate(makeCb());
Deps.afterInvalidate(makeCb());
Deps.onInvalidate(makeCb()); // 1, 6, ...
Deps.afterFlush(makeCb()); // 2, 7, ...
Deps.run(function (x) {
x.stop();
// should be ok to attach callback from
// nested run
c.onInvalidate(makeCb());
c.afterInvalidate(makeCb());
c.onInvalidate(makeCb()); // 3, 8, ...
Deps.onInvalidate(makeCb());
Deps.afterInvalidate(makeCb());
Deps.onInvalidate(makeCb()); // 4, 9, ...
Deps.afterFlush(makeCb()); // 5, 10, ...
});
runCount++;
@@ -275,44 +266,29 @@ Tinytest.add("deps - lifecycle", function (test) {
c.stop();
});
test.throws(function () {
c1.onInvalidate(function () {});
});
test.throws(function () {
c1.afterInvalidate(function () {});
});
firstRun = false;
test.equal(runCount, 1);
test.equal(buf, []);
test.equal(buf, [4]);
c1.invalidate();
test.equal(runCount, 1);
test.equal(c1.invalidated, true);
test.equal(c1.stopped, false);
test.equal(c1.active, false);
test.equal(buf, []);
test.equal(buf, [4, 1, 3]);
Deps.flush();
test.equal(runCount, 2);
test.equal(c1.invalidated, false);
// 5/6, 11/12, etc. are from the nested run, whose
// invalidation is scheduled each time by the outer
// rerun.
// 1/3 are onInvalidate and 2/4 are afterInvalidate.
test.equal(buf, [5, 6, 1, 3, 2, 4, 11, 12]);
test.equal(buf, [4, 1, 3, 9, 2, 5, 7, 10]);
// test self-stop
buf.length = 0;
shouldStop = true;
c1.invalidate();
test.equal(buf, [6, 8]);
Deps.flush();
// when the computation stops itself, all the
// callbacks from last time and this time should
// get called consecutively, followed by the inner
// computation's 17/18.
test.equal(buf, [7, 9, 8, 10, 13, 15, 14, 16, 17, 18]);
test.equal(buf, [6, 8, 14, 11, 13, 12, 15]);
});

14
packages/livedata/livedata_connection.js
View File

@@ -506,17 +506,19 @@ _.extend(Meteor._LivedataConnection.prototype, {
// computation is invalidated... but not if the rerun just re-subscribes
// to the same subscription! When a rerun happens, we use onInvalidate
// as a change to mark the subscription "inactive" so that it can
// be reused from the rerun. If it isn't reused, it's killed fro
// mafterInvalidate.
// be reused from the rerun. If it isn't reused, it's killed from
// an afterFlush.
Deps.onInvalidate(function (c) {
if (c.stopped)
handle.stop();
else if (_.has(self._subscriptions, id))
self._subscriptions[id].inactive = true;
});
Deps.afterInvalidate(function () {
if (_.has(self._subscriptions, id) && self._subscriptions[id].inactive)
handle.stop();
Deps.afterFlush(function () {
if (_.has(self._subscriptions, id) &&
self._subscriptions[id].inactive)
handle.stop();
});
});
}

2
packages/meteor/timers.js
View File

@@ -44,7 +44,7 @@ _.extend(Meteor, {
// won't be necessary once we clobber the global setTimeout
//
// XXX consider making this guarantee ordering of defer'd callbacks, like
// Deps.atFlush or Node's nextTick (in practice). Then tests can do:
// Deps.afterFlush or Node's nextTick (in practice). Then tests can do:
// callSomethingThatDefersSomeWork();
// Meteor.defer(expect(somethingThatValidatesThatTheWorkHappened));
defer: function (f) {

11
packages/minimongo/minimongo_tests.js
View File

@@ -1770,13 +1770,8 @@ Tinytest.add("minimongo - reactive stop", function (test) {
test.equal(y, "EDCBA");
c.stop();
// stopping doesn't kill the observes until flush.
// observe callbacks don't wait for flush, they are inline.
// stopping kills the observes immediately
coll.insert({_id: 'F'});
test.equal(x, "FEDCBA");
test.equal(y, "FEDCBA");
Deps.flush();
coll.insert({_id: 'G'});
test.equal(x, "FEDCBA");
test.equal(y, "FEDCBA");
test.equal(x, "EDCBA");
test.equal(y, "EDCBA");
});

6
packages/spark/spark.js
View File

@@ -22,7 +22,7 @@
// XXX in landmark-demo, if Template.timer.created throws an exception,
// then it is never called again, even if you push the 'create a
// timer' button again. the problem is almost certainly in atFlush
// timer' button again. the problem is almost certainly in afterFlush
// (not hard to see what it is.)
(function() {
@@ -352,7 +352,7 @@ var scheduleOnscreenSetup = function (frag, landmarkRanges) {
finalized = true;
};
Deps.atFlush(function () {
Deps.afterFlush(function () {
if (finalized)
return;
@@ -976,7 +976,7 @@ Spark.list = function (cursor, itemFunc, elseFunc) {
};
var later = function (f) {
Deps.atFlush(function () {
Deps.afterFlush(function () {
if (! stopped)
withEventGuard(f);
});

2
packages/spark/spark_tests.js
View File

@@ -412,7 +412,7 @@ Tinytest.add("spark - heuristic finalize", function (test) {
Deps.flush();
R.set(456); // won't take effect until flush()
test.equal(div.html(), "<p>The number is 123.</p><hr><br><br><u>underlined</u>");
test.equal(R.numListeners(), 1);
test.equal(R.numListeners(), 0); // listener already gone
Deps.flush();
test.equal(div.html(), "<p>The number is 456.</p><hr><br><br><u>underlined</u>");
test.equal(R.numListeners(), 1);

2
packages/test-helpers/onscreendiv.js
View File

@@ -50,7 +50,7 @@ OnscreenDiv.prototype.kill = function() {
if (self.div.parentNode)
self.div.parentNode.removeChild(self.div);
Deps.atFlush(function () {
Deps.afterFlush(function () {
Spark.finalize(self.div);
});
};

2
packages/test-helpers/wrappedfrag.js
View File

@@ -31,7 +31,7 @@ WrappedFrag.prototype.release = function() {
// decrement frag's GC protection reference count
// Clean up on flush, if hits 0. Wait to decrement
// so no one else cleans it up first.
Deps.atFlush(function () {
Deps.afterFlush(function () {
if (! --frag["_protect"]) {
Spark.finalize(frag);
}