node.js - 是否可以在nodejs Async( waterfall ，系列等...)中构建动态任务列表

Question

我正在从 mongo 中包含 Node 和边缘数据的一些集合中提取信息。首先我必须得到 Node ，这样我才能捕获它的边缘。一旦我有一个边缘列表，我就会返回并获取更多 Node (等等。基于深度值)。以下代码是我尝试使用 async.waterfall 和任务列表的一个松散示例。

最初我只有一个任务，但是一旦我进行了第一个查询，我就会添加到任务数组中。不幸的是，这似乎没有向 async 注册，并且它不会继续处理我正在添加的任务。

有没有更好的方法来做到这一点？

var async = require('async')
var mongoose = require('mongoose')
var _ = requrie('underscore')

var Client = this.Mongo.connect(/*path to mongo*/)
var Node = mongoose.Schema({
    id : String,
    graph_id : String
})
var Edge = mongoose.Schema({
    id : String,
    source_id : String,
    destination_id : String
})
var Nodes = Client.model('myNode', Node)
var Edges = Client.model('myEdge', Edge)
var funcs = []
var round = 1
var depth = 2

var query = {
    node : {
        id : '12345'
    },
    edge : {
        id : '12345'   
    }
}

var addTask = function(Nodes, Edges, query, round, depth) {
    return function(callback) {
    queryData(Nodes, Edges, query, function(err, node_list) {
        if(depth > round) {
             round++
             function_array.push(addTask(Nodes, Edges, query, round, depth))
        }
    })
}
}

var queryData = function(Nodes, Edges, query, cb) {
async.waterfall([
    function(callback) {
        Nodes.find(query.node, function(err, nodes) {
            var node_keys = _.map(nodes, function(node) {
                 return node.id  
            })
            callback(null, nodes, node_keys)  
        })
    },
    function(nodes, node_keys, callback) {
        query.edge.$or = [ {'source_id' : {$in:node_keys}}, {'destination_id' : {$in:node_keys}} ]
        Edges.find(query.edge, function(err, edges) {
            var edge_keys = _.map(edges, function(edge) {
                if(edge['_doc']['source_id'] != query.node.id) {
                     return edge['_doc']['source_id']
                } else {
                     return edge['_doc']['destination_id']
                }
                callback(null, nodes, edges, node_keys, edge_keys)
            })
        })
    }
], function(err, nodes, edges, node_keys, edge_keys) {
    // update the results object then...
    cb(null, _.uniq(edge_keys)
 })
}

var function_array = []
function_array.push(addTask(Nodes, Edges, query, round, depth))

async.waterfall(function_array, function(err) {
    Client.disconnect()
    //this should have run more than just the initial task but does not
})    

---------- 更新 ------------- --

因此，在尝试通过添加尾随函数来尝试让 Async waterfall 或系列做到这一点之后，我决定改用 async.whilst 并且现在对解决方案感到满意。

function GraphObject() {
  this.function_array = []
}

GraphObject.prototype.doStuff = function() {
  this.function_array.push(this.buildFunction(100))
  this.runTasks(function(err) {
     console.log('done with all tasks')
  }
}

GraphObject.prototype.buildFunction = function(times) {
  return function(cb) {
    if(times != 0) {
      this.function_array.push(this.buildFunction(times - 1))
    }
    cb(null)
  }
}

GraphObject.prototype.runTasks = function(cb) {
  var tasks_run = 0
  async.whilst(
    function(){
      return this.function_array.length > 0
    }.bind(this),
    function(callback) {
      var func = this.function_array.shift()
      func.call(this, function(err) { 
        tasks_run++
        callback(err) 
      })
    }.bind(this),
    function(err) {
      console.log('runTasks ran '+tasks_run+' tasks')
      if(err) {
        cb(500)
      }
      cb(null)
    }.bind(this)
  )
}

Answer 1

function_array 中的任务只能向数组中添加新任务，前提是它不是数组中的最后一个任务。

在您的情况下，您的 function_array 仅包含 1 个任务。该任务本身无法添加其他任务，因为它是最后一个任务。

解决方案是在数组中有 2 个任务。一个 startTask 引导进程，一个 finalTask​​ 更像是一个虚拟任务。在这种情况下，

function_array = [startTask, finalTask​​];

然后startTask会添加taskA，taskB会添加task C，最终

function_array = [startTask, taskA, taskB, taskC, finalTask​​];

下面的示例代码说明了这些概念。

    var async = require('async');
    var max = 6;

    var nodeTask = function(taskId, value, callback){
        var r = Math.floor(Math.random() * 20) + 1;
        console.log("From Node Task %d: %d", taskId, r);

        // add an edge task
        if (taskId < max) {
            function_array.splice(function_array.length-1, 0, edgeTask);
        }

        callback(null, taskId + 1, value + r);
    };

    var edgeTask = function(taskId, value, callback){
        var r = Math.floor(Math.random() * 20) + 1;
        console.log("From Edge Task %d: %d", taskId, r);

        // add a node task
        if (taskId < max) {
            function_array.splice(function_array.length-1, 0, nodeTask);
        }

        callback(null, taskId + 1, value + r);
    };

    var startTask = function(callback) {
        function_array.splice(function_array.length-1, 0, nodeTask);
        callback(null, 1, 0);
    };

    var finalTask = function(taskId, value, callback) {
        callback(null, value);
    };

    var function_array = [startTask, finalTask];

    async.waterfall(function_array, function (err, result) {
        console.log("Sum is ", result);
    });