r3-reactive-extensions

R3：.NET 向けモダンな Reactive Extensions

R3 は、Cysharp が Reactive Extensions をゼロから再実装したもので、「dotnet/reactive と UniRx の新たな未来」です。イベントに LINQ を適用するプログラミングモデルはそのままに、コア型、エラーの契約、スケジューラーを再構築し、System.Reactive（Rx.NET）に長年残されてきた問題を解決します。C# でイベントストリーム、UI 入力、タイマー、プッシュ型パイプラインを組み立てる際に、このスキルを使用してください。

正式な情報源（コードやドキュメントからリンクしてください）：

• リポジトリ：https://github.com/Cysharp/R3
• README（全オペレーターのリファレンス）：https://github.com/Cysharp/R3/blob/main/README.md
• 作者による設計の背景：https://neuecc.medium.com/r3-a-new-modern-reimplementation-of-reactive-extensions-for-c-cf29abcc5826

このスキルを使用する場面

次のような場合に、このスキルを使用してください：

• 時間とともに発生するイベントを組み立てる場合 — UI 入力、センサーやフィードの更新、WebSocket メッセージ、ドメインイベント
• debounce、throttle、merge、combine-latest、distinct-until-changed などのオペレーターが必要な場合
• ReactiveProperty / BindableReactiveProperty を使ってMVVM の状態を構築する場合
• プッシュ型ストリームを Task / async や IAsyncEnumerable と連携させる場合
• System.Reactive、UniRx、IObservable<T> のコードから移行する場合
• 例外によるサブスクリプションの停止、スケジューラーのオーバーヘッド、リークの調査など、Rx の課題に直面している場合

適さない用途： リクエスト／レスポンス型の I/O（async/await を使用してください）、バックプレッシャーを伴う有界のプロデューサー／コンシューマー（System.Threading.Channels を使用してください）、バッチ処理やバックプレッシャーを伴うサーバー側のストリーム処理（Akka.NET Streams を使用してください）。R3 は、すべての Rx と同じく、バックプレッシャーのないプッシュ型です。これらの使い分けについては、csharp-concurrency-patterns スキルを参照してください。

リファレンスファイル

• rx-net-differences.md：System.Reactive（Rx.NET）との主な違いすべて — 新しいコア型、エラーモデル、オペレーター名の変更、削除された API、スケジューラーの変更、移行チェックリスト。
• async-and-integration-patterns.md：一般的なパターン — AwaitOperation による非同期ディスパッチ、Task との連携、IAsyncEnumerable との相互変換、ReactiveProperty / MVVM、サブジェクト、サブスクリプションのライフサイクル。
• scheduling-and-concurrency.md：R3 が並行更新（スレッド処理の契約、Synchronize、ObserveOn）をどのように扱うか、TimeProvider と FrameProvider の違い、それぞれが必要な場面、フェイクプロバイダーを使った決定論的なテストについて。

このスキルの内容はすべて、R3 1.3.1 を使用して実証的に検証済みです。検証時に取得した出力は、証拠としてリファレンスファイルに記載されています。

R3 が存在する理由（「なぜ使うのか」）

作者の neuecc は、System.Reactive の具体的な欠点を解消するために R3 を開発しました。

1. 例外によってサブスクリプションが予告なく終了します。 Rx では、パイプライン内で例外が1つ発生すると OnError が呼び出され、サブスクリプションが永久に解除されます。長期間存続するイベントストリームでは「数十億ドルの失敗」ともいえる問題です（UI イベントが1つ失敗しただけで、サブスクリプション全体が終了します）。R3 はエラーを OnErrorResume に渡し、デフォルトではサブスクリプションを維持します。
2. IScheduler は重く、分かりにくいものです。 ImmediateScheduler／Merge は、実際にサーバーのメモリや CPU を肥大化させることが測定されました。R3 は IScheduler を廃止し、.NET 8 の TimeProvider（ウォールクロック）と、新しい FrameProvider（フレームクロック）を使用します。
3. サブスクリプションのリークを見つけるのは困難です。 R3 はすべての Observable<T> を抽象クラスにし、すべてのサブスクリプションが一か所に集約されるようにしています。これにより、ObservableTracker はスタックトレースとともに、存続中のすべてのサブスクリプションを一覧表示できます。
4. Rx と非同期処理は扱いにくい形で融合していました。 R3 は Rx をイベント優先として扱い、イベントをプル型シーケンスであるかのように見せかけるのではなく、明示的なブリッジ（AwaitOperation、FromAsync、ToAsyncEnumerable）を追加しています。
5. 1つのライブラリで、あらゆる UI に対応します。 プラットフォームに依存しないコアと、Unity、Godot、WPF、WinForms、Avalonia、WinUI3、MAUI、Stride、MonoGame、Blazor 向けの薄いプロバイダーパッケージで構成されています。

---

インストール

dotnet add package R3
# Platform glue (pick what applies): R3.WPF, R3.Avalonia, R3.WinForms, R3.Unity (UPM),
# R3.Godot, ObservableCollections.R3, etc. See the repo README for the full list.

