使用 cassandra 和流星。
let client = new cassandra.Client({contactPoints: [cassandraHost]});
var cassandraExecSync = Meteor.wrapAsync(client.execute, client);
MyProject.Feed.CassandraMeteorWrap = {
insertNewPost: function (userId, postContentJson, relevance) {
var insertCommand = insert(userId, postContentJson, relevance);
try {
return cassandraExecSync(insertCommand);
} catch (err) {
console.log("error inserting: " + insertCommand);
console.log(err);
}
}
所以我用 Meteor.wrapAsync 包装了 Cassandra.client.execute (最后一个 arg 作为回调)
前几个插入工作,但在几个插入后(插入被定期调用)我得到:
[错误:Meteor 代码必须始终在 Fiber 中运行。尝试使用 Meteor.bindEnvironment 包装传递给非 Meteor 库的回调。]
更新:调试流星显示堆栈跟踪,异常从我在.onTimeout()上使用“cassandra-driver”的npm包开始:
function listOnTimeout() {
var msecs = this.msecs;
var list = this;
debug('timeout callback ' + msecs);
var now = Timer.now();
debug('now: %d', now);
var first;
while (first = L.peek(list)) {
// If the previous iteration caused a timer to be added,
// update the value of "now" so that timing computations are
// done correctly. See test/simple/test-timers-blocking-callback.js
// for more information.
if (now < first._monotonicStartTime) {
now = Timer.now();
debug('now: %d', now);
}
var diff = now - first._monotonicStartTime;
if (diff < msecs) {
list.start(msecs - diff, 0);
debug(msecs + ' list wait because diff is ' + diff);
return;
} else {
L.remove(first);
assert(first !== L.peek(list));
if (!first._onTimeout) continue;
// v0.4 compatibility: if the timer callback throws and the
// domain or uncaughtException handler ignore the exception,
// other timers that expire on this tick should still run.
//
// https://github.com/joyent/node/issues/2631
var domain = first.domain;
if (domain && domain._disposed) continue;
try {
if (domain)
domain.enter();
var threw = true;
**first._onTimeout();**
if (domain)
domain.exit();
threw = false;
} finally {
if (threw) {
// We need to continue processing after domain error handling
// is complete, but not by using whatever domain was left over
// when the timeout threw its exception.
var oldDomain = process.domain;
process.domain = null;
process.nextTick(function() {
list.ontimeout();
});
process.domain = oldDomain;
}
}
}
}
debug(msecs + ' list empty');
assert(L.isEmpty(list));
list.close();
delete lists[msecs];
}