Skip to content

Commit 141c229

Browse files
davidh44davidh44
authored
Compliance with Reactive Streams specs (#7191)
* Compliance with Reactive Streams specs * Move dependency to parent pom * Compliance with Reactive Streams specs
1 parent 03bb4e7 commit 141c229

13 files changed

Lines changed: 77 additions & 106 deletions

File tree

‎core/http-auth-aws/pom.xml‎

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -126,11 +126,6 @@
126126
<artifactId>equalsverifier</artifactId>
127127
<scope>test</scope>
128128
</dependency>
129-
<dependency>
130-
<groupId>org.reactivestreams</groupId>
131-
<artifactId>reactive-streams-tck</artifactId>
132-
<scope>test</scope>
133-
</dependency>
134129
<dependency>
135130
<groupId>software.amazon.awssdk</groupId>
136131
<artifactId>test-utils</artifactId>

‎core/http-auth-aws/src/test/java/software/amazon/awssdk/http/auth/aws/internal/signer/chunkedencoding/ChunkedEncodedPublisherTckTest.java‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -61,9 +61,14 @@ private Publisher<ByteBuffer> createChunkedPublisher(long chunksToProduce) {
6161

6262
Publisher<ByteBuffer> inputPublisher = Flowable.fromIterable(elements);
6363

64+
// Required by the builder; must match the data the input publisher produces. Production sets this from the
65+
// x-amz-decoded-content-length header (see AwsChunkedV4PayloadSigner).
66+
long contentLength = (long) totalElements * INPUT_STREAM_ELEMENT_SIZE;
67+
6468
return ChunkedEncodedPublisher.builder()
6569
.chunkSize(CHUNK_SIZE)
6670
.publisher(inputPublisher)
71+
.contentLength(contentLength)
6772
.addEmptyTrailingChunk(false)
6873
.build();
6974
}

‎core/sdk-core/pom.xml‎

Lines changed: 0 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -193,11 +193,6 @@
193193
<version>${awsjavasdk.version}</version>
194194
<scope>test</scope>
195195
</dependency>
196-
<dependency>
197-
<groupId>org.reactivestreams</groupId>
198-
<artifactId>reactive-streams-tck</artifactId>
199-
<scope>test</scope>
200-
</dependency>
201196
<dependency>
202197
<groupId>com.google.jimfs</groupId>
203198
<artifactId>jimfs</artifactId>
@@ -232,33 +227,6 @@
232227
</dependencies>
233228
<build>
234229
<plugins>
235-
<!-- The Reactive Streams TCK tests are based on TestNG. See http://maven.apache.org/surefire/maven-surefire-plugin/examples/testng.html#Running_TestNG_and_JUnit_Tests -->
236-
<plugin>
237-
<groupId>org.apache.maven.plugins</groupId>
238-
<artifactId>maven-surefire-plugin</artifactId>
239-
<version>${maven.surefire.version}</version>
240-
<configuration>
241-
<properties>
242-
<property>
243-
<name>junit</name>
244-
<value>false</value>
245-
</property>
246-
</properties>
247-
<threadCount>1</threadCount>
248-
</configuration>
249-
<dependencies>
250-
<dependency>
251-
<groupId>org.apache.maven.surefire</groupId>
252-
<artifactId>surefire-junit-platform</artifactId>
253-
<version>${maven.surefire.version}</version>
254-
</dependency>
255-
<dependency>
256-
<groupId>org.apache.maven.surefire</groupId>
257-
<artifactId>surefire-testng</artifactId>
258-
<version>${maven.surefire.version}</version>
259-
</dependency>
260-
</dependencies>
261-
</plugin>
262230
<plugin>
263231
<groupId>org.codehaus.mojo</groupId>
264232
<artifactId>build-helper-maven-plugin</artifactId>

‎http-client-spi/pom.xml‎

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -75,11 +75,6 @@
7575
<artifactId>assertj-core</artifactId>
7676
<scope>test</scope>
7777
</dependency>
78-
<dependency>
79-
<groupId>org.reactivestreams</groupId>
80-
<artifactId>reactive-streams-tck</artifactId>
81-
<scope>test</scope>
82-
</dependency>
8378
<dependency>
8479
<groupId>nl.jqno.equalsverifier</groupId>
8580
<artifactId>equalsverifier</artifactId>

‎http-clients/aws-crt-client/pom.xml‎

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -112,11 +112,6 @@
112112
<artifactId>assertj-core</artifactId>
113113
<scope>test</scope>
114114
</dependency>
115-
<dependency>
116-
<groupId>org.reactivestreams</groupId>
117-
<artifactId>reactive-streams-tck</artifactId>
118-
<scope>test</scope>
119-
</dependency>
120115
<dependency>
121116
<groupId>org.apache.logging.log4j</groupId>
122117
<artifactId>log4j-api</artifactId>

‎http-clients/netty-nio-client/pom.xml‎

Lines changed: 0 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -147,11 +147,6 @@
147147
<artifactId>assertj-core</artifactId>
148148
<scope>test</scope>
149149
</dependency>
150-
<dependency>
151-
<groupId>org.reactivestreams</groupId>
152-
<artifactId>reactive-streams-tck</artifactId>
153-
<scope>test</scope>
154-
</dependency>
155150
<dependency>
156151
<groupId>org.apache.logging.log4j</groupId>
157152
<artifactId>log4j-api</artifactId>
@@ -254,33 +249,6 @@
254249

255250
<build>
256251
<plugins>
257-
<!-- The Reactive Streams TCK tests are based on TestNG. See http://maven.apache.org/surefire/maven-surefire-plugin/examples/testng.html#Running_TestNG_and_JUnit_Tests -->
258-
<plugin>
259-
<groupId>org.apache.maven.plugins</groupId>
260-
<artifactId>maven-surefire-plugin</artifactId>
261-
<version>${maven.surefire.version}</version>
262-
<configuration>
263-
<properties>
264-
<property>
265-
<name>junit</name>
266-
<value>false</value>
267-
</property>
268-
</properties>
269-
<threadCount>1</threadCount>
270-
</configuration>
271-
<dependencies>
272-
<dependency>
273-
<groupId>org.apache.maven.surefire</groupId>
274-
<artifactId>surefire-junit-platform</artifactId>
275-
<version>${maven.surefire.version}</version>
276-
</dependency>
277-
<dependency>
278-
<groupId>org.apache.maven.surefire</groupId>
279-
<artifactId>surefire-testng</artifactId>
280-
<version>${maven.surefire.version}</version>
281-
</dependency>
282-
</dependencies>
283-
</plugin>
284252
<plugin>
285253
<groupId>org.apache.maven.plugins</groupId>
286254
<artifactId>maven-jar-plugin</artifactId>

‎pom.xml‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -213,6 +213,15 @@
213213

214214
<dependencies>
215215
<!-- Internal dependencies are managed in the bom-internal module. -->
216+
217+
<!-- reactive-streams-tck brings TestNG, which is needed globally so the surefire-testng provider
218+
can discover TCK tests in any module without per-module surefire configuration. -->
219+
<dependency>
220+
<groupId>org.reactivestreams</groupId>
221+
<artifactId>reactive-streams-tck</artifactId>
222+
<version>${reactive-streams.version}</version>
223+
<scope>test</scope>
224+
</dependency>
216225
</dependencies>
217226

218227
<build>
@@ -302,6 +311,13 @@
302311
<include>**/*TestCase.java</include>
303312
</includes>
304313
<skipTests>${skip.unit.tests}</skipTests>
314+
<!-- Prevent TestNG from re-running JUnit tests -->
315+
<properties>
316+
<property>
317+
<name>junit</name>
318+
<value>false</value>
319+
</property>
320+
</properties>
305321
</configuration>
306322
<!-- Have to explicitly set surefire provider because reactivestreamsTCK is using TestNG-->
307323
<dependencies>
@@ -310,6 +326,11 @@
310326
<artifactId>surefire-junit-platform</artifactId>
311327
<version>${maven.surefire.version}</version>
312328
</dependency>
329+
<dependency>
330+
<groupId>org.apache.maven.surefire</groupId>
331+
<artifactId>surefire-testng</artifactId>
332+
<version>${maven.surefire.version}</version>
333+
</dependency>
313334
</dependencies>
314335
</plugin>
315336
<plugin>

‎services-custom/s3-transfer-manager/pom.xml‎

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -185,11 +185,6 @@
185185
<artifactId>commons-lang3</artifactId>
186186
<scope>test</scope>
187187
</dependency>
188-
<dependency>
189-
<groupId>org.reactivestreams</groupId>
190-
<artifactId>reactive-streams-tck</artifactId>
191-
<scope>test</scope>
192-
</dependency>
193188
<dependency>
194189
<groupId>com.google.jimfs</groupId>
195190
<artifactId>jimfs</artifactId>

‎services-custom/s3-transfer-manager/src/main/java/software/amazon/awssdk/transfer/s3/internal/AsyncBufferingSubscriber.java‎

Lines changed: 14 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,7 @@ public class AsyncBufferingSubscriber<T> implements Subscriber<T> {
4040
private final int maxConcurrentExecutions;
4141
private final AtomicInteger numRequestsInFlight;
4242
private volatile boolean upstreamDone;
43+
private volatile boolean onErrorInvoked;
4344
private volatile Subscription subscription;
4445

4546
private final Set<CompletableFuture<?>> requestsInFlight;
@@ -56,9 +57,14 @@ public AsyncBufferingSubscriber(Function<T, CompletableFuture<?>> consumer,
5657
returnFuture.whenComplete((r, t) -> {
5758
if (t != null) {
5859
requestsInFlight.forEach(f -> f.cancel(true));
59-
synchronized (this) {
60-
if (subscription != null) {
61-
subscription.cancel();
60+
// Skip cancelling when the failure came from onError: upstream has already terminated, and cancelling here
61+
// would call Subscription::cancel from within onError (Reactive Streams rule 2.3). Still cancel on an
62+
// external abort.
63+
if (!onErrorInvoked) {
64+
synchronized (this) {
65+
if (subscription != null) {
66+
subscription.cancel();
67+
}
6268
}
6369
}
6470
}
@@ -79,6 +85,8 @@ public void onSubscribe(Subscription subscription) {
7985

8086
@Override
8187
public void onNext(T item) {
88+
// Reactive Streams rule 2.13: onNext must throw NullPointerException on a null element.
89+
Validate.paramNotNull(item, "item");
8290
numRequestsInFlight.incrementAndGet();
8391
CompletableFuture<?> currentRequest;
8492

@@ -111,6 +119,9 @@ public void onNext(T item) {
111119

112120
@Override
113121
public void onError(Throwable t) {
122+
// Set before completing the future: completeExceptionally may run the whenComplete handler synchronously, and it
123+
// must see this flag to avoid cancelling the subscription from within onError (see constructor).
124+
onErrorInvoked = true;
114125
// Need to complete future exceptionally first to prevent
115126
// accidental successful completion by a concurrent checkForCompletion.
116127
returnFuture.completeExceptionally(t);

‎services-custom/s3-transfer-manager/src/test/java/software/amazon/awssdk/transfer/s3/internal/AsyncBufferingSubscriberTest.java‎

Lines changed: 3 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -128,12 +128,9 @@ public void consumerFunctionThrows_shouldCancelSubscriptionAndCompleteFutureExce
128128
subscriber.onSubscribe(mockSubscription);
129129
subscriber.onNext("item");
130130

131-
/*
132-
subscription.cancel() now exists in two codepaths:
133-
- in onNext() catch block.
134-
- in future.whenComplete()
135-
*/
136-
verify(mockSubscription, times(2)).cancel();
131+
// Cancelled once, from the onNext() catch block. The whenComplete() handler does not cancel again, which would
132+
// violate Reactive Streams rule 2.3 (cancel from within onError).
133+
verify(mockSubscription, times(1)).cancel();
137134
assertThatThrownBy(future::join).hasCause(exception);
138135
}
139136

0 commit comments

Comments
 (0)