附录 E：RS TCK 测试方法参考

先把边界说清楚

Reactive Streams 1.0.4 有 43 条规则，TCK 并没有为每一条都写断言。它只提供四类基类，用例由 TestNG 驱动；方法名前缀直接表示覆盖强度：

TCK 1.0.4 的 POM 直接依赖 TestNG 7.3.0；依赖树里还会出现 JUnit 4.12，那是 TestNG 自己的 compile 依赖，不代表 TCK 用例是 JUnit 用例。运行入口是 org.testng.TestNG，不是 JUnit Runner。

四类基类的真实规模

数量按 1.0.4 源码逐个方法统计：

IdentityProcessorVerification 的「其他」是 mustImmediatelyPassOnOnErrorEventsReceivedFromItsUpstreamToItsDownstream，对应 §4.2 的「不恢复就必须立即向下游传播 onError」。

规则覆盖到什么程度

这张表回答的是「某条规则到底有没有自动化断言」，而不是「实现是否合规」。同一个规则可能同时有 required、optional 和 untested 三种形态。

结论很直接：§1.6、§1.8、§1.10、§2.4、§2.6、§2.7、§2.11、§2.12、§3.1、§3.4、§3.5、§3.10、§3.11、§3.14、§3.15、§3.16 在 1.0.4 中没有 required 断言。TCK 全绿只说明「被检查的部分」合规，这些条款需要自己补测试或做代码审查。

PublisherVerification

必须实现的抽象方法

public abstract Publisher<T> createPublisher(long elements);
public abstract Publisher<T> createFailedPublisher();

两个方法的语义：

• createPublisher(long elements) 返回的 Publisher 必须恰好发射 elements 个元素，并在之后终止。TCK 至少会用 1 和 3 调用它，所以「永远只发一个元素」的实现不能直接拿来跑。
• createFailedPublisher() 返回 null 表示跳过失败路径用例；返回实例时，该实例必须在 onSubscribe 之后发 onError，而不是直接抛异常。

可选覆写：

public long maxElementsFromPublisher() { return Long.MAX_VALUE - 1; }          // 默认值
public long boundedDepthOfOnNextAndRequestRecursion() { return 1; }            // 默认值
public boolean skipStochasticTests() { return false; }                         // 默认值

maxElementsFromPublisher() 的默认值是 Long.MAX_VALUE - 1，不是 Long.MAX_VALUE。返回 Long.MAX_VALUE 会让依赖于「有限流」的用例直接跳过。

22 个必测方法

配置与元素数：

§1 发布侧：

§3 订阅侧：

可选与未测

8 个 optional 方法：

7 个 untested 方法会直接跳过：spec106（终止后订阅视为取消）、spec107（错误分支终止后静默）、spec108（取消后不再收到终止信号）、spec109（subscribe 抛非致命异常）、spec110（同一 Subscriber 重复订阅）、spec304（request 不做重计算）、spec305（cancel 不做重计算）。

另有 stochastic_spec103_mustSignalOnMethodsSequentially，重复检查信号是否串行。

SubscriberWhiteboxVerification

必须实现的方法

public abstract Subscriber<T> createSubscriber(WhiteboxSubscriberProbe<T> probe);
public abstract T createElement(int element);          // 来自 WithHelperPublisher

publisherExecutorService() 在基类里已有默认实现：@BeforeClass 创建 4 线程池，@AfterClass 关闭。只有需要自定义线程模型时才覆写，且覆写后要自己负责关闭。

白盒没有 triggerRequest。测试通过 WhiteboxSubscriberProbe 注册的 SubscriberPuppet 控制请求与取消：

SubscriberPuppet 只有两个方法：triggerRequest(long n) 和 signalCancel()。只实现 onNext 而不调用 registerOnNext，TCK 会认为信号从未到达，最终以超时失败。

14 个必测方法

13 个 untested：spec202、spec204、spec206、spec207、spec211、spec212、spec213_failingOnSignalInvocation、spec301、spec310、spec311、spec314、spec315、spec316。

SubscriberBlackboxVerification

必须实现的方法

public abstract Subscriber<T> createSubscriber();
public abstract T createElement(int element);

publisherExecutorService() 同样有默认实现。黑盒额外提供：

public void triggerRequest(Subscriber<? super T> subscriber) {
    // 默认什么都不做
}

如果你的 Subscriber 不在 onSubscribe 里主动请求，而是等待外部触发，就必须覆写 triggerRequest；否则 TCK 发不出元素，用例会超时。

11 个必测方法

15 个 untested：spec202、spec204、spec206、spec207、spec208、spec211、spec212、spec213_failingOnSignalInvocation、spec301、spec308、spec310、spec311、spec314、spec315、spec316。

IdentityProcessorVerification

必须实现的四个方法

public abstract Processor<T, T> createIdentityProcessor(int bufferSize);
public abstract Publisher<T> createFailedPublisher();
public abstract T createElement(int element);              // 来自 WithHelperPublisher
public abstract ExecutorService publisherExecutorService(); // 来自 WithHelperPublisher

createPublisher(long) 和 createSubscriber(probe) 都有默认实现：前者用 createHelperPublisher(elements) 生成元素再喂给 Processor，后者把同一个 Processor 包装成 Subscriber。不要自己再写一遍。

68 个用例的构成

required_spec104_mustCallOnErrorOnAllItsSubscribers... 验证 Processor 遇到不可恢复错误时，所有下游都收到 onError；required_mustRequestFromUpstream... 验证「下游很早以前请求过的元素，Processor 必须向上游补请求」，这是 Processor 背压桥接的核心。

