如何编写带有动态节流阀的供应？

Question

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

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

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

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

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

Answer 2

有趣的。我尝试了一下，从我在文档中读到的内容来看，该文档很稀疏，类似以下内容应该可以工作：

my $s = Supply.from-list(^Inf);
my $control = Supplier.new.Supply;
my $throttle = $s.throttle( 10, 1, :$control );

react 
{
  whenever $throttle -> $n 
  { 
    $n.say 
  };

  # change speed every 5 seconds
  whenever Supply.interval(5) -> $x
  {
    "limit: { (1..10).pick }".say;
    $control.emit( "limit: { (1..10).roll }" );
  }
}

但事实并非如此。当遇到 时，程序就会冻结$control.emit( ... )。如果您将其注释掉，它将按预期运行。该方法的相关文档部分throttle：

从给定的 Supply 生成 Supply，但确保通过的消息数量是有限的。

[...]

如果第二个位置参数是数值，则将其解释为时间单位（以秒为单位）。如果您指定 0.1 作为值，那么它可以确保您不会超过十分之一秒的限制。

[...]

:control 命名参数可以选择指定一个 Supply，您可以在节流阀运行时使用它来控制它。可以发送的消息是“key:value”形式的字符串。

[...]

这些消息可以发送到 :control Supply。控制消息由“key:value”形式的字符串组成，例如“limit:4”。