javascript - 使用 NPM 包和 Meteor.wrapAsync 的异步问题

Question

当使用 npm 模块 (node-celery) 中的函数 client.call 时，client.call 的回调函数似乎未执行。

Meteor.methods({
    'estimates.request'(data) {
        var celery = require('node-celery')
        var client = celery.createClient({...})

        client.on('connect', function() {
            console.log('connected');                   // this is executed
            client.call('proj.tasks.register', [name],
                function(err, result) {
                    console.log('result: ', result);    // this is not executed
                    client.end();
                    return result;
                }
            );
        });
    }
});

尝试将 client.call 包装在 Meteor.wrapAsync 中:

callFunc = client.on('connect', function() {
    console.log('connected');                   // this is executed
    client.call('proj.tasks.register', [name],
        function(err, result) {
            console.log('result: ', result);    // this is not executed
            client.end();
            return result;
        }
    );
});

callFuncSync = Meteor.wrapAsync(callFunc)
callFuncSync()

但这会在 Meteor 服务器控制台中引发错误:

err:  [Error: Meteor code must always run within a Fiber. Try wrapping callbacks that you pass to non-Meteor libraries with Meteor.bindEnvironment.]
err:  { [Error: read ECONNRESET] code: 'ECONNRESET', errno: 'ECONNRESET', syscall: 'read' }

问题:我们应该如何使用Meteor.bindEnvironment来解决这个问题？

Answer 1

来自文档，

Wrap a function that takes a callback function as its final parameter. The signature of the callback of the wrapped function should be function(error, result){}

您的代码只是包装事件附件调用的返回值。

您可以包装整个事情(连接+任务调用)，但就您而言，我建议采用不同的方法:

现在，每次有人调用该方法时，您都会连接到 Celery。如果可能的话，我建议与 Celery 保持持久连接。

您尝试包装的函数不符合 wrapAsync 要求，因此您必须将它们包装在符合 wrapAsync 要求的函数中。

在下面的代码中，连接和调用函数都被处理了。请注意，这些函数采用一个 cb 参数，该参数将由 Meteor 提供给它们，并根据需要使用错误和/或结果来调用它。

然后这些函数将传递给 wrapAsync。

如果错误传递给回调，则它们的同步版本会抛出异常；如果以同步方式调用它们(即不传递回调)，则使用纤程模拟同步运行。这就是 try..catch block 的原因。

import { Meteor } from 'meteor/meteor';
const celery = require('node-celery'); // or `import` it

function createAndConnectAsync(details, cb) {
    const client = celery.createClient(details);
    const errHandler = function(e) {
        cb(e, client);
    };
    client.once('connect', function() {
        client.off('error', errHandler);
        cb(null, client); // success
    });
    client.once('error', errHandler);
}

function callAsync(client, task, args, cb) {
    client.call(task, args, function(result) {
       cb(null, result);
    });
}

const createAndConnect = Meteor.wrapAsync(createAndConnectAsync);
const call = Meteor.wrapAsync(callAsync);

let client;
try {
   client = createAndConnect({...}); // returns after the client is connected
} catch(e) {
    // connection error
}

Meteor.methods({
    'estimates.request'(data) {
        // generate `name`
        const result = call(client, 'proj.tasks.register', [name]);
        return result;
    }
});