R3は、Cysharpが開発しているC#向けのReactive Extensionsです。UniRxの作者(neuecc)が作り直したライブラリで、UniRxのリポジトリは2024年2月の最終コミットを最後にアーカイブされました。Unityで新しくRxを使うなら、選択肢はR3です。
この記事では、UnityへのR3の導入手順、Observableを作るファクトリーメソッド、用途別のオペレーター、UniRxからの名称変更、R3独自のエラー処理を順に扱います。対象は2026年5月19日公開のR3 1.3.1です。Unity以外で動く部分(オペレーターの挙動・エラー処理)は、.NET 10のコンソールアプリにR3 1.3.1を入れて実行した結果を載せています。
まとめ:Unity R3の使い方の要点
- 導入は2段階。コア本体の
R3をNuGetForUnityで入れ、Unity連携のR3.Unityをgit URLでPackage Managerから入れる。対応はUnity 2021.3以上。 - Observableは
Observable.Return・Range・Interval・EveryUpdateなどのファクトリーメソッドか、Subject・ReactivePropertyで作る。JustやFromIterableはR3に無い。 - UniRxからの主な改名は
Throttle→Debounce、Sample→ThrottleLast、Buffer→Chunk、StartWith→Prepend。 Selectなどの処理中の例外はOnErrorResumeに流れ、購読が続く。ただし、FromAsyncの処理失敗などは失敗完了になる。止めたいときはOnErrorResumeAsFailure()を挟む。Retryオペレーターは1.3.1に存在しない。- 購読は
AddTo(this)でGameObjectの破棄に紐づける。リークはエディターの Window > Observable Tracker で探す。
R3とは:UniRxの後継になった理由と現行版
R3の最新版は1.3.1(2026年5月19日公開・MITライセンス)です。1.3.1での変更はUnity 6.2以降で出るTreeViewの非推奨警告の修正と、NuGetパッケージへのREADME同梱などで、APIの追加はありません。コアはNuGetで配布され、対象は.NET Standard 2.0/2.1、.NET 6以上です。
UniRxとの違いは、書き味よりも内部の設計にあります。READMEが挙げている変更点のうち、Unityで使うときに効くのは次の4つです。
IObservable<T>を使わない:Observable<T>とObserver<T>を抽象クラスとして定義し直した。全購読をライブラリ側で追跡できるので、購読リークを後から探せる。- 例外で購読を止めない:通知は
OnNext・OnErrorResume・OnCompleted(Result)の3つ。従来のOnErrorで購読が切れる挙動をやめた。 ISchedulerを廃止:時間の抽象化は.NET 8で入ったTimeProviderに置き換えた。UnityではUnityTimeProviderがPlayerLoopに沿って動く。- フレーム単位の処理を標準装備:
FrameProviderを導入し、時間を受け取るオペレーターにはフレーム版(IntervalFrame・DelayFrame・DebounceFrameなど)がある。
R3はUnity専用ではありません。同じコアがGodot、WPF、Avalonia、WinForms、WinUI3、MAUI、Blazorなどでも動き、プラットフォームごとにTimeProviderとFrameProviderの実装が用意され、UIイベントの変換やライフサイクル連携などの専用APIも提供されています。同じCysharpのライブラリを.NETとUnityで共用する例にはZLinqとは?C#のゼロアロケーションLINQ|.NET・Unityでの導入と性能・制約もあります。
UnityへのR3導入手順(NuGetForUnity+R3.Unity)
R3はUnity 2021.3以上に対応しています(R3.Unityのpackage.jsonの "unity": "2021.3")。導入は「コア本体をNuGetから」「Unity連携部分をgit URLから」の2段階です。
- NuGetForUnityを入れる:Package Managerの「Add package from git URL」で
https://github.com/GlitchEnzo/NuGetForUnity.git?path=/src/NuGetForUnityを指定する(2026年10月時点の最新はv4.5.0)。 - R3本体を入れる:メニューの NuGet > Manage NuGet Packages で「R3」を検索してInstallする。依存の
Microsoft.Bcl.TimeProviderなどもまとめて入る。 - R3.Unityを入れる:Package Managerの「Add package from git URL」に次のURLを指定する。
https://github.com/Cysharp/R3.git?path=src/R3.Unity/Assets/R3.Unity#1.3.1
末尾の #1.3.1 はリリースタグで、付けないとmainブランチの先端を取り込みます。NuGet側のR3本体と版を揃えるため、タグは付けておく方が安全です。
導入でつまずきやすい点は3つあります。
- バージョン競合のエラー:依存DLLの版がぶつかったら、Project Settings > Player > Other Settings の「Assembly Version Validation」を外す(README記載の対処)。
- asmdefを使っている場合:自分のアセンブリ定義の参照に
R3.Unityを足す。R3.dll本体はプリコンパイル済みDLLなので、asmdefで「Override References」をオンにしていなければ自動で参照される。 - TextMeshProへの反映:
SubscribeToTextのTextMeshPro版は、com.unity.textmeshproが入っているか、uGUI 2.0.0以上(TextMeshProを統合したUnity 6の構成)のときだけ有効になる。
R3.Unityを入れると、起動時に ObservableSystem の既定値が自動で差し替わります。時間は UnityTimeProvider.Update、フレームは UnityFrameProvider.Update、未処理例外の出力先は Debug.LogException です。Time.timeScale = 0 でポーズすると、既定のTimeProviderを使う Interval や ThrottleFirst も止まります。ポーズ中も動かしたいUIには、UnityTimeProvider.UpdateIgnoreTimeScale を明示します。Unity本体の版ごとの変更はUnity 6.3 LTS(6000.3)の新機能と6.2からの変更点|サポート期限・対応OS・移行の注意点にまとめています。
最初のコード:ボタン連打防止とHP表示の連動
uGUIのButtonとTextを使う例です。EnemyView.csをGameObjectへ追加し、InspectorでattackButtonとhpTextへ対応するコンポーネントを割り当ててください。攻撃ボタンの連打を0.5秒に1回へ絞り、HPの変化をテキストへ反映し、HPが0になったらボタンを無効にします。
using System;
using R3;
using UnityEngine;
using UnityEngine.UI;
public class EnemyView : MonoBehaviour
{
[SerializeField] Button attackButton;
[SerializeField] Text hpText;
// インスペクターで値を確認・変更できる ReactiveProperty
public SerializableReactiveProperty<int> hp = new SerializableReactiveProperty<int>(100);
void Start()
{
attackButton.OnClickAsObservable()
.ThrottleFirst(TimeSpan.FromSeconds(0.5)) // 最初の1回を通し、0.5秒間は捨てる
.Subscribe(_ => hp.Value -= 10)
.AddTo(this); // このGameObjectの破棄で購読解除
hp.Select(x => $"HP: {x}")
.SubscribeToText(hpText)
.AddTo(this);
hp.Where(x => x <= 0)
.Take(1)
.Subscribe(_ => attackButton.interactable = false)
.AddTo(this);
}
}
ReactiveProperty<T> は [SerializeField] に使えないため、インスペクターに出すときは SerializableReactiveProperty<T> を使います。値は .Value で書き換え、同じ値を入れたときは通知されません。同じ値でも通知したいときは .OnNext(value) を呼びます(下の実行結果で確認済み)。読み取り専用で公開したいときは、privateの ReactiveProperty を ReadOnlyReactiveProperty 型のプロパティで返します。値の参照は CurrentValue です。
var rp = new ReactiveProperty<int>(0);
rp.Subscribe(x => Console.WriteLine($"rp {x}"));
rp.Value = 1;
rp.Value = 1; // 同じ値なので通知されない
rp.OnNext(1); // 強制的に通知する
// 出力(R3 1.3.1 / .NET 10で実行)
// rp 0
// rp 1
// rp 1
R3.Triggersを using すると、OnCollisionEnterAsObservable() のようにMonoBehaviourのメッセージもObservableとして受け取れます。値の変化を持つモデルとそれを表示するビューを分ける設計は、MVVMとは?仕組み・構成要素とMVC/MVPとの違いを実装例で解説のViewModelとViewの関係と同じです。
ファクトリーメソッド一覧:Observableの作り方
ファクトリーメソッドは Observable クラスの静的メソッドで、ストリームの起点を作ります。R3 1.3.1のREADMEに載っている全33種のうち、Unityでよく使うものを目的別に並べました。
| 目的 | R3のメソッド | 補足 |
|---|---|---|
| 値を1つ流す | Return / ReturnUnit |
RxJavaの just に相当 |
| 連番を流す | Range |
同期的に流れて完了する |
| 配列・リストを流す | ToObservable |
IEnumerable などから変換 |
| 時間で発火 | Timer / Interval |
TimeProviderに従う |
| フレームで発火 | EveryUpdate / IntervalFrame / TimerFrame / NextFrame |
FrameProviderに従う |
| 値の変化を監視 | EveryValueChanged |
毎フレームのポーリング |
| C#イベントを変換 | FromEvent / FromEventHandler |
デリゲートの変換が要る |
| 非同期処理を変換 | FromAsync |
単発の非同期はawaitが基本 |
| 購読時に生成 | Defer / Create |
遅延生成・手組み |
| 複数を合成 | Merge / Concat / Zip / CombineLatest / Race |
静的メソッド版 |
自分で値を流すなら、ファクトリーメソッドではなくSubject系を使います。R3のSubjectは Subject・BehaviorSubject・ReactiveProperty・ReplaySubject・ReplayFrameSubject の5種です。dotnet/reactiveと違い、R3のSubjectは Dispose() すると OnCompleted を流し、下流の購読をすべて解除します(通知させたくないときは Dispose(false))。
フレーム系のファクトリーを使うと、UniRxでコルーチンや Update() に書いていた処理を宣言的に書けます。
// FixedUpdateごとに、rbへinput * speedの力を加える
Observable.EveryUpdate(UnityFrameProvider.FixedUpdate)
.Subscribe(_ => rb.AddForce(input * speed))
.AddTo(this);
// timeScaleを無視して5秒ごとに発火(ポーズ中も動くUI向け)
Observable.Interval(TimeSpan.FromSeconds(5), UnityTimeProvider.UpdateIgnoreTimeScale)
.Subscribe(_ => RefreshClock())
.AddTo(this);
// プロパティの変化を監視(INotifyPropertyChangedが無い値でも使える)
Observable.EveryValueChanged(this, x => x.transform.position)
.Subscribe(pos => Debug.Log(pos))
.AddTo(this);
RxJava由来のメソッド名とR3での対応名
R3やUniRxの解説には、RxJavaの名前をそのまま載せたものがあります。次のメソッドはR3 1.3.1のREADMEの一覧にも、タグ1.3.1のソースにもありません。コピーしてもコンパイルエラーになります。
| RxJava等の名前 | R3での書き方 |
|---|---|
just |
Observable.Return |
fromIterable |
ToObservable() |
fromCallable |
Observable.Defer / FromAsync |
flatMap / map / filter |
SelectMany / Select / Where |
onErrorReturn / onErrorResumeNext |
OnErrorResumeAsFailure() + Catch |
retry |
オペレーターは無い。SelectAwait 内で再試行 |
window |
無い。まとめるなら Chunk |
主要オペレーターの用途別早見表
オペレーターは Observable<T> への拡張メソッドとして ObservableExtensions に定義されています。LINQと同じ名前が多く、Select と Where で書き始められます。
| 用途 | オペレーター |
|---|---|
| 変換 | Select / SelectMany / Scan / Pairwise / Cast |
| 絞り込み | Where / WhereNotNull / DistinctUntilChanged / Take / Skip / TakeUntil |
| 合成 | Merge / Concat / Zip / CombineLatest / WithLatestFrom / Switch |
| 時間 | Delay / Debounce / ThrottleFirst / ThrottleLast / Chunk / Timeout |
| フレーム | DelayFrame / DebounceFrame / ThrottleFirstFrame / ChunkFrame |
| 非同期 | SelectAwait / WhereAwait / SubscribeAwait |
| スレッド | ObserveOn / SubscribeOn / ObserveOnMainThread |
| 共有 | Share / Publish / Replay / RefCount |
| 単一値 | FirstAsync / LastAsync / ToArrayAsync(戻り値はTask) |
最後の行のとおり、1つの値を返すメソッドはすべて ***Async になり、Task<T> を返します。UniRxの First() のようにObservableを返す版はありません。
ThrottleFirst・ThrottleLast・Debounceの違い(実測)
3つとも短時間に集中した値を間引きますが、どの値をいつ流すかが違います。同じ入力(0秒・0.3秒・0.6秒・1.5秒・1.7秒に a〜e を流す)に1秒の窓を掛け、FakeTimeProvider で時間を100ミリ秒ずつ進めて出力時刻を記録しました。
| オペレーター | 出力(値@秒) | 向く用途 |
|---|---|---|
ThrottleFirst |
[email protected]、[email protected] | ボタン連打の防止 |
ThrottleLast |
[email protected]、[email protected] | センサー値の定期反映 |
Debounce |
[email protected] | 入力が止まってからの検索 |
ThrottleFirst は窓の最初の値をすぐ流し、窓の間は捨てます。ThrottleLast(UniRxの Sample)は窓の終わりに最後の値を流します。Debounce(UniRxの Throttle)は最後の入力から1秒間なにも来なかったときだけ流すため、入力間隔がいずれも1秒未満のこの例では、最後の e だけが2.7秒に出ています。
もう1つ、dotnet/reactiveの Sample との違いがあります。dotnet/reactiveの Sample は購読した時点からタイマーを回し続けます。R3の ThrottleFirst/ThrottleLast は値が来たときにタイマーを始め、窓が終わると止めます。上の表で ThrottleLast の2回目が2.0秒ではなく2.5秒なのはこのためです。値が来ない間はタイマーも動きません。
using R3;
// .NETコンソール用。事前に dotnet add package Microsoft.Extensions.TimeProvider.Testing を実行
using Microsoft.Extensions.Time.Testing;
var fake = new FakeTimeProvider();
var src = new Subject<string>();
src.ThrottleLast(TimeSpan.FromSeconds(1), fake)
.Subscribe(x => Console.WriteLine(x));
src.OnNext("a");
fake.Advance(TimeSpan.FromSeconds(0.6));
src.OnNext("c");
fake.Advance(TimeSpan.FromSeconds(0.4)); // 1秒経過 -> "c" が出る
SelectAwaitの非同期連結と実行中の入力制御
SelectAwait・WhereAwait・SubscribeAwait は非同期のラムダを受け取り、実行中に次の値が来たときの扱いを AwaitOperation で選べます。
| AwaitOperation | 実行中に次の値が来たとき |
|---|---|
Sequential(既定) |
キューに積み、完了後に順に実行 |
Drop |
新しい値を捨てる |
Switch |
実行中をキャンセルして新しい値で実行 |
Parallel |
すぐ並行実行(maxConcurrent で上限) |
SequentialParallel |
並行実行し、結果は入力順に流す |
ThrottleFirstLast |
実行中は最初と最後の値だけを通す |
通信ボタンの二重送信を防ぐなら Drop、検索窓の入力に追従させるなら Switch です。キャンセルは第2引数の CancellationToken で届くので、UnityWebRequest などへそのまま渡します。
UniRxからR3への名称変更と置き換え表
R3のREADMEは、dotnet/reactiveとUniRxからの名称変更を一覧にしています。UniRxのコードを移すときに引っかかるものを抜き出しました。
| UniRx | R3 |
|---|---|
Throttle / ThrottleFrame |
Debounce / DebounceFrame |
Sample / SampleFrame |
ThrottleLast / ThrottleLastFrame |
Buffer / BatchFrame |
Chunk / ChunkFrame |
StartWith |
Prepend |
Amb |
Race |
Distinct(selector) |
DistinctBy |
DistinctUntilChanged(selector) |
DistinctUntilChangedBy |
Finally |
Do(onDispose:) |
ObserveEveryValueChanged |
Observable.EveryValueChanged |
IReadOnlyReactiveProperty.Value |
ReadOnlyReactiveProperty.CurrentValue |
SkipLatestValueOnSubscribe() |
Skip(1) |
ToTask() / ToUniTask() |
FirstAsync() / LastAsync() |
IScheduler |
TimeProvider |
ReactiveCollection / ReactiveDictionary |
ObservableCollections.R3 |
MessageBroker |
MessagePipe |
Sample を別名として残さなかった理由もREADMEに書かれています。ThrottleLast と Sample が併存すると「何が違うのか」という質問が繰り返されるため、名前を ThrottleFirst と対になる形へ一本化したという説明です。移行でいちばん危ないのは Throttle です。R3にも ThrottleFirst/ThrottleLast があるので、名前の似たものへ機械的に置き換えると挙動が変わります。UniRxの Throttle は必ず Debounce へ置き換えてください。
Finally の置き換え先は、READMEの一覧では Do(onDisposed:) と書かれていますが、1.3.1の実装の引数名は onDispose です。READMEどおりに書くとコンパイルエラーになります(.NET 10でビルドして確認)。UniRxの ObjectPool は後継が無く、READMEはUniTaskで自作するよう案内しています。MainThreadDispatcher.OnApplicationQuitAsObservable は Unity 2022.2以降では Application.exitCancellationToken で置き換えられます。Unity 2021.3では Application.quitting イベントを使います。
エラー処理:OnErrorResumeで購読が止まらない仕様
R3で最も挙動が変わったのがエラー処理です。従来のRxでは、パイプラインで例外が出ると OnError が呼ばれて購読が切れました。R3では、次の例のように Select の処理中に出た例外は OnErrorResume に流れ、購読が続きます。ただし、FromAsync の処理失敗などは OnCompleted に失敗を渡して終了します。0で割る例外を途中で起こして確かめました。
var s = new Subject<int>();
s.Select(x => 10 / x).Subscribe(
x => Console.WriteLine($"OnNext {x}"),
ex => Console.WriteLine($"OnErrorResume {ex.GetType().Name}"),
r => Console.WriteLine($"OnCompleted IsFailure={r.IsFailure}"));
s.OnNext(5); s.OnNext(0); s.OnNext(2);
// 出力(R3 1.3.1 / .NET 10)
// OnNext 2
// OnErrorResume DivideByZeroException
// OnNext 5 ← 例外の後も購読が続いている
ゲームやUIのイベント処理では、1回の例外でボタンが二度と反応しなくなるより、こちらの方が扱いやすい挙動です。従来どおり最初の例外で止めたいときは、OnErrorResumeAsFailure() を挟みます。例外が OnCompleted(Result.Failure) に変わり、購読が終わります。
s.Select(x => 10 / x)
.OnErrorResumeAsFailure()
.Subscribe(
x => Console.WriteLine($"OnNext {x}"),
r => Console.WriteLine($"OnCompleted IsFailure={r.IsFailure}"));
// 5, 0, 2 を流したときの出力
// OnNext 2
// OnCompleted IsFailure=True ← 2は流れない
注意点は Catch です。Catch が捕まえるのは OnCompleted に載った失敗だけで、OnErrorResume は捕まえません。実行すると、Catch を直接つないだ側は代替値を出さずに OnErrorResume へ素通りし、OnErrorResumeAsFailure() の後ろにつないだ側だけが代替値 -1 を出して正常完了しました。「エラー時に既定値を返す」処理は、この2つを組み合わせて書きます。
R3 1.3.1には Retry オペレーターがありません。READMEは、非同期処理の単位で再試行するよう案内しています。パイプライン全体を購読し直すのではなく、SelectAwait のラムダの中で失敗した1回だけをやり直す書き方です。
button.OnClickAsObservable()
.SelectAwait(async (_, ct) =>
{
for (var i = 0; ; i++)
{
try
{
ct.ThrowIfCancellationRequested();
using var req = UnityWebRequest.Get(url);
await req.SendWebRequest().WithCancellation(ct);
return req.downloadHandler.text;
}
catch (Exception ex) when (i < 3 && ex is not OperationCanceledException && !ct.IsCancellationRequested)
{
// 3回までやり直す。4回目の失敗は OnErrorResume へ流れる
}
}
}, AwaitOperation.Drop)
.SubscribeToText(resultText)
.AddTo(this);
WithCancellation はUniTaskの拡張メソッドなので、この例はUniTaskの導入が前提です。必要な名前空間は System、R3、UnityEngine.Networking、Cysharp.Threading.Tasks です。どこでも処理されなかった例外は ObservableSystem の未処理例外ハンドラーに渡り、R3.Unityの既定では Debug.LogException でコンソールに出ます。
購読の解除とリーク検出(AddTo・Disposable・Observable Tracker)
Rxの購読は、Subscribe が返す IDisposable を破棄するか、上流から OnCompleted が届くまで残ります。Unityでは、GameObjectが消えた後もコールバックが走って MissingReferenceException になるのが典型的な事故です。
AddTo(this):購読をComponentまたはGameObjectの破棄に紐づける。一度もアクティブになっていないGameObjectはOnDestroyが呼ばれずdestroyCancellationTokenもキャンセルされないが、AddToはその場合にObservableDestroyTriggerを付けて確実に解除する。Disposable.CreateBuilder():複数の購読をまとめるためのビルダー。ビルダー自体は構造体だが、Build時の結合オブジェクトなどにはアロケーションが発生する。d.Build().AddTo(this)でGameObjectの破棄に紐づける。CompositeDisposable:途中で個別にRemoveしたいときの選択肢。スレッドセーフだが、READMEの比較では最も遅い。
var d = Disposable.CreateBuilder();
Observable.EveryUpdate().Subscribe(_ => Move()).AddTo(ref d);
hp.Subscribe(x => UpdateBar(x)).AddTo(ref d);
d.Build().AddTo(this); // ビルドしてGameObjectの破棄時に解除
解除漏れを探すには、エディターの Window > Observable Tracker を開きます。「Enable Tracking」をオンにすると生きている購読の一覧が出て、「Enable StackTrace」も付けると購読した場所のスタックトレースまで追えます。StackTraceの取得は負荷が高いとREADMEに書かれているので、リーク調査が終わったら両方オフに戻します。コード側では ObservableTracker.EnableTracking = true で同じ情報を取れます。
R3を採用すべきでない場面
R3の作者はREADMEで、R3が扱うのはメモリ内のイベント処理(LINQ to Events)に限ると明言しています。次のケースではR3以外を選ぶ方が素直です。
- 1回で終わる非同期処理:通信1回、アセットのロード1回などは、UniTaskや
async/awaitで書く。READMEの設計方針にも「単発の非同期処理はasync/awaitに任せるべき」とある。R3を使うのは、繰り返し発生するイベントを加工するときに限る。 - 流量制御が要るデータ処理:生産側が速すぎる場合の制御は、R3ではなく
IAsyncEnumerableやSystem.Threading.Channelsに任せる方針。考え方はバックプレッシャーとは?流量制御の仕組みと実装方式の選び分けを実装視点で解説で扱っている。 - 複数スレッドからOnNextを呼ぶ処理:R3の
Subject.OnNextはdotnet/reactiveと違ってスレッドセーフではない。ReactivePropertyも同様で、別スレッドから書き込むならSynchronizedReactivePropertyかロックが要る。ワーカースレッドの結果はObserveOnMainThread()でメインスレッドへ戻してからUIに触る。 - イベントが数個しかない小さなスクリプト:C#の
eventで足りる。R3の依存DLLとチーム内の学習コストに見合うのは、イベントの合成や時間での間引きが複数箇所に出てくるプロジェクト。
R3の考え方の土台になっているのはObserverパターンで、Subjectが通知元、Observerが購読者にあたります。パターンそのものはObserverパターンとは?Java・TypeScriptの実装例とPub/Subとの違いで、Swiftの同系統フレームワークはCombine(Swift)とは?Publisher・Subscriberの仕組みと2026年の採用判断を解説で解説しています。
よくある質問
UniRxとR3はどちらを使うべきですか?
新規ならR3です。UniRxのリポジトリはアーカイブされ、更新がありません。既存のUniRxプロジェクトを移す場合は、Throttle → Debounce などの改名と、例外で購読が止まらない仕様の違いを確認してから置き換えます。
R3の最新バージョンはいくつですか?
2026年10月時点の最新は1.3.1(2026年5月19日公開)です。NuGetのR3と、R3.Unityのgit URLに付けるタグ(#1.3.1)を同じ版に揃えます。最新の版は R3公式のReleases で確認してください。
Unity 2020以前でR3は使えますか?
使えません。R3.Unityのパッケージ定義が要求するのはUnity 2021.3以上です。2020以前のプロジェクトでは、UniRxを使い続けるかUnityを更新します。
ThrottleLastとSampleの違いは何ですか?
名称上の対応関係はありますが、時間指定版の発火時刻まで同じではありません。R3はdotnet/reactiveとUniRxの Sample を ThrottleLast に改名しました。ただし、R3版は値が来たときにタイマーを始めるので、購読直後からタイマーが回り続けるdotnet/reactiveの Sample とは出力の時刻がずれることがあります。
R3にRetryはありますか?
1.3.1にはありません。例外が出ても購読が止まらない設計なので、失敗した処理を SelectAwait の中で再試行します。パイプライン全体を止めてやり直したい場合は、OnErrorResumeAsFailure() で失敗として完了させ、購読を作り直します。