Highland.js中流的异步转换

Question

我正在尝试使用Highland.js在一组Mongoose模型上执行数据库更新脚本，它似乎非常适合在Model.find()上进行QueryStream调用。我有一些同步的事情要做(更新我的模型以符合新的模式，一些清理操作)，最后我想对文档执行save()操作。我已经配置了一些需要运行的预保存钩子，并且更新并不真正与直接的Model.update()兼容。我设法让它在某种程度上通过Q.js和Highland的组合工作：

var sender_stream = Sender.find({}).stream();
var promise_save = function(document) {
    var deferred = Q.defer();
    document.save(deferred.makeNodeResolver());
    return _(deferred.promise);
}

var sender_deferred = Q.defer();
_(sender_stream).map(function(sender) {
    // set some fields on sender...
    return sender;
}).map(promise_save).series().on('done', sender_deferred.resolve).resume();

然而，这似乎并没有解决承诺，我不确定这是否是让事情变得美好和顺畅的“正确”方式……将Q.js和Highland.js如此紧密地结合在一起似乎也很奇怪。有没有更好的方法？

Answer 1

我对Q或Highland了解不多。但这似乎是querystreams上transform函数的一个简单用例。

var stream = Sender.find({}).stream({ transform: manipulate })

function manipulate(document) {
    // do stuff here
    return document;
}

stream.on("data", function(document) {
    stream.pause()
    document.save(function(error) {
        // error handle, maybe stream.destroy([err]) if you want it to stop immediately
        stream.resume();
    });
});

stream.on("error", function(err){
    //error handle
});

stream.on("close", function(){
    console.log("hopefully this worked for you");
});

转换函数将在发出'data‘事件之前在文档上运行。一旦转换函数完成了它的工作，它的返回值就被发送给'data‘函数。然后，您只需暂停/保存/恢复。

Answer 2

您可以使用Highland的异步功能：http://highlandjs.org/#async，而不是Promise。Mongoose也会返回一个Promise，所以你可以用Highland而不是async function风格来包装它，但仍然要避免添加Q。

我建议使用.flatMap()而不是.map()和.series()来将这些流展平为一个文档流。然后，添加.done()也可以用来创建一个.resume()，而不是使用与'done‘事件侦听器相结合的Thunk。

老实说，我不是100%确定为什么你在调用“完成”事件时会有问题。

var sender_stream, set_fields, save, sender_deferred;

sender_stream = Sender.find({}).stream();

save = function save(document) {
    return _(function(push, next) {
        document.save(function(err, result) {
            push(err, document);
            push(null, _.nil);
        });
    });
};

set_fields = function setFields(sender) {
    // set some fields on sender...
    return sender;
};

sender_deferred = Q.defer();

_(sender_stream)
    .map(setFields)
    .flatMap(save)
    .done(function() {
        sender_deferred.resolve();
    });