using R3;

---

基本的な考え方

R3 は Rx のインターフェイスを抽象クラスに置き換え、2つのメソッドからなる Rx のエラー契約を、結果を伴う単一の完了通知に置き換えています。

public abstract class Observable<T>
{
    public IDisposable Subscribe(Observer<T> observer);     // tracked centrally
    protected abstract IDisposable SubscribeCore(Observer<T> observer);
}

public abstract class Observer<T> : IDisposable               // the observer IS the subscription
{
    public void OnNext(T value);
    public void OnErrorResume(Exception error);               // error WITHOUT unsubscribing
    public void OnCompleted(Result result);                   // success OR failure terminates
}

構文は (OnNext | OnErrorResume)* OnCompleted(Result)? です。Rx の OnNext* (OnError | OnCompleted)? との違いに注目してください。エラーと終了は切り離されています。エラーは単なる通知です。ストリームを終了するのは OnCompleted だけで、その際に Result.Success または Result.Failure(exception) のいずれかの Result を渡します。

クイックスタート

using R3;

var subscription = Observable
    .EveryValueChanged(model, m => m.SearchText)   // emits when the property changes
    .Debounce(TimeSpan.FromMilliseconds(300))      // Rx called this "Throttle" (see differences)
    .DistinctUntilChanged()
    .SubscribeAwait(async (text, ct) =>
    {
        var results = await _api.SearchAsync(text, ct);
        Render(results);
    }, AwaitOperation.Switch);                      // cancel the in-flight search on a new keystroke

// Dispose to unsubscribe; or route into a DisposableBag / AddTo(token).
subscription.Dispose();

---

コアの動作を検証済み

エラーはデフォルトでは終了を引き起こしません

var subject = new Subject<int>();
subject.Select(x => 100 / x).Subscribe(
    onNext:        x => Console.WriteLine($"next {x}"),
    onErrorResume: e => Console.WriteLine($"errorResume {e.GetType().Name}"),
    onCompleted:   (Result r) => Console.WriteLine($"completed IsSuccess={r.IsSuccess}"));

subject.OnNext(2);   // next 50
subject.OnNext(0);   // errorResume DivideByZeroException   <-- NOT terminated
subject.OnNext(5);   // next 20                             <-- subscription is still alive!
subject.OnCompleted(); // completed IsSuccess=True

これは Rx との動作上の最大の違いです。従来の「エラーがシーケンスを終了させる」動作に戻すには、.OnErrorResumeAsFailure() を挿入します。すると、エラーは OnCompleted(Result.Failure(e)) に渡され、それ以降の OnNext は停止します。Catch で回復できます。実行結果の全記録と、意図的に存在しない Retry については、rx-net-differences.md を参照してください。

非同期ディスパッチは明示的です

R3 の非同期オペレーター（SubscribeAwait、SelectAwait、WhereAwait、…）は、非同期処理が完了するより速く値が到着した場合の動作を決める AwaitOperation を受け取ります。

AwaitOperation	重複実行時の動作	典型的な用途
Sequential（デフォルト）	値をキューに入れ、1つずつ処理します	順序を保った処理
Drop	1つの処理が実行中の間、新しい値を無視します	デバウンス付き送信／クールダウン
Switch	実行中の処理をキャンセルし、新しい処理を開始します	入力中の検索、最新の値を優先
Parallel	すべてを並行して実行します	独立したファンアウト
SequentialParallel	並行して実行し、結果を順番に出力します	並列マップ、順序を保った出力
ThrottleFirstLast	バーストの最初と最後を実行します	先頭／末尾のサンプリング

これらが説明どおりに動作することを検証済みです（Switch が置き換えられた処理の CancellationToken をキャンセルすることも含みます）。async-and-integration-patterns.md を参照してください。

Task と IAsyncEnumerable のブリッジ

// Task -> Observable
await Observable.FromAsync(async ct => await LoadAsync(ct)).FirstAsync();

// Observable -> Task (terminal operators return Task<T>)
List<int> all = await source.ToListAsync();
int last      = await source.LastAsync();

// IAsyncEnumerable -> Observable, and back
await asyncEnumerable.ToObservable().ForEachAsync(Handle);
await foreach (var x in source.ToAsyncEnumerable()) { /* ... */ }

すべて動作を検証済みです。詳細と終端オペレーターの全一覧については、async-and-integration-patterns.md を参照してください。

---

R3 における同時更新の処理方法

R3 は同時に発行するプロデューサーを直列化しません。 Rx と同様に、Rx の文法に従うことを前提としています。つまり、複数のスレッドから同時または再入的に OnNext を呼び出してはいけません。オペレーター（Where、Select、Subject など）は内部でロックされていません。多数のスレッドからステートフルな下流に OnNext を一斉にプッシュすると、状態が破損します。テストでは、List<T> サブスクライバーに対する 20,000 件の同時 OnNext 呼び出しで、アイテムの約半数が失われ、オペレーターチェーン内で例外が発生しました。

