在消息之间引入延迟

Question

我用Node.js。我有一个 MQTT 消息事件处理程序
index.js

client.on('message', function (topic, message) {
  // calls another function
  my_function(topic,message);
})

在接收消息时调用另一个函数 my_function。

async function my_function(topic,message) {
   const value = await dataFromPLC();
///processes the value together with message
}

使用 exports.dataFromPLC = dataFromPLC 从另一个文件导出并导入到我的主函数中的函数 dataFromPLC 看起来像这样
PLCfunctions.js

let client = new S7Client(plcSettings);
client.on('error', console.error);
 
async function dataFromPLC (){
  try {
    await client.connect();
  } catch (err){
    console.error(err);
  }
 
  try {
    // Read DB
    const res = await client.readDB(dbNr, dbVars);
    return res;
  } catch (err) {
    console.error(err);
  } finally {
    client.disconnect();
  }
}

当我收到单个 MQTT 消息或消息之间有足够的延迟时，没有问题。但是，当我收到两条 MQTT 消息时，它们都调用 my_function 并随后调用 dataFromPLC，中间没有太多延迟。我收到一条错误消息，因为在第二条消息尝试再次使用连接之前，没有足够的时间关闭 PLC 连接。我查看了不同的选项，但不太确定如何解决问题。我能得到一些帮助吗？

Answer 1

您必须设置消息队列，以便 onMessage 仅将输入放入队列并将其处理推迟到以后。例如，您可以使队列成为一个带有 then 作为入队操作的 Promise。这样可以保证在所有先前的处理完成之前不会开始处理。

这是一个小演示，点击按钮模拟传入消息：

let QUEUE = Promise.resolve()

function onMessage(msg) {
    console.log('GOT MESSAGE', msg)
    QUEUE = QUEUE.then(() => process(msg))
}

let pause = n => new Promise(r => setTimeout(r, n));

async function process(msg) {
    console.log('BEGIN', msg)
    await pause(200); console.log('busy', msg)
    await pause(200); console.log('busy', msg)
    await pause(200); console.log('busy', msg)
    await pause(200); console.log('busy', msg)
    await pause(200); console.log('busy', msg)
    console.log('END', msg)
}

msg = 0
document.querySelector('button').addEventListener('click', () => onMessage(++msg))

<button>message</button>

在消息之间引入延迟

Introducing delay between messages

javascript

node.js

mqtt