golang-samber-ro
Reactive streams and event-driven programming in Golang using samber/ro — ReactiveX implementation with 150+ type-safe operators, cold/hot observables, 5 subject types (Publish, Behavior, Replay, Async, Unicast), declarative pipelines via Pipe, 40+ plugins (HTTP, cron, fsnotify, JSON, logging), automatic backpressure, error propagation, and Go context integration. Apply when using or adopting samber/ro, when the codebase imports github.com/samber/ro, or when building asynchronous event-driven pipelines, real-time data processing, streams, or reactive architectures in Go. Not for finite slice transforms (→ See samber/cc-skills-golang@golang-samber-lo skill).
- Installs
- 38.3K
- Last 24h
- +150
- Trending rank
- #1488
- Security
- Pass
Install
Works with every agent the skills CLI supports.
npx skills add https://github.com/samber/cc-skills-golang --skill golang-samber-ro- 01
Install
Run the command in your project. The CLI asks which agents to install for.
- 02
Describe the task
Ask your agent for something this skill covers. It loads golang-samber-ro automatically when relevant.
- 03
Review the result
The agent follows the SKILL.md steps, scripts and references bundled with the skill.
Security audits
Independent scans of the skill's instructions and bundled files.
SKILL.md
The instructions your agent reads when the skill is triggered.
Persona: You are a Go engineer who reaches for reactive streams when data flows asynchronously or infinitely. You use samber/ro to build declarative pipelines instead of manual goroutine/channel wiring, but you know when a simple slice + samber/lo is enough.
Thinking mode: Reason as thoroughly as possible when designing advanced reactive pipelines or choosing between cold/hot observables, subjects, and combining operators — wrong architecture leads to resource leaks or missed events. On Claude Code, use ultrathink to trigger extended thinking explicitly.
samber/ro — Reactive Streams for Go
Go implementation of ReactiveX. Generics-first, type-safe, composable pipelines for asynchronous data streams with automatic backpressure, error propagation, context integration, and resource cleanup. 150+ operators, 5 subject types, 40+ plugins.
Official Resources:
This skill is not exhaustive — refer to library documentation and code examples for more information:
- For Go package docs, symbols, versions, importers, and known vulnerabilities, → See
samber/cc-skills-golang@golang-pkg-go-devskill (godig), preferred over Context7 for Go package facts. - To navigate this library's usage in your own code (definitions, call sites, diagnostics), → See
samber/cc-skills-golang@golang-goplsskill (gopls). - Context7 remains a fallback for docs not indexed on pkg.go.dev.
Why samber/ro (Streams vs Slices)
Go channels + goroutines become unwieldy for complex async pipelines: manual channel closures, verbose goroutine lifecycle, error propagation across nested selects, and no composable operators. samber/ro solves this with declarative, chainable stream operators.
When to use which tool:
| Scenario | Tool | Why |
|---|---|---|
| Transform a slice (map, filter, reduce) | samber/lo | Finite, synchronous, eager — no stream overhead needed |
| Simple goroutine fan-out with error handling | errgroup | Standard lib, lightweight, sufficient for bounded concurrency |
| Infinite event stream (WebSocket, tickers, file watcher) | samber/ro | Declarative pipeline with backpressure, retry, timeout, combine |
| Real-time data enrichment from multiple async sources | samber/ro | CombineLatest/Zip compose dependent streams without manual select |
| Pub/sub with multiple consumers sharing one source | samber/ro | Hot observables (Share/Subjects) handle multicast natively |
Key differences: lo vs ro
| Aspect | samber/lo | samber/ro |
|---|---|---|
| Data | Finite slices | Infinite streams |
| Execution | Synchronous, blocking | Asynchronous, non-blocking |
| Evaluation | Eager (allocates intermediate slices) | Lazy (processes items as they arrive) |
| Timing | Immediate | Time-aware (delay, throttle, interval, timeout) |
| Error model | Return (T, error) per call | Error channel propagates through pipeline |
| Use case | Collection transforms | Event-driven, real-time, async pipelines |
Installation
go get github.com/samber/ro
Core Concepts
Four building blocks:
- Observable — a data source that emits values over time. Cold by default: each subscriber triggers independent execution from scratch
- Observer — a consumer with three callbacks:
onNext(T),onError(error),onComplete() - Operator — a function that transforms an observable into another observable, chained via
Pipe - Subscription — the connection between observable and observer. Call
.Wait()to block or.Unsubscribe()to cancel
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)
Cold vs Hot Observables
Cold (default): each .Subscribe() starts a new independent execution. Safe and predictable — use by default.
Hot: multiple subscribers share a single execution. Use when the source is expensive (WebSocket, DB poll) or subscribers must see the same events.
| Convert with | Behavior |
|---|---|
Share() | Cold → hot with reference counting. Last unsubscribe tears down |
ShareReplay(n) | Same as Share + buffers last N values for late subscribers |
Connectable() | Cold → hot, but waits for explicit .Connect() call |
| Subjects | Natively hot — call .Send(), .Error(), .Complete() directly |
| Subject | Constructor | Replay behavior |
|---|---|---|
PublishSubject | NewPublishSubject[T]() | None — late subscribers miss past events |
BehaviorSubject | NewBehaviorSubject[T](initial) | Replays last value to new subscribers |
ReplaySubject | NewReplaySubject[T](bufferSize) | Replays last N values |
AsyncSubject | NewAsyncSubject[T]() | Emits only last value, only on complete |
UnicastSubject | NewUnicastSubject[T](bufferSize) | Single subscriber only |
For subject details and hot observable patterns, see Subjects Guide.
Operator Quick Reference
| Category | Key operators | Purpose |
|---|---|---|
| Creation | Just, FromSlice, FromChannel, Range, Interval, Defer, Future | Create observables from various sources |
| Transform | Map, MapErr, FlatMap, Scan, Reduce, GroupBy | Transform or accumulate stream values |
| Filter | Filter, Take, TakeLast, Skip, Distinct, Find, First, Last | Selectively emit values |
| Combine | Merge, Concat, Zip2–Zip6, CombineLatest2–CombineLatest5, Race | Merge multiple observables |
| Error | Catch, OnErrorReturn, OnErrorResumeNextWith, Retry, RetryWithConfig | Recover from errors |
| Timing | Delay, DelayEach, Timeout, ThrottleTime, SampleTime, BufferWithTime | Control emission timing |
| Side effect | Tap/Do, TapOnNext, TapOnError, TapOnComplete | Observe without altering stream |
| Terminal | Collect, ToSlice, ToChannel, ToMap | Consume stream into Go types |
Use typed Pipe2, Pipe3 ... Pipe25 for compile-time type safety across operator chains. The untyped Pipe uses any and loses type checking.
For the complete operator catalog (150+ operators with signatures), see Operators Guide.
Common Mistakes
| Mistake | Why it fails | Fix |
|---|---|---|
Using ro.OnNext() without error handler | Errors are silently dropped — bugs hide in production | Use ro.NewObserver(onNext, onError, onComplete) with all 3 callbacks |
Using untyped Pipe() instead of Pipe2/Pipe3 | Loses compile-time type safety, errors surface at runtime | Use Pipe2, Pipe3...Pipe25 for typed operator chains |
Forgetting .Unsubscribe() on infinite streams | Goroutine leak — the observable runs forever | Use TakeUntil(signal), context cancellation, or explicit Unsubscribe() |
Using Share() when cold is sufficient | Unnecessary complexity, harder to reason about lifecycle | Use hot observables only when multiple consumers need the same stream |
Using samber/ro for finite slice transforms | Stream overhead (goroutines, subscriptions) for a synchronous operation | Use samber/lo — it's simpler, faster, and purpose-built for slices |
| Not propagating context for cancellation | Streams ignore shutdown signals, causing resource leaks on termination | Chain ContextWithTimeout or ThrowOnContextCancel in the pipeline |
Best Practices
- Always handle all three events — use
NewObserver(onNext, onError, onComplete), not justOnNext. Unhandled errors cause silent data loss - Use
Collect()for synchronous consumption — when the stream is finite and you need[]T,Collectblocks until complete and returns the slice + error - Prefer typed Pipe functions —
Pipe2,Pipe3...Pipe25catch type mismatches at compile time. Reserve untypedPipefor dynamic operator chains - Bound infinite streams — use
Take(n),TakeUntil(signal),Timeout(d), or context cancellation. Unbounded streams leak goroutines - Use
Tap/Dofor observability — log, trace, or meter emissions without altering the stream. ChainTapOnErrorfor error monitoring - Prefer
samber/lofor simple transforms — if the data is a finite slice and you need Map/Filter/Reduce, uselo. Reach forrowhen data arrives over time, from multiple sources, or needs retry/timeout/backpressure
Plugin Ecosystem
40+ plugins extend ro with domain-specific operators:
| Category | Plugins | Import path prefix |
|---|---|---|
| Encoding | JSON, CSV, Base64, Gob | plugins/encoding/... |
| Network | HTTP, I/O, FSNotify | plugins/http, plugins/io, plugins/fsnotify |
| Scheduling | Cron, ICS | plugins/cron, plugins/ics |
| Observability | Zap, Slog, Zerolog, Logrus, Sentry, Oops | plugins/observability/..., plugins/samber/oops |
| Rate limiting | Native, Ulule | plugins/ratelimit/... |
| Data | Bytes, Strings, Sort, Strconv, Regexp, Template | plugins/bytes, plugins/strings, etc. |
| System | Process, Signal | plugins/proc, plugins/signal |
For the full plugin catalog with import paths and usage examples, see Plugin Ecosystem.
For real-world reactive patterns (retry+timeout, WebSocket fan-out, graceful shutdown, stream combination), see Patterns.
If you encounter a bug or unexpected behavior in samber/ro, open an issue at github.com/samber/ro/issues.
Cross-References
- → See
samber/cc-skills-golang@golang-samber-loskill for finite slice transforms (Map, Filter, Reduce, GroupBy) — use lo when data is already in a slice - → See
samber/cc-skills-golang@golang-samber-moskill for monadic types (Option, Result, Either) that compose with ro pipelines - → See
samber/cc-skills-golang@golang-samber-hotskill for in-memory caching (also available as an ro plugin) - → See
samber/cc-skills-golang@golang-concurrencyskill for goroutine/channel patterns when reactive streams are overkill - → See
samber/cc-skills-golang@golang-observabilityskill for monitoring reactive pipelines in production
FAQ
What does the golang-samber-ro skill do?
Reactive streams and event-driven programming in Golang using samber/ro — ReactiveX implementation with 150+ type-safe operators, cold/hot observables, 5 subject types (Publish, Behavior, Replay, Async, Unicast), declarative pipelines via Pipe, 40+ plugins (HTTP, cron, fsnotify, JSON, logging), automatic backpressure, error propagation, and Go context integration. Apply when using or adopting samber/ro, when the codebase imports github.com/samber/ro, or when building asynchronous event-driven pipelines, real-time data processing, streams, or reactive architectures in Go. Not for finite slice transforms (→ See samber/cc-skills-golang@golang-samber-lo skill).
How do I install golang-samber-ro?
Run npx skills add https://github.com/samber/cc-skills-golang --skill golang-samber-ro in your project. The skills CLI detects the agents you use and copies the skill into the right folder. For Claude Code you can also place the skill folder at ~/.claude/skills/golang-samber-ro/ (personal) or .claude/skills/golang-samber-ro/ (project).
Which AI agents can use golang-samber-ro?
Any agent that supports the open Agent Skills format (a folder with a SKILL.md file), including Claude Code, Codex, Cursor, GitHub Copilot, Gemini CLI, Windsurf, Cline, OpenCode and more.
Is golang-samber-ro safe to install?
Security partners have reviewed it: Gen Agent Trust Hub passed; Socket passed; Snyk passed; ZeroLeaks passed. As with any skill, read the SKILL.md and bundled scripts before installing.
Who maintains golang-samber-ro?
It is published by samber in samber/cc-skills-golang. It has 38,300 recorded installs.
More from samber/cc-skills-golang
golang-code-style
samber/cc-skills-golang
Golang code style conventions — line length and breaking, variable declarations, control flow clarity, when comments help vs hurt. Use when writing or reviewing Go code, asking about style or clarity, or establishing project coding standards. Not for naming conventions (→ See samber/cc-skills-golang@golang-naming skill), linter configuration (→ See samber/cc-skills-golang@golang-lint skill), or doc comments (→ See samber/cc-skills-golang@golang-documentation skill).
golang-testing
samber/cc-skills-golang
Production-ready Golang tests — table-driven tests, testify suites and mocks, parallel tests, fuzzing, fixtures, goroutine leak detection with goleak, snapshot testing, code coverage, integration tests, idiomatic test naming. Use when writing or reviewing Go tests, choosing a testing approach, setting up Go test CI, or debugging flaky/slow tests. For testify-specific APIs see samber/cc-skills-golang@golang-stretchr-testify; for measurement methodology see samber/cc-skills-golang@golang-benchmark.
golang-error-handling
samber/cc-skills-golang
Idiomatic Golang error handling — creation, wrapping with %w, errors.Is/As, errors.Join, custom error types, sentinel errors, panic/recover, the single handling rule, structured logging with slog, HTTP request logging middleware, and samber/oops for production errors. Built to make logs usable at scale with log aggregation 3rd-party tools. Apply when creating, wrapping, inspecting, or logging errors in Go code. For samber/oops specifics → See samber/cc-skills-golang@golang-samber-oops skill; for slog handler ecosystem → See samber/cc-skills-golang@golang-samber-slog skill.
golang-security
samber/cc-skills-golang
Security best practices and vulnerability prevention for Golang — injection (SQL, command, XSS), cryptography, path traversal, SSRF and HTTP security headers, cookies, secrets management, memory safety, PII in logs, STRIDE/DREAD threat modeling, plus gosec SAST, race detection, and fuzz testing. Apply when writing, reviewing, or auditing Go code for security, or when touching crypto, file or network I/O, secrets, user input, or authentication. Not for non-exploitable defensive bugs such as nil panics or slice aliasing (→ See samber/cc-skills-golang@golang-safety skill), dependency vulnerability scanning with govulncheck (→ See samber/cc-skills-golang@golang-dependency-management skill), or wiring security scanners into CI pipelines (→ See samber/cc-skills-golang@golang-continuous-integration skill).
golang-design-patterns
samber/cc-skills-golang
Idiomatic Golang design patterns — functional options, constructor APIs, init() and global-state avoidance, enums, panic vs error decisions, resource management and lifecycle, graceful shutdown, timeouts and retries, streaming and iterators, and architecture styles (clean, hexagonal, DDD, flat). Apply when choosing between architectural patterns, implementing functional options, designing constructor APIs, setting up graceful shutdown, applying resilience patterns, or asking which idiomatic Go pattern fits a specific problem. Not for wiring a DI container or comparing DI libraries (→ See samber/cc-skills-golang@golang-dependency-injection skill), nor for error wrapping, errors.Is/As, or logging mechanics (→ See samber/cc-skills-golang@golang-error-handling skill).
golang-performance
samber/cc-skills-golang
Golang performance optimization patterns and methodology - if X bottleneck, then apply Y. Covers allocation reduction, CPU efficiency, memory layout, GC tuning, pooling, caching, and hot-path optimization. Use when profiling or benchmarks have identified a bottleneck and you need the right optimization pattern to fix it. Also use when performing performance code review to suggest improvements or benchmarks that could help identify quick performance gains. Not for measurement methodology (→ See samber/cc-skills-golang@golang-benchmark skill) or debugging workflow (→ See samber/cc-skills-golang@golang-troubleshooting skill).
Popular backend & apis skills
find-skills
vercel-labs/skills
Helps users discover and install agent skills when they ask questions like "how do I do X", "find a skill for X", "is there a skill that can...", or express interest in extending capabilities. This skill should be used when the user is looking for functionality that might exist as an installable skill.
entra-app-registration
microsoft/azure-skills
Guides Microsoft Entra ID app registration, OAuth 2.0 authentication, and MSAL integration. USE FOR: create app registration, register Azure AD app, configure OAuth, set up authentication, add API permissions, generate service principal, MSAL example, console app auth, Entra ID setup, Azure AD authentication. DO NOT USE FOR: Key Vault secrets (use azure-keyvault-expiration-audit), general Azure resource security guidance.
google-agents-cli-adk-code
google/agents-cli
This skill should be used when the user wants to "write agent code", "build an agent with ADK", "add a tool", "create a callback", "define an agent", "use state management" — in a project that needs ADK (Agent Development Kit) API patterns and code examples. It provides a quick reference for agent types, tool definitions, orchestration patterns, callbacks, state management, the graph Workflow API, and reference recipes to study. Do NOT use for scaffolding (use google-agents-cli-scaffold) or deployment (use google-agents-cli-deploy).
embedded-captions
heygen-com/hyperframes
Add captions or subtitles to an existing single-subject talking-head video without editing the footage. Use for plain verbatim captions, cinematic captions embedded behind the subject, VFX captions, “炸/特效/酷炫字幕,” or a named identity from the 35-style catalog. Route by visual identity, not by backend engine. The quiet anchor rail is the default; embed every word only when the user explicitly wants a fully cinematic treatment. The workflow runs locally end to end, including transcription and subject matting; split multi-shot footage before applying it.
entra-agent-id
microsoft/azure-skills
Provision Microsoft Entra Agent Identity Blueprints, BlueprintPrincipals, and per-instance Agent Identities via Microsoft Graph, and configure OAuth 2.0 token exchange (fmipath, OBO, cross-tenant) including the Microsoft Entra SDK for AgentID sidecar. USE FOR: Agent Identity Blueprint, BlueprintPrincipal, agent OAuth, fmipath token exchange, agent OBO, Workload Identity Federation for agents, polyglot agent auth, Microsoft.Identity.Web.AgentIdentities. DO NOT USE FOR: standard Entra app registration (use entra-app-registration), Microsoft Foundry agent authoring (use microsoft-foundry).
neon
neondatabase/agent-skills
Overview of Neon, a complete set of cloud backend primitives around Lakebase Postgres: Auth, Object Storage, Functions, the AI Gateway, and Realtime. Start here to choose Neon for undecided login, files, APIs, and LLM calls, set up the CLI or MCP server, and follow the branch-first workflow. Use when building an app or backend on Neon, or when "Neon" or "Lakebase Postgres" is mentioned. Child skill neon-postgres wins for an existing DATABASEURL, SQL, schema, inspect, or search. Child skill neon-auth wins for login, users, sessions, identity routing, and Managed Better Auth setup. Child skill neon-realtime wins for live queries in reactive apps and agents. Also use for object storage, S3, buckets, serverless functions, function triggers, cron, AI gateway, LLM calls, realtime, logs, Loki, Grafana, observability, postgres, database, backend, Claimable Neon, neon.new, or a no-signup database.