RxGo 贡献指南：为 Reactive Extensions for Go 开发高质量 Operator 的完整规范

RxGo 是 ReactiveX 官方 API 的 Go 语言实现，目前仓库中已包含 80+ 个 Operator 与 250+ 个单元测试（见 CONTRIBUTING.md）。本文基于项目官方贡献指南展开，系统讲解在 RxGo 中新增或改进 Operator 时必须遵循的开发规范：从一致性审查、边界情况处理、单元测试、时间相关输入的确定性测试，到代码格式化与提交流程。读完本文，你将掌握一套可直接落地的 Operator 开发流程，并能理解 observable()/single()/optionalSingle() 等核心工具函数与 timeCausality() 确定性测试机制的底层原理。

项目背景：Operator 是 RxGo 的灵魂

RxGo 以 Go 的 channel 与 goroutine 原语实现 ReactiveX 概念，其核心抽象是 Iterable——一个可通过 Observe(opts ...Option) <-chan Item 观察的对象。每个 Operator 都是一段转换流水线阶段：默认顺序执行，也可通过多实例（goroutine）并行化。项目架构概览可见 README 中的示例（README.md）：

• 用 Just 基于静态列表创建 Observable；
• 用 Map 对每个元素做变换；
• 用 Filter 过滤不符合条件的元素；
• 最终结果发送到 channel 供消费者读取。

理解"每个 Operator 是一个转换阶段、默认顺序执行、可并行多实例"这一模型，是读懂贡献指南的前提——指南中反复强调的"边界情况"（eager/lazy、顺序/并行、错误策略）全部建立在这套模型之上。

一致性优先：先看现有实现，不要重复造轮子

贡献指南的第一条准则是 Consistency（一致性）。仓库已有 80+ Operator 和 250+ 单元测试，在动手写新代码前：

1. 查阅 README.md 中完整的 Operator 分类清单（Creating / Transforming / Filtering / Combining / Error Handling / Utility / Conditional / Mathematical / Converting 九大类），确认你要实现的功能是否已存在。
2. 阅读 doc/ 目录下对应 Operator 的文档，例如 map.md、window.md，这些文档按"创建 Observable / 转换 / 过滤 / 组合 / 错误处理"等主题组织，是理解现有实现约定的第一手资料。
3. 直接阅读对应源码。所有 Operator 的实现集中在 observable_operator.go 与 observable.go 中，如 WindowWithTime、Debounce 等复杂 Operator 都遵循统一的 operator 接口。

从源码结构看，每个 Operator 都实现统一的 operator 接口（observable.go）：

type operator interface {
	next(ctx context.Context, item Item, dst chan<- Item, operatorOptions operatorOptions)
	err(ctx context.Context, item Item, dst chan<- Item, operatorOptions operatorOptions)
	end(ctx context.Context, dst chan<- Item)
	gatherNext(ctx context.Context, item Item, dst chan<- Item, operatorOptions operatorOptions)
}

新 Operator 必须实现这四个方法，保持一致的行为约定，否则下游流水线无法正确处理数据流。写代码之前先搞清楚"当前实现如何解决常见问题"，是维持代码库统一风格的最有效手段。

边界情况：三个核心工具函数与参考实现

开发新 Operator 时会遇到大量边界情况，贡献指南明确列出三类最典型的：

• eager/lazy observation（急切/惰性观察）：默认惰性——Operator 只在被订阅后才处理上游数据；Eager 模式下则一旦产生 item 就立即处理（rxgo.WithObservationStrategy(rxgo.Eager)）。
• sequential vs parallel（顺序 vs 并行）：默认每个 Operator 是单个 goroutine 顺序执行；可用 WithPool(n) 创建 n 个 goroutine 并发消费，或用 WithCPUPool() 按逻辑 CPU 数创建。
• on error strategy（错误处理策略）：默认 Observable 遇到错误即停止（StopOnError），可配置为继续（ContinueOnError）等策略。

为了统一处理这些复杂逻辑，仓库提供了三个底层工具函数（定义于 observable.go）：

函数	签名	返回类型	适用场景
observable	(parent context.Context, iterable Iterable, operatorFactory func() operator, forceSeq, bypassGather bool, opts ...Option) Observable（observable.go）	Observable	发射 0 或多个 item
single	同左（observable.go）	Single	恰好发射 1 个 item
optionalSingle	同左（observable.go）	OptionalSingle	发射 0 或 1 个 item

以 observable() 为例，其内部逻辑（observable.go）完整覆盖了上述边界情况：

