diff --git a/core/http-auth-aws/pom.xml b/core/http-auth-aws/pom.xml index 0a4568858194..ffa6cd67523d 100644 --- a/core/http-auth-aws/pom.xml +++ b/core/http-auth-aws/pom.xml @@ -126,11 +126,6 @@ equalsverifier test - - org.reactivestreams - reactive-streams-tck - test - software.amazon.awssdk test-utils diff --git a/core/http-auth-aws/src/test/java/software/amazon/awssdk/http/auth/aws/internal/signer/chunkedencoding/ChunkedEncodedPublisherTckTest.java b/core/http-auth-aws/src/test/java/software/amazon/awssdk/http/auth/aws/internal/signer/chunkedencoding/ChunkedEncodedPublisherTckTest.java index e539e5fe8ad4..69601858fdb1 100644 --- a/core/http-auth-aws/src/test/java/software/amazon/awssdk/http/auth/aws/internal/signer/chunkedencoding/ChunkedEncodedPublisherTckTest.java +++ b/core/http-auth-aws/src/test/java/software/amazon/awssdk/http/auth/aws/internal/signer/chunkedencoding/ChunkedEncodedPublisherTckTest.java @@ -61,9 +61,14 @@ private Publisher createChunkedPublisher(long chunksToProduce) { Publisher inputPublisher = Flowable.fromIterable(elements); + // Required by the builder; must match the data the input publisher produces. Production sets this from the + // x-amz-decoded-content-length header (see AwsChunkedV4PayloadSigner). + long contentLength = (long) totalElements * INPUT_STREAM_ELEMENT_SIZE; + return ChunkedEncodedPublisher.builder() .chunkSize(CHUNK_SIZE) .publisher(inputPublisher) + .contentLength(contentLength) .addEmptyTrailingChunk(false) .build(); } diff --git a/core/sdk-core/pom.xml b/core/sdk-core/pom.xml index eca7d3dbbbc4..0ffc18f518c4 100644 --- a/core/sdk-core/pom.xml +++ b/core/sdk-core/pom.xml @@ -193,11 +193,6 @@ ${awsjavasdk.version} test - - org.reactivestreams - reactive-streams-tck - test - com.google.jimfs jimfs @@ -232,33 +227,6 @@ - - - org.apache.maven.plugins - maven-surefire-plugin - ${maven.surefire.version} - - - - junit - false - - - 1 - - - - org.apache.maven.surefire - surefire-junit-platform - ${maven.surefire.version} - - - org.apache.maven.surefire - surefire-testng - ${maven.surefire.version} - - - org.codehaus.mojo build-helper-maven-plugin diff --git a/http-client-spi/pom.xml b/http-client-spi/pom.xml index feaac2313dda..340130c43937 100644 --- a/http-client-spi/pom.xml +++ b/http-client-spi/pom.xml @@ -75,11 +75,6 @@ assertj-core test - - org.reactivestreams - reactive-streams-tck - test - nl.jqno.equalsverifier equalsverifier diff --git a/http-clients/aws-crt-client/pom.xml b/http-clients/aws-crt-client/pom.xml index 8701e9091481..2463a0dda1df 100644 --- a/http-clients/aws-crt-client/pom.xml +++ b/http-clients/aws-crt-client/pom.xml @@ -112,11 +112,6 @@ assertj-core test - - org.reactivestreams - reactive-streams-tck - test - org.apache.logging.log4j log4j-api diff --git a/http-clients/netty-nio-client/pom.xml b/http-clients/netty-nio-client/pom.xml index 8572f6944a2c..67aefc681241 100644 --- a/http-clients/netty-nio-client/pom.xml +++ b/http-clients/netty-nio-client/pom.xml @@ -147,11 +147,6 @@ assertj-core test - - org.reactivestreams - reactive-streams-tck - test - org.apache.logging.log4j log4j-api @@ -254,33 +249,6 @@ - - - org.apache.maven.plugins - maven-surefire-plugin - ${maven.surefire.version} - - - - junit - false - - - 1 - - - - org.apache.maven.surefire - surefire-junit-platform - ${maven.surefire.version} - - - org.apache.maven.surefire - surefire-testng - ${maven.surefire.version} - - - org.apache.maven.plugins maven-jar-plugin diff --git a/pom.xml b/pom.xml index 78b309e29d89..10416b32e25c 100644 --- a/pom.xml +++ b/pom.xml @@ -213,6 +213,15 @@ + + + + org.reactivestreams + reactive-streams-tck + ${reactive-streams.version} + test + @@ -302,6 +311,13 @@ **/*TestCase.java ${skip.unit.tests} + + + + junit + false + + @@ -310,6 +326,11 @@ surefire-junit-platform ${maven.surefire.version} + + org.apache.maven.surefire + surefire-testng + ${maven.surefire.version} + diff --git a/services-custom/s3-transfer-manager/pom.xml b/services-custom/s3-transfer-manager/pom.xml index b1eb8736fd54..71017bff8771 100644 --- a/services-custom/s3-transfer-manager/pom.xml +++ b/services-custom/s3-transfer-manager/pom.xml @@ -185,11 +185,6 @@ commons-lang3 test - - org.reactivestreams - reactive-streams-tck - test - com.google.jimfs jimfs diff --git a/services-custom/s3-transfer-manager/src/main/java/software/amazon/awssdk/transfer/s3/internal/AsyncBufferingSubscriber.java b/services-custom/s3-transfer-manager/src/main/java/software/amazon/awssdk/transfer/s3/internal/AsyncBufferingSubscriber.java index a992d308857e..4375ecd6b76e 100644 --- a/services-custom/s3-transfer-manager/src/main/java/software/amazon/awssdk/transfer/s3/internal/AsyncBufferingSubscriber.java +++ b/services-custom/s3-transfer-manager/src/main/java/software/amazon/awssdk/transfer/s3/internal/AsyncBufferingSubscriber.java @@ -40,6 +40,7 @@ public class AsyncBufferingSubscriber implements Subscriber { private final int maxConcurrentExecutions; private final AtomicInteger numRequestsInFlight; private volatile boolean upstreamDone; + private volatile boolean onErrorInvoked; private volatile Subscription subscription; private final Set> requestsInFlight; @@ -56,9 +57,14 @@ public AsyncBufferingSubscriber(Function> consumer, returnFuture.whenComplete((r, t) -> { if (t != null) { requestsInFlight.forEach(f -> f.cancel(true)); - synchronized (this) { - if (subscription != null) { - subscription.cancel(); + // Skip cancelling when the failure came from onError: upstream has already terminated, and cancelling here + // would call Subscription::cancel from within onError (Reactive Streams rule 2.3). Still cancel on an + // external abort. + if (!onErrorInvoked) { + synchronized (this) { + if (subscription != null) { + subscription.cancel(); + } } } } @@ -79,6 +85,8 @@ public void onSubscribe(Subscription subscription) { @Override public void onNext(T item) { + // Reactive Streams rule 2.13: onNext must throw NullPointerException on a null element. + Validate.paramNotNull(item, "item"); numRequestsInFlight.incrementAndGet(); CompletableFuture currentRequest; @@ -111,6 +119,9 @@ public void onNext(T item) { @Override public void onError(Throwable t) { + // Set before completing the future: completeExceptionally may run the whenComplete handler synchronously, and it + // must see this flag to avoid cancelling the subscription from within onError (see constructor). + onErrorInvoked = true; // Need to complete future exceptionally first to prevent // accidental successful completion by a concurrent checkForCompletion. returnFuture.completeExceptionally(t); diff --git a/services-custom/s3-transfer-manager/src/test/java/software/amazon/awssdk/transfer/s3/internal/AsyncBufferingSubscriberTest.java b/services-custom/s3-transfer-manager/src/test/java/software/amazon/awssdk/transfer/s3/internal/AsyncBufferingSubscriberTest.java index 9b570af8cd42..be82440f86ad 100644 --- a/services-custom/s3-transfer-manager/src/test/java/software/amazon/awssdk/transfer/s3/internal/AsyncBufferingSubscriberTest.java +++ b/services-custom/s3-transfer-manager/src/test/java/software/amazon/awssdk/transfer/s3/internal/AsyncBufferingSubscriberTest.java @@ -128,12 +128,9 @@ public void consumerFunctionThrows_shouldCancelSubscriptionAndCompleteFutureExce subscriber.onSubscribe(mockSubscription); subscriber.onNext("item"); - /* - subscription.cancel() now exists in two codepaths: - - in onNext() catch block. - - in future.whenComplete() - */ - verify(mockSubscription, times(2)).cancel(); + // Cancelled once, from the onNext() catch block. The whenComplete() handler does not cancel again, which would + // violate Reactive Streams rule 2.3 (cancel from within onError). + verify(mockSubscription, times(1)).cancel(); assertThatThrownBy(future::join).hasCause(exception); } diff --git a/services/s3/pom.xml b/services/s3/pom.xml index f00ca9067a1e..22a328722194 100644 --- a/services/s3/pom.xml +++ b/services/s3/pom.xml @@ -180,11 +180,6 @@ wiremock-jre8-standalone test - - org.reactivestreams - reactive-streams-tck - test - software.amazon.awssdk auth-crt diff --git a/utils/pom.xml b/utils/pom.xml index 0e69fd898729..08ef3ffea2f7 100644 --- a/utils/pom.xml +++ b/utils/pom.xml @@ -94,11 +94,6 @@ commons-io test - - org.reactivestreams - reactive-streams-tck - test - org.apache.logging.log4j log4j-api diff --git a/utils/src/main/java/software/amazon/awssdk/utils/async/SimplePublisher.java b/utils/src/main/java/software/amazon/awssdk/utils/async/SimplePublisher.java index e931af7b1c84..15526e19c55d 100644 --- a/utils/src/main/java/software/amazon/awssdk/utils/async/SimplePublisher.java +++ b/utils/src/main/java/software/amazon/awssdk/utils/async/SimplePublisher.java @@ -95,8 +95,23 @@ public final class SimplePublisher implements Publisher { */ private final FailureMessage failureMessage = new FailureMessage(); + /** + * Whether {@link #subscribe(Subscriber)} has already been called, used to reject a second subscription. Tracked + * separately from {@link #subscriber} because that field is cleared on termination and must not re-open subscription. + */ + private final AtomicBoolean subscribed = new AtomicBoolean(false); + + /** + * True while the subscriber's {@code onSubscribe} is executing. Used to prevent {@link #processEventQueue()} from + * delivering {@code onNext} signals before {@code onSubscribe} has returned, as required by Reactive Streams rule 1.03. + */ + private volatile boolean onSubscribeInProgress = false; + /** * The subscriber provided via {@link #subscribe(Subscriber)}. This publisher only supports a single subscriber. + * + *

Cleared when the stream terminates (complete, error, or cancel) so the publisher does not retain the subscriber + * reference, per Reactive Streams rule 3.13. */ private Subscriber subscriber; @@ -196,14 +211,19 @@ public CompletableFuture error(Throwable error) { */ @Override public void subscribe(Subscriber s) { - if (subscriber != null) { + if (!subscribed.compareAndSet(false, true)) { s.onSubscribe(new NoOpSubscription()); s.onError(new IllegalStateException("Only one subscription may be active at a time.")); return; } this.subscriber = s; - s.onSubscribe(new SubscriptionImpl()); + onSubscribeInProgress = true; + try { + s.onSubscribe(new SubscriptionImpl()); + } finally { + onSubscribeInProgress = false; + } processEventQueue(); } @@ -275,6 +295,7 @@ private void doProcessQueue() { log.trace(() -> "Calling onComplete()"); subscriber.onComplete(); + subscriber = null; break; case ON_ERROR: @@ -283,9 +304,11 @@ private void doProcessQueue() { onErrorEntry.failure)); log.trace(() -> "Calling onError() with " + onErrorEntry.failure, onErrorEntry.failure); subscriber.onError(onErrorEntry.failure); + subscriber = null; break; case CANCEL: failureMessage.trySet(() -> new CancellationException("subscription has been cancelled.")); + subscriber = null; break; default: // Should never happen. Famous last words? @@ -317,6 +340,11 @@ private boolean shouldProcessQueueEntry(QueueEntry entry) { return false; } + if (onSubscribeInProgress) { + // Do not deliver signals until onSubscribe has returned (Reactive Streams rule 1.03). + return false; + } + if (entry.type() != ON_NEXT) { // This event isn't an on-next event, so we don't need subscriber demand in order to process it. return true; @@ -338,7 +366,10 @@ private void panicAndDie(Throwable cause) { // Create exception here instead of in supplier to preserve a more-useful stack trace. RuntimeException failure = new IllegalStateException("Encountered fatal error in publisher", cause); failureMessage.trySet(() -> failure); - subscriber.onError(cause instanceof Error ? cause : failure); + if (subscriber != null) { + subscriber.onError(cause instanceof Error ? cause : failure); + subscriber = null; + } while (true) { QueueEntry entry = standardPriorityQueue.poll();