拆分进程以等待 NodeJS 上的事件

Posted 2023-02-18

【中文标题】拆分进程以等待 NodeJS 上的事件

【英文标题】：Split a process to wait for event on NodeJS

【发布时间】：2021-07-03 18:13:57

【问题描述】：

我正在编写一个使用 web3js 通过智能合约传输代币的方法。

当您启动传输事件时，您会得到 txHash，如果您想获取与传输相关的所有其他值，您必须订阅一个事件并等待它发生以获取值.

我必须将值返回给客户，所以我订阅了 transfer 事件并等待它广播返回数据。

问题是这可能需要很长时间（想想从 10 秒到几小时），有时它会给我一个超时，所以前端团队建议通知我一个 webhook 端点，当它发生时我将事件信息转发给它发生。

所以我必须把这个过程一分为二：

进行传输并通知 txHash，并启动一个单独的进程 (2) 来监听事件。

订阅事件，并在事件发生时将其转发到提供的 webhook。

我的代码现在看起来像这样：

function transfer(req, res, next) 

    try 

            contractRouter.transfer(from, to, tokenId).then(function(result)
                transferToWebhook(whHostname, whPath, result);
                next();
            ).fail(function(err)
                return res.status(500).json(status: 'error', name: err.name, message: err.message);
    
    catch (ex) 
        return res.status(500).json(status: 'error', name: ex.name, message: ex.message);
    

传输到 webhook 的函数如下所示：

function transferToWebhook(whHostname, whPath, txHash)
    contractRouter.getTransferEvent(txHash).then(function(result)

        var postData = JSON.stringify(result);
        var options = 
            hostname: whHostname,
            port: 80,
            path: whPath,
            method: 'POST',
            headers: 
                'Content-Type': 'application/json',
            
        
            var req = https.request(options, (res) => 
              console.log(`STATUS: $res.statusCode`);
              console.log(`HEADERS: $JSON.stringify(res.headers)`);
              res.setEncoding('utf8');
              res.on('data', (chunk) => 
                console.log(`BODY: $chunk`);
              );
              res.on('end', () => 
                console.log('No more data in response.');
              );
            );

            req.on('error', (e) => 
              console.log(`problem with request: $e.message`);
            );

            req.write(postData);
            req.end();
    );

订阅传输事件的函数如下所示：

function getTransferEvent(txHash)
    var deferred = q.defer();

    ethereumHandler.setContract(config.get(cABIAddress), cAddress).then(function(abiContract)

        Promise.resolve(txHash).then(function resolver(txHash)

            abiContract.getPastEvents('Transfer',filter: transactionHash: txHash, function(error, events))
            .then(function(event)
                var returnValues = 
                    from: event.returnValues.from,
                    to: event.returnValues.to,
                    tokenId: event.returnValues.tokenId
                
                deferred.resolve(returnValues);
            );
        );
    
    return deferred.promise;
    );

如果我将最后一个函数的代码直接放在传输函数上，则它的代码有效，但如果我尝试通过 transferToWebhook 函数调用它，则不会调用它。

应答第一个请求后如何启动 transferToWebhook 功能？

【参考方案1】：

您可以使用 child_process 模块中的 spawn() 方法生成您的进程，然后监听事件 (process.on('data')) 并使用承诺来使用返回的数据。我不确定它是否能解决您的问题，因为您的函数是contractRouter contractRouter.getTransferEvent(txHash) 的对象，但您应该能够以某种方式对其进行调整。看一个例子来说明我的意思。

在你的 file.js 中

const  spawn  = require('child_process')

function transfertToWebHook(data) 
    getTransfertEvent(data)
        .then((result) => 
            const dts = JSON.parse(result)
            console.log('the res: ', dts)
            // send the res back
        )
        .catch(e => console.log('handle the err: ', e))
   
       console.log('carry on mother process')



function getTransfertEvent(data) 
    return new Promise((resolve, reject) => 
        const sp = spawn(process.execPath, ['childProcess.js'])
        // pass the arguments, here it will be the txHash
        sp.stdin.write(data)
        sp.stdin.end()
        sp.stdout.on('data', (d) => 
            // when datas get proceed you get it back.
            resolve(d.toString())
        )
        sp.stdout.on('error', (e) => 
            reject(e)
        )


        console.log('run what ever need to be proceed on the mother process')
    )

transfertToWebHook('test')

创建另一个文件名 childProcess.js。

使用Tranform流处理process.sdtin数据，然后通过process.stdout返回

const  Transform, pipeline  = require('stream')

const createT = () => 
    return new Transform(
        transform(chunk, enc, next) 
            // here run the code of you getTransfertEventFunction()
            
            // simulate some async process
            setTimeout(() => 
                // chunk is your passed arguments
                // !! chunk is a buffer so encode it as utf8 using 'toString()'
                // make it upperCase to simulate some changement
                const dt = chunk.toString().toUpperCase()

                // return an object as it's what your func return
                const proceededDatas = 
                                name: dt,
                                from: "from datas",
                                to: "to datas",
                                tokenId: "the token"
                        
                const dataString = JSON.stringify(proceededDatas)
                next(null, dataString)
            , 1000)
        
    )

pipeline(process.stdin, createT(), process.stdout, e => console.log(e))

运行代码：node file.js

【讨论】：

谢谢杰罗姆，我在理解如何通过标准输入传递 args 并在另一边读取它们时遇到了一些麻烦，但我想我可能能够调整我的代码以生成子进程在你的帮助下。如果您能在论据部分帮助我，将不胜感激。

NVM，我只是想通了，即使我想将对象作为 arg 传递，但似乎不可能。如果您对如何做有任何见解，请告诉我。感谢您的帮助！

是的，您可以将对象作为参数传递（这样您可以根据需要传递任意数量的参数/参数），您只需先将其字符串化，然后在转换流中解析它（它将在块变量中）并使用它（就像我从子进程回到母进程一样）。

请记住，子进程位于具有不同 PID 的新进程中，因此来自母进程的任何实例化连接都不会在子进程中可用。例如，如果您连接到 API 并从中获取实例，则无法将其获取/传输到子进程（但您可以传输所需的任何数据），连接必须在子进程中完成，然后返回数据然后可以返回到母进程。

你也可以在childProcess中放一些console.log()来调试，它们会出现在你的控制台中（因为母亲和孩子之间有一个开放的流连接）