Stream

Streamとは

以下の条件を満たすイベントがあったとします。

引数の一つに配列(Enumerable)を含んでいる。
その引数に2要素の配列を渡して一度イベントを発生させることと、1要素ずつに分割して2度発生させることがほぼ同じ意味になる

このようなイベントは、Streamイベントとして扱うことができます。具体的には、以下のようなイベントはStreamイベントです。

on_extract_receive_message do |datasource, messages|
  # 処理
end

以下、そのような配列を期待する引数のことを「ストリーム引数」、ストリーム引数を含むイベントを「ストリームイベント」と呼びます。

Streamが解決すること

Stream機能は、ストリームイベントを無限リストとして扱うことで、イベントリスナのコードを簡略化するためのものです。

以下の3種類のコードは、ほとんど同じ意味です。

従来のイベントリスナ

on_extract_receive_message do |datasource, messages|
  if datasource == hoge
    messages.each do |message|
      # 処理
    end
  end
end

subscribeを使った例

subscribe(:extract_receive_message, :mastodon_appear_toots) do |messages|
  messages.each do |message|
    # 処理
  end
end

subscribeとPluggaloid::Streamを使った例

subscribe(:extract_receive_message, :mastodon_appear_toots).each do |message|
  # 処理
end

頻出するイベントリスナのパターンを簡略化

上記の3つの例を見ればわかるとおり、単純にsubscribeを使えばコードの行数が短くなります。

具体的には、以下のようなものを書く必要がなくなります。

早期リターン処理 :: ストリーム引数以外の引数を参照して、必要がなければ処理を実行しない。
繰り返し処理 :: ストリーム引数をeachなどで走査処理する。

呼び出し回数の最適化

いくつかの場合に於いて、subscribeのほうがonより高速に処理できます。以下のように、10回イベントが呼び出されたとします。

Plugin.call(:extract_receive_message, [message_0])
Plugin.call(:extract_receive_message, [message_1])
Plugin.call(:extract_receive_message, [message_2])
Plugin.call(:extract_receive_message, [message_3])
Plugin.call(:extract_receive_message, [message_4])
Plugin.call(:extract_receive_message, [message_5])
Plugin.call(:extract_receive_message, [message_6])
Plugin.call(:extract_receive_message, [message_7])
Plugin.call(:extract_receive_message, [message_8])
Plugin.call(:extract_receive_message, [message_9])

従来のイベントリスナだと、10回イベントが呼ばれます。一度目の呼び出しは [message_0] のような1要素の配列、次の呼び出しは [message_1] のような1要素の配列……といった具合です。

一方、subscribeを使った例やsubscribeとPluggaloid::Streamを使った例であれば、処理のタイミングによってはストリーム引数が結合され、 [message_0, message_1, …, message_9] のような10要素の配列にしたうえで、1度のイベント呼び出しになる場合があります。イベントの呼び出し回数は少なくなるぶん高速な処理が期待できます。

ストリーム引数以外の引数によるイベント呼び出しの省略

subscribeを使った例やsubscribeとPluggaloid::Streamを使った例の場合、ストリーム引数以外の引数を事前に指定しておいて、ストリーム引数以外の引数が全て一致するイベントを受け取った時のみ処理が実行されます。従来のイベントリスナでもそのようなものを自力で実装していたと思いますが、subscribeに任せたほうが高速になります。

いつストリームを使うか

まず、リスナを定義する時、対象のイベントがストリームイベントである必要があります。ストリームイベントではない、または不明な場合は従来のイベントリスナを使います。

ストリームイベントであることがわかっている場合は原則ストリームリスナを使い、呼び出し回数の最適化やストリーム引数以外の引数によるイベント呼び出しの省略のような最適化をされると処理の都合が悪いという場合は、従来のイベントリスナを使ってください。

従来のイベントリスナは自由度が高く、subscribeを使った例やsubscribeとPluggaloid::Streamを使った例は自由度が低い代わりに最適化のヒントを与えることができます。

ストリームリスナ

Pluginコンテキストのsubscribe()メソッドで、頻出するイベントリスナのパターンを簡略化を解決します。

subscribe(:extract_receive_message, :mastodon_appear_toots) do |message|
  # 処理
end

subscribeは以下のような引数を取ります。

Table 1. subscribe(event_slug, *rest) { |contents| … } → nil
引数名	型	意味
event_slug	Symbol	subscribeするイベントの識別子
*rest	-	ストリーム引数以外の引数
contents	Array	受け取ったイベントのストリーム引数を一つ以上含む配列

Table 2. subscribe(event_slug, *rest) → Pluggaloid::Stream
引数名	型	意味
event_slug	Symbol	subscribeするイベントの識別子
*rest	-	ストリーム引数以外の引数
戻り値	Pluggaloid::Stream	受け取ったイベントのストリーム引数を、受け取り次第列挙する無限リスト

ブロックは、全ての *rest が等しい場合にのみ呼ばれます。ここでいう「等しい」とはObject#hashの結果が等しいものです。

無限リストとして利用する

subscribeにブロックを渡さない場合、Pluggaloid::Streamを返します。Pluggaloid::StreamはEnumerableです。

subscribe(:extract_receive_message, :mastodon_appear_toots).each do |message|
  # 処理
end

無限リストなので、たとえばEnumerable#selectで更に結果を絞り込むこともできます。

subscribe(:extract_receive_message, :mastodon_appear_toots).select { |message|
  message.description.include?('まちカドまぞくを見ろ')
}.each do |message|
  # 処理
end

ストリームイベントの定義

defevent :extract_receive_message, prototype: [Plugin::Extract::Setting, Pluggaloid::STREAM]

defevent でイベントを定義するとき、 prototype: キーワード引数に渡す配列の中に、 Pluggaloid::STREAM という定数を含めると、ストリームイベントになります。