baconjs：使用标准限制连续事件答案

【问题标题】：baconjs: throttle consecutive events with criteriabaconjs：使用标准限制连续事件
【发布时间】：2018-04-14 08:36:38
【问题描述】：

我正在使用 Node.js 编写消息传递应用程序，我需要检测同一用户何时在一个组中发送 N 条连续消息（以避免垃圾邮件发送者）。我正在使用 bacon.js Bus push 来自所有用户的传入消息。

消息如下所示：

{
  "text": "My message",
  "user": { "id": 1, name: "Pep" }
}

这是我目前的工作代码：

const Bacon = require('baconjs')
const bus = new Bacon.Bus();

const CONSECUTIVE_MESSAGES = 5;

bus.slidingWindow(CONSECUTIVE_MESSAGES)
   .filter((messages) => {
       return messages.length === MAX_CONSECUTIVE_MESSAGES &&
              _.uniqBy(messages, 'user.id').length === 1;
    })
   .onValue((messages) => {
       console.log(`User ${__.last(messages).user.id}`);
   });

// ... on every message
bus.push(message);

它创建一个滑动窗口，只保留我想要检测的连续消息的数字。在每个事件上，它过滤数组以让数据流到下一步，只有当窗口中的所有消息都属于同一个用户时。最后，在onValue中，取最后一条消息获取用户id。

对我来说，代码看起来很脏/很复杂：

filter 在流中看起来不太自然。当 N 连续事件符合某些条件时，是否有更好的方法来发出事件？ .
有没有更好的方法在onValue 函数中只接收一个事件与用户（而不是一组消息）。
它并没有真正节流。如果用户在一年内发送 N 条消息，则不应检测到他或她。信息流应该以某种方式忘记旧事件。

有什么改进的方法吗？如果有帮助，我愿意将其迁移到 rxjs。

【问题讨论】：

标签： node.js reactive-programming bacon.js

【解决方案1】：

也许开始

latestMsgsP = bus.slidingWindow(CONSECUTIVE_MESSAGES)
  .map(msgs => msgs.filter(msg => msgAge(msg) < AGE_LIMIT))

看看我们是否应该阻止某人

let blockedUserIdP = latestMsgsP.map(getUserToBlock)

你可以在哪里使用一些无耻的命令，例如

function getUserToBlock(msgs) { if (msgs.length < CONSECUTIVE_MESSAGES) return let prevUserId; for (var i = 0; i < msgs.length; i++) { let userId = msgs[i].user.id if (prevUserId && prevUserId != userId) return prevUserId = userId } return prevUserId }

【讨论】：

【解决方案2】：

考虑尽早映射您感兴趣的属性，然后流的其余部分会更简单。此外，当您增加阈值时，滑动窗口中每个项目的相等性检查不会很好地扩展。考虑改用scan，这样您只需保留一个计数，当当前值和之前的值不匹配时会重置。

bus
    .map('.user.id')
    .scan([0], ([n, a], b) => [a === b ? n + 1 : 1, b])
    .filter(([n]) => n >= MAX_CONSECUTIVE_MESSAGES)
    .onValue(([count, userId]) => void console.log(`User ${userId}`));

【讨论】：