Async.waterfall:已经调用了回调?

Question

我知道这是以前在这里提出的一个话题，但我正在将async.waterfall与rethinkdb一起使用，并且我得到了Error: Callback was already called。奇怪的是，即使它抛出该错误并使应用程序崩溃，它仍然会创建我需要的数据库和表。我已经阅读了其他一些带有答案的帖子，例如 NodeJS 异步：已经调用了回调？或使用 async.waterfall，但我似乎无处可去。我的控制台还告诉我错误在db.js:40:9，但我是 Node 的新手，只是不确定它想要什么回调。我做错了什么？我需要在这里嵌套我的回调吗？我使用的代码发布在下面。非常感谢我能在这里获得的任何帮助，如果需要，我可以发布其他相关代码。谢谢大家。

分贝.js：

exports.setDatabaseAndTables = function() {
async.waterfall([
function connect(callback) {
r.connect(config.rethinkdb, callback);
},
function createDatabase(connection, callback) {
// Create the database if needed.
r.dbList().contains(config.rethinkdb.db).do(function(containsDb) {
return r.branch(
containsDb,
{created: 0},
r.dbCreate(config.rethinkdb.db)
);
}).run(connection, function(err) {
callback(err, connection);
});
},
function createTable(connection, callback) {
// Create the tables if needed.
r.tableList().contains('todos').do(function(containsTable) {
return r.branch(
containsTable,
{created: 0},
r.tableCreate('todos')
);
}).run(connection, function(err) {
callback(err, connection);
});
r.tableList().contains('users').do(function(containsTable) {
return r.branch(
containsTable,
{created: 0},
r.tableCreate('users')
);
}).run(connection, function(err) {
callback(err, connection);
});
},
function createIndex(connection, callback) {
// Create the indexes if needed.
r.table('todos').indexList().contains('createdAt').do(function(hasIndex) {
return r.branch(
hasIndex,
{created: 0},
r.table('todos').indexCreate('createdAt')
);
}).run(connection, function(err) {
callback(err, connection);
});
r.table('users').indexList().contains('createdAt').do(function(hasIndex) {
return r.branch(
hasIndex,
{created: 0},
r.table('users').indexCreate('createdAt')
);
}).run(connection, function(err) {
callback(err, connection);
});
},
function waitForIndex(connection, callback) {
// Wait for the index to be ready.
r.table('todos').indexWait('createdAt').run(connection, function(err, result) {
callback(err, connection);
});
r.table('users').indexWait('createdAt').run(connection, function(err, result) {
callback(err, connection);
});
}
], 
function(err, connection) {
if(err) {
console.error(err);
process.exit(1);
return;
}
});
};

Answer 1

对于那些尝试"async.waterfall"的人来说，这是一个常见问题。

以下是将"createTable"，"createIndex"和"waitForIndex"拆分为2个函数的解决方案：

exports.setDatabaseAndTables = function() {
async.waterfall([
function connect(callback) {
r.connect(config.rethinkdb, callback);
},
function createDatabase(connection, callback) {
// Create the database if needed.
r.dbList().contains(config.rethinkdb.db).do(function(containsDb) {
return r.branch(
containsDb,
{created: 0},
r.dbCreate(config.rethinkdb.db)
);
}).run(connection, function(err) {
callback(err, connection);
});
},
function createTableTodos(connection, callback) {
// Create the tables if needed.
r.tableList().contains('todos').do(function(containsTable) {
return r.branch(
containsTable,
{created: 0},
r.tableCreate('todos')
);
}).run(connection, function(err) {
callback(err, connection);
});
},
function createTableUsers(connection, callback) {
r.tableList().contains('users').do(function(containsTable) {
return r.branch(
containsTable,
{created: 0},
r.tableCreate('users')
);
}).run(connection, function(err) {
callback(err, connection);
});
},
function createIndexTodos(connection, callback) {
// Create the indexes if needed.
r.table('todos').indexList().contains('createdAt').do(function(hasIndex) {
return r.branch(
hasIndex,
{created: 0},
r.table('todos').indexCreate('createdAt')
);
}).run(connection, function(err) {
callback(err, connection);
});
},
function createIndexUsers(connection, callback) {
r.table('users').indexList().contains('createdAt').do(function(hasIndex) {
return r.branch(
hasIndex,
{created: 0},
r.table('users').indexCreate('createdAt')
);
}).run(connection, function(err) {
callback(err, connection);
});
},
function waitForIndexTodos(connection, callback) {
// Wait for the index to be ready.
r.table('todos').indexWait('createdAt').run(connection, function(err, result) {
callback(err, connection);
});
},
function waitForIndexUsers(connection, callback) {
r.table('users').indexWait('createdAt').run(connection, function(err, result) {
callback(err, connection);
});
}
],
function(err, connection) {
if(err) {
console.error(err);
process.exit(1);
return;
}
});
};