解決策は、境界を明示することです。

// Multiple producer threads -> one serialized consumer
subject.Synchronize()                  // lock-based gate; delivery becomes single-threaded
       .Where(x => x.IsValid)
       .Subscribe(Handle);             // verified: 10000/10000 items, no corruption

// Or marshal onto a context/threadpool, which also serializes delivery:
source.ObserveOnThreadPool().Subscribe(Handle);

// For shared MVVM state written from many threads:
var counter = new SynchronizedReactiveProperty<int>(0);   // thread-safe writes

**実践的なルール：**複数のスレッドからストリームに発行する可能性がある場合は、ソースの直後に Synchronize()（または ObserveOn*）を配置するか、SynchronizedReactiveProperty を使用してください。競合状態を完全に再現する方法と出力は、scheduling-and-concurrency.md に記載されています。

---

時間とフレーム：TimeProvider と FrameProvider

R3 には「いつ」を表す概念が2 つあり、どちらもテストで置き換えられる抽象化です。

• TimeProvider（.NET 8 BCL の型）＝実時間です。Delay、Debounce、Interval、Timer、Timeout で使用されます。サーバーやビジネスロジックのコードで使用するものです。
• FrameProvider（R3 固有）＝フレームクロックです。EveryUpdate、DelayFrame(n)、IntervalFrame(n) などで使用されます。

**FrameProvider が必要なのはどのような場合ですか？**経過時間ではなく、レンダリングや更新のティックで「進行」を計測する場合です。

• ゲームエンジン（Unity、Godot、Stride、MonoGame）— ロジックのティックがエンジンの更新ループに同期するため、一時停止やタイムスケールが反映され、レンダリングと同期した状態が保たれます。
• UI のレンダリングループ（WPF／Avalonia／WinUI のコンポジションフレーム）— フレームごとに反応します。
• 決定論的なテスト— FakeFrameProvider.Advance(n) は実時間をまったく進めずにフレームを進めます。FakeTimeProvider.Advance(timeSpan) がクロックを進めるのと同様です。

一般的なサーバーやビジネスロジックのコードで FrameProvider が必要になることはほとんどありません。その役割は TimeProvider の領域です。どちらのフェイクも、時間に依存するパイプラインを完全に決定論的にできます。例は scheduling-and-concurrency.md を参照してください。

---

ベストプラクティスのまとめ

推奨事項

• OnErrorResume をデフォルトとして扱い、個々のイベントに問題があってもストリームが存続するように設計します。
• エラーでストリームを終了させたい場合は、.OnErrorResumeAsFailure() を追加します。
• 各非同期オペレーターで AwaitOperation を意図的に選択します（最新の値を優先する場合は Switch、順序を保つ場合は Sequential、クールダウンには Drop）。
• 複数のスレッドから発行するソースの後には、Synchronize()／ObserveOn* を配置します。
• テストで FakeTimeProvider／FakeFrameProvider を使用できるよう、時間オペレーターには TimeProvider を、フレームオペレーターには FrameProvider を渡します。
• ライフタイムを管理します。サブスクリプションを DisposableBag、CompositeDisposable、または .AddTo(cancellationToken) に登録します。開発中は ObservableTracker を有効にして、リークを検出します。
• 重複を排除した監視可能な状態には ReactiveProperty を、XAML にバインドする状態には BindableReactiveProperty を使用します。

禁止事項

• 例外がストリームを終了させると思い込まないでください（それは Rx の動作であり、R3 では異なります）。
• R3 で名前が変更された Rx の名称をそのまま使わないでください。Throttle ではなく Debounce、Sample ではなく ThrottleLast、Buffer ではなく Chunk です。1.3.1 には Retry、GroupBy、Finally、通常の Buffer はありません。代替手段については差分ファイルを参照してください。
• Synchronize() を使わずに、複数のスレッドから同時または再入的に OnNext を呼び出さないでください。
• バックプレッシャーのあるスループットパイプラインに R3 を使用しないでください。Channels または Akka.NET Streams を使用してください。
• 終端オペレーターでブロッキングしないでください（.Result／.Wait()）。これらは Task<T> を返すため、await を使用します。

---

追加リソース

• R3 リポジトリ： https://github.com/Cysharp/R3
• README 全文／オペレーターリファレンス： https://github.com/Cysharp/R3/blob/main/README.md
• 設計の背景（neuecc）： https://neuecc.medium.com/r3-a-new-modern-reimplementation-of-reactive-extensions-for-c-cf29abcc5826
• NuGet： https://www.nuget.org/packages/R3
• TimeProvider（BCL）： https://learn.microsoft.com/en-us/dotnet/api/system.timeprovider
• FakeTimeProvider： https://www.nuget.org/packages/Microsoft.Extensions.TimeProvider.Testing
• 関連スキル： csharp-concurrency-patterns（R3、async/await、Channels、Akka.NET の選択）