golang-samber-ro

ペルソナ： 非同期または無限にデータが流れる場合にリアクティブストリームを利用する Go エンジニアです。手動でゴルーチンやチャネルを接続する代わりに samber/ro を使って宣言的なパイプラインを構築しますが、シンプルなスライスと samber/lo で十分な場面も理解しています。

思考モード： 高度なリアクティブパイプラインの設計や、コールド／ホットオブザーバブル、サブジェクト、結合オペレーターの選択では、できる限り徹底的に検討してください。アーキテクチャを誤ると、リソースリークやイベントの取りこぼしにつながります。Claude Code では、ultrathink を使って拡張思考を明示的に有効にしてください。

samber/ro — Go 向けリアクティブストリーム

ReactiveX の Go 実装です。ジェネリクスを優先した、型安全で合成可能な非同期データストリーム用パイプラインを提供し、自動バックプレッシャー、エラー伝播、コンテキスト連携、リソースクリーンアップに対応します。150 以上のオペレーター、5 種類のサブジェクト、40 以上のプラグインを備えています。

公式リソース：

• github.com/samber/ro
• ro.samber.dev
• pkg.go.dev/github.com/samber/ro

このスキルは網羅的ではありません。詳しくはライブラリのドキュメントとコード例を参照してください。

• Go パッケージのドキュメント、シンボル、バージョン、インポーター、既知の脆弱性については、→ samber/cc-skills-golang@golang-pkg-go-dev スキル（godig）を参照してください。Go パッケージの情報については Context7 よりこちらが推奨されます。
• 独自のコード内でこのライブラリの使い方（定義、呼び出し箇所、診断情報）を調べるには、→ samber/cc-skills-golang@golang-gopls スキル（gopls）を参照してください。
• Context7 は、pkg.go.dev にインデックスされていないドキュメントのフォールバックとして引き続き利用できます。

samber/ro を選ぶ理由（ストリームとスライスの比較）

Go のチャネルとゴルーチンは、複雑な非同期パイプラインでは扱いにくくなります。手動でのチャネルクローズ、冗長なゴルーチンのライフサイクル管理、ネストされた select 間でのエラー伝播、合成可能なオペレーターがないことが課題です。samber/ro は、宣言的で連結可能なストリームオペレーターによってこれらを解決します。

用途に応じたツールの選び方：

シナリオ	ツール	理由
スライスの変換（map、filter、reduce）	samber/lo	有限かつ同期的で、即時評価されます。ストリームのオーバーヘッドは不要です
エラー処理を伴うシンプルなゴルーチンのファンアウト	errgroup	標準ライブラリで軽量です。並行数に上限がある場合は十分です
無限イベントストリーム（WebSocket、ティッカー、ファイル監視）	samber/ro	バックプレッシャー、リトライ、タイムアウト、結合に対応する宣言的なパイプラインです
複数の非同期ソースからのリアルタイムデータ拡充	samber/ro	CombineLatest/Zip により、手動で select を使わずに依存するストリームを合成できます
複数のコンシューマーで 1 つのソースを共有する pub/sub	samber/ro	ホットオブザーバブル（Share/Subjects）がマルチキャストにネイティブ対応します

主な違い：lo と ro

項目	samber/lo	samber/ro
データ	有限のスライス	無限のストリーム
実行	同期的、ブロッキング	非同期、ノンブロッキング
評価	即時（中間スライスを確保）	遅延（到着した項目を処理）
タイミング	即時	時間を考慮（delay、throttle、interval、timeout）
エラーモデル	呼び出しごとに (T, error) を返す	エラーチャネルを通じてパイプライン全体に伝播
用途	コレクションの変換	イベント駆動、リアルタイム、非同期パイプライン

インストール

go get github.com/samber/ro

基本概念

4 つの構成要素：

1. オブザーバブル — 時間の経過とともに値を発行するデータソースです。デフォルトではコールドで、各サブスクライバーが最初から独立した実行を開始します
2. オブザーバー — 3 つのコールバック onNext(T)、onError(error)、onComplete() を持つコンシューマーです
3. オペレーター — オブザーバブルを別のオブザーバブルに変換する関数です。Pipe で連結します
4. サブスクリプション — オブザーバブルとオブザーバーの接続です。ブロックするには .Wait()、キャンセルするには .Unsubscribe() を呼び出します

