Scan

1. 概要

Scan は、上流から値が流れてくるたびに「前回までの累積値」と「今回の値」から新しい累積値を作り、その累積値を逐次下流へ流すオペレーターです。

AggregateAsync がシーケンス全体を 1 つの累積結果に集約して Task として返すのに対し、Scan は途中経過を含むすべての累積値を Observable として発行します。累積合計や移動平均のように、値が流れるたびに更新される状態を扱うのに適しています。

2. シグネチャ

シードなし（最初の値を初期値とする）

public static Observable<TSource> Scan<TSource>(
    this Observable<TSource> source,
    Func<TSource, TSource, TSource> accumulator)

最初の値をそのまま下流へ流し、2 番目以降は accumulator(累積値, 新しい値) の結果を下流へ流します。上流が空の場合は何も発行されません。

source.Scan((acc, x) => acc + x)

シードあり（初期値を指定する）

public static Observable<TAccumulate> Scan<TSource, TAccumulate>(
    this Observable<TSource> source,
    TAccumulate seed,
    Func<TAccumulate, TSource, TAccumulate> accumulator)

seed を初期値として使い、最初の値から accumulator(seed, 値) の結果を下流へ流します。累積値の型を上流の型と異なるものにできます。

source.Scan(0, (acc, x) => acc + x)

overload の使い分け

overload	使う場面
Scan(Func<TSource, TSource, TSource>)	累積値の型が上流と同じで、初期値が不要な場合
Scan(TAccumulate, Func<..., TAccumulate>)	初期値を明示したい場合や、累積値の型を変えたい場合

3. マーブルダイアグラム

上流から 1、2、3 が流れ、Scan((acc, x) => acc + x) によって累積合計 1、3、6 が下流へ流れます。

4. サンプルコード

using R3;

// シードなし: 累積合計
var source = Observable.Range(1, 5);

source.Scan((acc, x) => acc + x)
    .Subscribe(x => Console.WriteLine(x));
// 出力: 1, 3, 6, 10, 15

シードありの例:

using R3;

// シードあり: 初期値 100 から累積合計
var source = Observable.Range(1, 5);

source.Scan(100, (acc, x) => acc + x)
    .Subscribe(x => Console.WriteLine(x));
// 出力: 101, 103, 106, 110, 115

型を変える例:

using R3;

var source = Observable.Range(1, 4);

source.Scan("合計:", (acc, x) => $"{acc} {x}")
    .Subscribe(x => Console.WriteLine(x));
// 出力:
// 合計: 1
// 合計: 1 2
// 合計: 1 2 3
// 合計: 1 2 3 4