Async.waterfall：回调已经被调用？

Question

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

db.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;
                }
            });
};