observable := ro.Pipe2(
    ro.RangeWithInterval(0, 5, 1*time.Second),
    ro.Filter(func(x int) bool { return x%2 == 0 }),
    ro.Map(func(x int) string { return fmt.Sprintf("even-%d", x) }),
)

observable.Subscribe(ro.NewObserver(
    func(s string) { fmt.Println(s) },      // onNext
    func(err error) { log.Println(err) },    // onError
    func() { fmt.Println("Done!") },         // onComplete
))
// Output: "even-0", "even-2", "even-4", "Done!"

// Or collect synchronously:
values, err := ro.Collect(observable)

コールドオブザーバブルとホットオブザーバブル

コールド（デフォルト）：.Subscribe() を呼び出すたびに、新しい独立した実行が開始されます。安全で予測しやすいため、デフォルトではこちらを使用してください。

ホット：複数のサブスクライバーが単一の実行を共有します。ソースが高コストな場合（WebSocket、データベースのポーリング）や、サブスクライバー全員が同じイベントを確認する必要がある場合に使用します。

変換方法	動作
Share()	コールドからホットに変換し、参照カウントを行います。最後のサブスクライバーが購読解除すると、実行を終了します
ShareReplay(n)	Share() と同じですが、後から購読したサブスクライバー向けに最後の N 個の値をバッファーします
Connectable()	コールドからホットに変換しますが、明示的な .Connect() の呼び出しを待ちます
Subject	ネイティブでホットです。.Send()、.Error()、.Complete() を直接呼び出します

Subject	コンストラクター	リプレイ動作
PublishSubject	NewPublishSubject[T]()	なし。後から購読したサブスクライバーは過去のイベントを受け取りません
BehaviorSubject	NewBehaviorSubject[T](initial)	新しいサブスクライバーに最後の値をリプレイします
ReplaySubject	NewReplaySubject[T](bufferSize)	最後の N 個の値をリプレイします
AsyncSubject	NewAsyncSubject[T]()	完了時にのみ、最後の値を送出します
UnicastSubject	NewUnicastSubject[T](bufferSize)	サブスクライバーは 1 つのみです

Subject の詳細とホットオブザーバブルのパターンについては、Subjects Guide を参照してください。

オペレーター早見表

カテゴリー	主なオペレーター	目的
作成	Just、FromSlice、FromChannel、Range、Interval、Defer、Future	さまざまなソースからオブザーバブルを作成します
変換	Map、MapErr、FlatMap、Scan、Reduce、GroupBy	ストリームの値を変換または蓄積します
フィルター	Filter、Take、TakeLast、Skip、Distinct、Find、First、Last	値を選択的に送出します
結合	Merge、Concat、Zip2–Zip6、CombineLatest2–CombineLatest5、Race	複数のオブザーバブルを結合します
エラー	Catch、OnErrorReturn、OnErrorResumeNextWith、Retry、RetryWithConfig	エラーから回復します
タイミング	Delay、DelayEach、Timeout、ThrottleTime、SampleTime、BufferWithTime	値を送出するタイミングを制御します
副作用	Tap／Do、TapOnNext、TapOnError、TapOnComplete	ストリームを変更せずに監視します
終端	Collect、ToSlice、ToChannel、ToMap	ストリームを Go の型に変換して消費します

オペレーターチェーン全体でコンパイル時の型安全性を確保するには、型付きの Pipe2、Pipe3 ... Pipe25 を使用してください。型なしの Pipe は any を使用するため、型チェックが失われます。

シグネチャ付きの完全なオペレーター一覧（150 以上）については、Operators Guide を参照してください。

よくある間違い

