反应流 TCK(技术兼容性工具包)对简单应用程序停滞/速度缓慢的测试将不会运行

Question

问题：

Need help successfully running test from Reactive Streams TCK (Technology Compatibility Kit)?

简单发布者.java(示例)

package aaa.bbb.ccc.jar;
import java.util.Iterator;
import java.util.stream.IntStream;
import org.reactivestreams.FlowAdapters;
import java.util.concurrent.Flow;
public class SimplePublisher implements Flow.Publisher<Integer> {
private final Iterator<Integer> iterator;
SimplePublisher(int count) {
this.iterator = IntStream.rangeClosed(1, count).iterator();
}
@Override
public void subscribe(Flow.Subscriber<? super Integer> subscriber) {
iterator.forEachRemaining(subscriber::onNext);
subscriber.onComplete();
}
public static void main(String[] args) {
new SimplePublisher(10).subscribe(new Flow.Subscriber<>() {
@Override
public void onSubscribe(Flow.Subscription subscription) {
}
@Override
public void onNext(Integer item) {
System.out.println("item = [" + item + "]");
}
@Override
public void onError(Throwable throwable) {
}
@Override
public void onComplete() {
System.out.println("complete");
}
});
}    
}

。运行"main()"方法成功运行，生成预测的输出...

item = [1]
item = [2]
item = [3]
item = [4]
item = [5]
item = [6]
item = [7]
item = [8]
item = [9]
item = [10]
complete

SimplePublisherTest.java(test - 无法运行，即速度变慢/挂起)

package aaa.bbb.ccc.jar;
import java.util.concurrent.Flow;
import org.reactivestreams.tck.TestEnvironment;
import org.reactivestreams.tck.flow.FlowPublisherVerification;
//FlowPublisherVerification
public class SimplePublisherTest extends FlowPublisherVerification<Integer> {
public SimplePublisherTest() {
super(new TestEnvironment());
}
@Override
public Flow.Publisher<Integer> createFlowPublisher(long elements) {
return new SimplePublisher((int) elements);
}
@Override
public Flow.Publisher<Integer> createFailedFlowPublisher() {
return null;
}
}

运行测试的输出

--- maven-clean-plugin:2.5:clean (default-clean) @ sp ---
Deleting C:toolssptarget
--- maven-resources-plugin:2.6:resources (default-resources) @ sp ---
Using 'UTF-8' encoding to copy filtered resources.
skip non existing resourceDirectory C:toolsspsrcmainresources
--- maven-compiler-plugin:3.8.0:compile (default-compile) @ sp ---
Changes detected - recompiling the module!
Compiling 1 source file to C:toolssptargetclasses
--- maven-resources-plugin:2.6:testResources (default-testResources) @ sp ---
Using 'UTF-8' encoding to copy filtered resources.
skip non existing resourceDirectory C:toolsspsrctestresources
--- maven-compiler-plugin:3.8.0:testCompile (default-testCompile) @ sp ---
Changes detected - recompiling the module!
Compiling 1 source file to C:toolssptargettest-classes
--- maven-surefire-plugin:2.12.4:test (default-test) @ sp ---
Surefire report directory: C:toolssptargetsurefire-reports
-------------------------------------------------------
T E S T S
-------------------------------------------------------
Running aaa.bbb.ccc.jar.SimplePublisherTest
Configuring TestNG with: org.apache.maven.surefire.testng.conf.TestNGMapConfigurator@123a439b
(test hangs at this point!)

庞.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>aaa.bbb.ccc</groupId>
<artifactId>sp</artifactId>
<version>1.0</version>
<packaging>jar</packaging>
<name>sp</name>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<skip.unit.tests>false</skip.unit.tests>
<maven.compiler.source>11</maven.compiler.source>
<maven.compiler.target>11</maven.compiler.target>
</properties>
<dependencies>
<dependency>
<groupId>org.reactivestreams</groupId>
<artifactId>reactive-streams</artifactId>
<version>1.0.2</version>
</dependency> 
<dependency>
<groupId>org.reactivestreams</groupId>
<artifactId>reactive-streams-flow-adapters</artifactId>
<version>1.0.2</version>
</dependency>           
<dependency>
<groupId>org.reactivestreams</groupId>
<artifactId>reactive-streams-tck-flow</artifactId>
<version>1.0.2</version>
</dependency>
</dependencies>
<build>
<finalName>${project.artifactId}</finalName>        
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<version>3.8.0</version>
<configuration>
<release>11</release>
</configuration> 
</plugin>           
</plugins>
</build>            
</project>

示例取自...

https://blog.softwaremill.com/how-not-to-use-reactive-streams-in-java-9-7a39ea9c2cb3

环境

爪哇 11

反应流 (v1.0.2)

反应流流适配器 (v1.0.2)

反应式流 TCK 流 (v1.0.2)

Answer 1

似乎答案只是将 maxElementsFromPublisher() 方法添加到 SimplePublisherTest.java 类中......

例如，

@Override
public long maxElementsFromPublisher() {
return 1000;
}

---

在此 github 页面上找到了此信息...

https://github.com/reactive-streams/reactive-streams-jvm/tree/master/tck

从 readme.md...

为了通知 TCK 发布服务器最多只能发出 2 个元素的信号，请覆盖 maxElementsFromPublisher 方法，如下所示：

@Override public long maxElementsFromPublisher() {
return 2;
}

TCK 还支持无法发出完成信号的发布者。想象一下，一个发布服务器由一个计时器支持，这样的发布服务器没有一种自然的方式来在一定数量的滴答声后"完成"。可以实现一个处理器，它将"从TickPublisher获取n个元素，然后向下游发出完成信号"，但这在TCK和最初想要测试的发布器之间增加了一个间接层。建议以任何一种方式测试这些无信任的发布者 - 使用"TakeNElementsProcessor"或通过通知TCK发布者无法发出完成信号。然后，TCK 将跳过所有需要发射 oncomplete信号的测试。

为了通知 TCK 您的 Publiher 无法发出完成信号，请覆盖 maxElementsFromPublisher 方法，如下所示：

@Override public long maxElementsFromPublisher() {
return publisherUnableToSignalOnComplete(); // == Long.MAX_VALUE == unbounded
}