• 先 parseOptions(opts...) 解析选项，getPool() 判断是否并行；
• Eager 观察：立即 buildChannel() 并启动 runSequential/runParallel，返回基于新 channel 的 iterable；
• Lazy 观察 + 顺序执行：返回 newFactoryIterable 包装的延迟工厂，真正 Observe() 时才运行 runSequential；
• Lazy + 并行 + 序列化（WithSerialize）：先运行 runFirstItem 提取首个 item 的 ID 作为序列化锚点，再 runParallel 处理后续 item；
• Lazy + 并行（默认）：直接 runParallel。

forceSeq 与 bypassGather 参数进一步控制了是否强制顺序执行、是否绕过 gather 阶段。这些细节说明了为什么指南建议：优先复用这三个工具函数，不要自己手写管道逻辑。

不过，工具函数并非万能——当需要同时观察多个 Observable（如合并、窗口等场景）时，工具函数就不适用了。此时指南给出的建议是：参考现有复杂 Operator 的完整实现，最典型的就是 WindowWithTime（observable_operator.go）。该 Operator 展示了教科书级的边界处理：

• timespan == nil 时返回 Thrown(IllegalInputError{...})，拒绝非法输入（WindowWithTimeOrCount 还会校验 count < 0，见 observable_operator.go）；
• 通过 select 同时监听 ctx.Done()（上下文取消）、done（内部通知）、time.After(timespan.duration())（定时窗口切割）与上游 observe channel 四个事件源；
• 用 sync.Mutex 保护窗口 channel 的切换与关闭，避免数据竞争；
• 按 option.getErrorStrategy() 决定遇错是继续写入当前窗口还是关闭流水线。

从源码结构可以推断：WindowWithTime 采用的"select 多路复用 + 互斥锁保护窗口切换 + 错误策略分支"模式，是 RxGo 中处理多数据源/时间窗口类 Operator 的既定范式，值得作为你实现同类 Operator 时的参考模板。

单元测试：用 RxGo 断言 API 保证一致性

指南强调：必须包含单元测试，且测试风格保持一致——优先使用 RxGo 断言 API。断言 API 定义在 assert.go，核心入口为 Assert（assert.go）：

func Assert(ctx context.Context, t *testing.T, iterable Iterable, assertions ...RxAssert)

它从 iterable 中收集所有 item 与 error，再与给定的断言（如 HasItems、HasNoError、HasError 等，assert.go）逐一比对。用法示例：

rxgo.Assert(ctx, t, observable, rxgo.HasItems(1, 2, 3))

测试代码中常用的还有 testObservable 辅助函数（util_test.go），它把一组值/错误快速包装成 Observable，让测试用例简洁可读：

func testObservable(ctx context.Context, items ...interface{}) Observable {
	return FromChannel(channelValue(ctx, items...))
}

实际测试中 Assertion API 与 testObservable 的组合随处可见，例如 factory_test.go 中：

obs := Amb([]Observable{testObservable(ctx, 1, 2, 3), Empty()})
Assert(context.Background(), t, obs, HasItems(1, 2, 3))

仓库的 Makefile 还定义了完整的测试入口，提交前应运行：

test:
	go clean -testcache ./...
	go test -race -timeout 10s ./... --tags=all
	go test -timeout 10s -run TestLeak

注意其中启用了 -race 竞态检测（对应上文提到的并发边界情况）以及 TestLeak goroutine 泄漏测试（仓库测试大量使用 goleak.VerifyNone(t)，见 observable_operator_test.go 等测试文件）——并发安全与资源释放是 RxGo 测试的双重红线。

时间相关输入：rxgo.Duration 与确定性测试

这是贡献指南中技术含量最高的一节：如果 Operator 的输入包含 duration，必须使用 rxgo.Duration 类型。这样做的目的有两个：

1. 可 mock：Duration 是一个接口（duration.go）：

type Duration interface {
	duration() time.Duration
}

2. 确定性测试：配合 timeCausality() 实现不依赖真实时钟的测试。

如何构造 Duration 实例：duration.go 提供 WithDuration 包装任意 time.Duration：

func WithDuration(d time.Duration) Duration {
	return &duration{d: d}
}

另有特殊常量 Infinite = -1（duration.go）表示无限等待时间。

为什么不用原生 time.Duration：因为无法在测试中"假扮"时间流逝。而 timeCausality()（duration.go）解决了这个问题。它接收一个元素序列，其中 tick 表示"时间推进一个最小步长"，其余元素表示"在该时刻发射的值或错误"：

func timeCausality(elems ...interface{}) (context.Context, Observable, Duration)

其实现原理（从源码可确认）：