間違い	失敗する理由	対処法
エラーハンドラーなしで ro.OnNext() を使用する	エラーが黙って破棄され、本番環境でバグが隠れてしまいます	3 つすべてのコールバックを指定して ro.NewObserver(onNext, onError, onComplete) を使用します
Pipe2／Pipe3 の代わりに型なしの Pipe() を使用する	コンパイル時の型安全性が失われ、実行時にエラーが表面化します	型付きオペレーターチェーンには Pipe2、Pipe3 ... Pipe25 を使用します
無限ストリームで .Unsubscribe() を忘れる	ゴルーチンリークが発生し、オブザーバブルが永遠に実行されます	TakeUntil(signal)、コンテキストのキャンセル、または明示的な Unsubscribe() を使用します
コールドで十分な場合に Share() を使用する	不要な複雑さが加わり、ライフサイクルを把握しにくくなります	複数のコンシューマーが同じストリームを必要とする場合にのみ、ホットオブザーバブルを使用します
有限のスライス変換に samber/ro を使用する	同期処理にストリームのオーバーヘッド（ゴルーチン、サブスクリプション）が発生します	samber/lo を使用します。よりシンプルで高速な、スライス専用のライブラリです
キャンセルのためにコンテキストを伝播しない	ストリームがシャットダウンシグナルを無視し、終了時にリソースリークが発生します	パイプラインに ContextWithTimeout または ThrowOnContextCancel をチェーンします

ベストプラクティス

1. 3 つすべてのイベントを必ず処理してください — OnNext だけではなく、NewObserver(onNext, onError, onComplete) を使用してください。未処理のエラーは、気付かないうちにデータが失われる原因になります
2. 同期的に消費する場合は Collect() を使用してください — ストリームが有限で []T が必要な場合、Collect は完了までブロックし、スライスとエラーを返します
3. 型付きの Pipe 関数を優先してください — Pipe2、Pipe3…Pipe25 はコンパイル時に型の不一致を検出します。型なしの Pipe は、動的なオペレーターの連鎖にのみ使用してください
4. 無限ストリームには上限を設けてください — Take(n)、TakeUntil(signal)、Timeout(d)、またはコンテキストのキャンセルを使用してください。上限のないストリームはゴルーチンのリークにつながります
5. 可観測性には Tap／Do を使用してください — ストリームを変更せずに、値の出力をログ記録、トレース、計測できます。エラー監視には TapOnError をチェーンしてください
6. 単純な変換には samber/lo を優先してください — データが有限のスライスで、Map／Filter／Reduce が必要な場合は lo を使用してください。データが時間とともに到着する場合、複数のソースから届く場合、または再試行／タイムアウト／バックプレッシャーが必要な場合は ro を使用してください

プラグインエコシステム

40 以上のプラグインが、ドメイン固有のオペレーターで ro を拡張します。

カテゴリ	プラグイン	インポートパスのプレフィックス
エンコーディング	JSON、CSV、Base64、Gob	plugins/encoding/...
ネットワーク	HTTP、I/O、FSNotify	plugins/http、plugins/io、plugins/fsnotify
スケジューリング	Cron、ICS	plugins/cron、plugins/ics
可観測性	Zap、Slog、Zerolog、Logrus、Sentry、Oops	plugins/observability/...、plugins/samber/oops
レート制限	Native、Ulule	plugins/ratelimit/...
データ	Bytes、Strings、Sort、Strconv、Regexp、Template	plugins/bytes、plugins/strings など
システム	Process、Signal	plugins/proc、plugins/signal

インポートパスと使用例を含むすべてのプラグイン一覧は、プラグインエコシステムを参照してください。

再試行＋タイムアウト、WebSocket のファンアウト、グレースフルシャットダウン、ストリームの結合など、実際のリアクティブパターンについては、パターンを参照してください。

samber/ro でバグや予期しない動作を見つけた場合は、github.com/samber/ro/issuesで issue を作成してください。

相互参照

• → 有限スライスの変換（Map、Filter、Reduce、GroupBy）については、samber/cc-skills-golang@golang-samber-lo スキルを参照してください。データがすでにスライスにある場合は lo を使用します
• → ro のパイプラインと組み合わせられるモナド型（Option、Result、Either）については、samber/cc-skills-golang@golang-samber-mo スキルを参照してください
• → インメモリキャッシュについては、samber/cc-skills-golang@golang-samber-hot スキルを参照してください（ro プラグインとしても利用できます）
• → リアクティブストリームが過剰な場合のゴルーチン／チャネルパターンについては、samber/cc-skills-golang@golang-concurrency スキルを参照してください
• → 本番環境でリアクティブパイプラインを監視する方法については、samber/cc-skills-golang@golang-observability スキルを参照してください