reactive-programming - 如何编写带有动态 throttle 的供应？

Question

对于我正在重构的聊天机器人，不需要锁定来管理其大部分状态，它通过 websocket 连接到的网站将可以从普通用户接收的消息节流到 0.6 秒，将语音用户接收到 0.3 秒的速率，而管理员没有 throttle 。直到建立连接后的某个时间点才知道用户是否有声音或管理员是否有声音；在此之前，每个人都被视为普通用户。

目前，我通过放置 react 来处理受限制的消息。用于在连接被强制关闭后退出的循环中监听消息的块。当 throttle 更新时，我调用 done进入下一次迭代，更新 whenever阻止发送消息的供应以具有更新的 throttle 。这是可怕的，活泼的代码!

我可以做些什么来 (a) 确保连接以 0.3 秒的节流开始，该节流可以在建立 websocket 连接后立即使用，(b) 可以调用在需要时更新此节流的方法，以及 (c ) 不保留与此 throttle 相关的任何状态(可以通过其他方式推断)？

编辑:我之前忘了提到这一点，但是所有类型的用户都有不受限制的消息，而不仅仅是管理员。

Answer 1

我找到了一种使用 Supply.migrate 的组合来做到这一点的方法, Supply.map , 和 Supply.merge .如果我为节流消息、未节流消息和节流更新创建供应商，我可以将节流更新供应映射到使用任何发出的节流进行节流的节流消息供应，请调用 Supply.migrate在结果供应上，并将其与不受限制的消息供应合并。这导致我可以使用一个电源来处理发送所有类型的消息:

react {
    my Supplier:D $unthrottled .= new;
    my Supplier:D $throttled   .= new;
    my Supplier:D $throttles   .= new;

    whenever Supply.merge(
        $unthrottled.Supply,
        $throttles.Supply.map({ $throttled.Supply.throttle: 1, $_ }).migrate,
    ) -> Str:D $message {
        say sprintf '%s @ %f', $message, now;
        done if ++$ == 12;
    }

    $throttles.emit: 1.2;
    $throttled.emit: "$_" for 1..3;
    whenever Promise.in(1) {
        $unthrottled.emit: "$_" for 7..9;
    }
    whenever Promise.in(5) {
        $throttles.emit: 0.6;
        $throttled.emit: "$_" for 4..6;
        whenever Promise.in(1) {
            $unthrottled.emit: "$_" for 10..12;
        }
    }
}

# OUTPUT:
# 1 @ 1579731916.713831
# 7 @ 1579731917.764047
# 8 @ 1579731917.769012
# 9 @ 1579731917.774584
# 2 @ 1579731917.913512
# 3 @ 1579731919.123057
# 4 @ 1579731921.749377
# 5 @ 1579731922.353073
# 10 @ 1579731922.768212
# 11 @ 1579731922.773777
# 12 @ 1579731922.780446
# 6 @ 1579731922.963087