• 为每个元素生成一个 execution（要么发射 item，要么标记为 tick，duration.go）；
• 返回的 causalityDuration 每次被调用 duration() 时，依次消费一个 execution 并执行：tick 返回 time.Nanosecond（时间只前进 1 纳秒），普通元素返回 time.Minute（时间前进 1 分钟）——从源码结构看，这种"按 tick 快进、按值跳进"的设计，让 Operator 的时间窗口逻辑以确定性顺序被驱动，而无需真实等待。

典型用法见 observable_operator_test.go 的 Debounce 测试：

func Test_Observable_Debounce(t *testing.T) {
	defer goleak.VerifyNone(t)
	ctx, obs, d := timeCausality(1, tick, 2, tick, 3, 4, 5, tick, 6, tick)
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()
	Assert(ctx, t, obs.Debounce(d, WithBufferedChannel(10), WithContext(ctx)),
		HasItems(1, 2, 5, 6))
}

这里用 tick 构造了"两次发射之间时间窗口被重置"的时序，从而断言 Debounce 只在静默期结束后才放行元素（1、2 因紧随的 tick 而被截断，5、6 之间因无 tick 而被放行）。错误路径同样可测——errFoo 混入序列后断言 HasError(errFoo)（同文件 L393-L399）。这一套 timeCausality + Assert 的组合，是 RxGo 实现时间类 Operator 确定性测试的标准方法论，你在写 Timer、Interval、Sample、Window 等带时间参数 Operator 的测试时应直接采用。

写出地道代码：风格指南与项目惯用法

指南要求遵循 Go 官方 CodeReviewComments 风格（对应 CONTRIBUTING.md 中的链接），并同时参考项目自身风格指南；两者冲突时以 Go 官方为准。

从仓库实现可以提炼出若干项目级惯用法，供贡献者参考：

• 函数式选项：所有 Operator 都以 opts ...Option 收尾（如 Debounce(d, WithBufferedChannel(10), WithContext(ctx))），新 Operator 必须沿用这一签名约定；
• Option 组合：WithPool、WithCPUPool、WithBufferedChannel、WithObservationStrategy、WithBackPressureStrategy、WithPublishStrategy、WithContext、WithSerialize 等选项（详见 options.go 与 doc/options.md）；
• 非法输入返回 Thrown：如 WindowWithTime 对空 timespan 返回 Thrown(IllegalInputError{...})，而不是 panic 或静默忽略；
• 上下文贯穿：next/err/end/gatherNext 均接收 context.Context，取消传播是并发安全的基石；
• 错误包装：item 通过 Of(value) 或 Error(err) 构造（见 item.go），消费端用 item.Error() 判断类型。

提交前的代码格式化：gofumpt 与 goimports

创建 Pull Request 之前，必须格式化代码。指南指定了两个工具：

gofumpt（更严格的 gofmt 超集）：

• 安装：go get mvdan.cc/gofumpt
• 执行：gofumpt -s -w .（-s 启用简化代码的简化模式，-w 直接写回文件）

goimports（负责 import 排序与自动补全）：

• 安装：go get golang.org/x/tools/cmd/goimports（注意：CONTRIBUTING.md 原文中的安装命令 go get golang.o 疑似截断，完整包路径为 golang.org/x/tools/cmd/goimports）
• 执行：goimports -w .

两者都以当前仓库根目录为作用域递归处理整个包，确保提交 diff 干净整洁。运行 gofumpt -s -w . 时会对 observable_operator.go 这类大文件做一致性重排，务必在格式化后重新跑一遍测试（make test）确认无行为变化。

先开 Issue 再动手：设计先行

指南的最后一条流程规范：在实现新功能或修复 Bug 之前，先开一个 Issue。这样做的目的是：

• 鼓励公开讨论，在动手编码前达成健全的设计决策；
• 让维护者与社区对 API 形态（尤其涉及 Option、返回类型选择 Observable/Single/OptionalSingle 等）提前达成共识；
• 如果提议的是新功能，务必在 Issue 主题中明确标注，方便维护者分类与 help needed 标签的筛选（README 的 Contributing 一节也提示新手可从带 help needed 标签的进行中 Issue 入手，见 README.md）。

结语

RxGo 的贡献指南虽短，却精准概括了这个仓库十余年沉淀的工程纪律：一致性优先、复用工具函数、边界情况全覆盖、断言 API 测试、rxgo.Duration 确定性测试、规范格式化、设计先行。对于希望为 Reactive Extensions for Go 贡献代码的开发者，建议按此流程行动：先阅读 README.md 的 Operator 清单与 doc/ 对应文档，再研读 observable.go 的三个工具函数与 duration.go 的测试机制，最后参考 observable_operator_test.go 中 timeCausality + Assert 的范式编写测试，提交前执行 gofumpt、goimports 与 make test。遵循这套规范，你的 Operator 将无缝融入这个拥有 80+ 成员与 250+ 测试的大家庭。