§4.1 没有单独的测试方法，它由 Publisher 与 Subscriber 两套委托用例共同覆盖。§4.2 的「不恢复就立即传播」由 mustImmediatelyPassOn... 专门验证；「恢复则必须把上游订阅视为已取消」在 1.0.4 中没有对应 required 用例。

Flow 版本

reactive-streams-tck-flow:1.0.4 提供 FlowPublisherVerification、FlowSubscriberWhiteboxVerification、FlowSubscriberBlackboxVerification 和 IdentityFlowProcessorVerification。抽象方法换成 Flow 类型：

Reactor 3.8.8 的 Flux 实现的是 org.reactivestreams.Publisher，不是 java.util.concurrent.Flow.Publisher。要跑 Flow TCK，用 org.reactivestreams.FlowAdapters.toFlowPublisher(...) 适配，不要手写一层转换。

最小依赖与运行

<dependency>
    <groupId>org.reactivestreams</groupId>
    <artifactId>reactive-streams-tck</artifactId>
    <version>1.0.4</version>
    <scope>test</scope>
</dependency>

<!-- 只有验证 java.util.concurrent.Flow 类型时才需要 -->
<dependency>
    <groupId>org.reactivestreams</groupId>
    <artifactId>reactive-streams-tck-flow</artifactId>
    <version>1.0.4</version>
    <scope>test</scope>
</dependency>

一个可以直接跑通的 Publisher 测试：

import org.reactivestreams.Publisher;
import org.reactivestreams.tck.PublisherVerification;
import org.reactivestreams.tck.TestEnvironment;
import reactor.core.publisher.Flux;

public class RangePublisherTckTest extends PublisherVerification<Integer> {

    public RangePublisherTckTest() {
        super(new TestEnvironment(5_000));
    }

    @Override
    public Publisher<Integer> createPublisher(long elements) {
        return Flux.range(0, (int) elements);
    }

    @Override
    public Publisher<Integer> createFailedPublisher() {
        return null;   // 跳过失败路径用例；需要覆盖 §1.4 时换成 Flux.error(...)
    }
}

用 TestNG 运行：

java -cp "testng-7.3.0.jar:jcommander-1.78.jar:\
reactive-streams-tck-1.0.4.jar:reactive-streams-1.0.4.jar:\
reactor-core-3.8.8.jar:out" \
  org.testng.TestNG -testclass RangePublisherTckTest

Flow 版本只需替换基类和创建方法：

import org.reactivestreams.FlowAdapters;
import org.reactivestreams.tck.flow.FlowPublisherVerification;

public class FlowRangeTckTest extends FlowPublisherVerification<Integer> {

    public FlowRangeTckTest() {
        super(new TestEnvironment(5_000));
    }

    @Override
    public java.util.concurrent.Flow.Publisher<Integer> createFlowPublisher(long elements) {
        return FlowAdapters.toFlowPublisher(Flux.range(0, (int) elements));
    }

    @Override
    public java.util.concurrent.Flow.Publisher<Integer> createFailedFlowPublisher() {
        return null;
    }
}

实测计数

以下数字来自本机 TestNG 7.3.0 运行；createFailedPublisher() 返回 null，Publisher 两类用 1 秒超时：

Publisher 的 9 个跳过 = 7 个 untested_* + 2 个因没有失败 Publisher（optional_spec104、required_spec109_mayRejectCalls...）。把 createFailedPublisher() 改成 Flux.error(new RuntimeException("boom")) 后，变为 31 通过、7 跳过。

IdentityProcessorVerification 用热的多播处理器 PublishProcessor 时出现 14 个失败，不是 RxJava 不合规，而是「identity processor」要求能缓冲并按下游需求转发；热处理器没有下游时直接丢元素，语义不匹配。Processor 的 24 个跳过主要是 20 个 untested_* 和 2 个失败 Publisher 用例，其余是可选多播用例的条件跳过。

TestEnvironment 的默认超时是 100ms，不是 5000ms。单参构造器把默认超时、无信号超时、轮询间隔都设成同一个值；慢速 CI 上建议从 new TestEnvironment(5_000) 起步。

失败时怎么读

先看方法名里的规范编号，再对照上文的规则表；不要第一反应调大超时。超时通常说明信号根本发错了顺序或根本没发。

TCK 之外还要测什么

TCK 的 untested_* 已经列出一批空白，此外还有它天然覆盖不到的部分：

一个补充 TCK 的 StepVerifier 例子，验证「逐个请求 + 中途取消」：

@Test
void requestOneByOneThenCancel() {
    StepVerifier.create(Flux.range(1, 5), 1)
        .expectNext(1)
        .thenRequest(1)
        .expectNext(2)
        .thenCancel()
        .verify();
}

再用一个手写 Subscriber 验证 SyncSinglePublisher 对 request(0) 的处理：

@Test
void nonPositiveRequestSignalsIllegalArgumentException() {
    AtomicReference<Throwable> error = new AtomicReference<>();

    new SyncSinglePublisher<>("x").subscribe(new Flow.Subscriber<>() {
        @Override public void onSubscribe(Flow.Subscription s) { s.request(0); }
        @Override public void onNext(String item) { fail("unexpected onNext"); }
        @Override public void onError(Throwable t) { error.set(t); }
        @Override public void onComplete() { fail("unexpected onComplete"); }
    });

    assertInstanceOf(IllegalArgumentException.class, error.get());
}

不要用 Runtime.totalMemory() - freeMemory() 判断泄漏：这个差值受分配速率和 GC 时机影响，既不是峰值也不是稳定占用。要证明「取消后 Subscriber 可回收」，用弱引用加 GC，或者直接用 JFR 看对象是否还被 Publisher 持有。

相关阅读

阅读导航

上一篇：附录 D 从 CompletableFuture 到响应式流迁移指南

下一篇：返回书稿书架 选择其他卷。