Addressed comments.

Author: millems

http-clients/aws-crt-client/src/main/java/software/amazon/awssdk/http/crt/internal/response/CrtResponseAdapter.java

@@ -80,7 +80,7 @@ public void onResponseHeadersDone(HttpStream stream, int headerType) {
 
     @Override
     public int onResponseBody(HttpStream stream, byte[] bodyBytesIn) {
-        CompletableFuture<Void> writeFuture = responsePublisher.write(ByteBuffer.wrap(bodyBytesIn));
+        CompletableFuture<Void> writeFuture = responsePublisher.send(ByteBuffer.wrap(bodyBytesIn));
 
         if (writeFuture.isDone() && !writeFuture.isCompletedExceptionally()) {
             // Optimization: If write succeeded immediately, return non-zero to avoid the extra call back into the CRT.
@@ -89,7 +89,7 @@ public int onResponseBody(HttpStream stream, byte[] bodyBytesIn) {
 
         writeFuture.whenComplete((result, failure) -> {
             if (failure != null) {
-                die(stream, failure);
+                failResponseHandlerAndFuture(stream, failure);
                 return;
             }
 
@@ -111,7 +111,7 @@ public void onResponseComplete(HttpStream stream, int errorCode) {
     private void onSuccessfulResponseComplete(HttpStream stream) {
         responsePublisher.complete().whenComplete((result, failure) -> {
             if (failure != null) {
-                die(stream, failure);
+                failResponseHandlerAndFuture(stream, failure);
                 return;
             }
 
@@ -128,10 +128,10 @@ private void onSuccessfulResponseComplete(HttpStream stream) {
     private void onFailedResponseComplete(HttpStream stream, HttpException error) {
         log.error(() -> "HTTP response encountered an error.", error);
         responsePublisher.error(error);
-        die(stream, error);
+        failResponseHandlerAndFuture(stream, error);
     }
 
-    private void die(HttpStream stream, Throwable error) {
+    private void failResponseHandlerAndFuture(HttpStream stream, Throwable error) {
         callResponseHandlerOnError(error);
         completionFuture.completeExceptionally(error);
         connection.shutdown();

utils/src/main/java/software/amazon/awssdk/utils/async/SimplePublisher.java

@@ -33,31 +33,31 @@
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
-import software.amazon.awssdk.annotations.SdkPublicApi;
+import software.amazon.awssdk.annotations.SdkProtectedApi;
 import software.amazon.awssdk.utils.Logger;
 import software.amazon.awssdk.utils.Validate;
 
 /**
- * A {@link Publisher} to which callers can {@link #write(Object)} messages, simplifying the process of implementing a publisher.
+ * A {@link Publisher} to which callers can {@link #send(Object)} messages, simplifying the process of implementing a publisher.
  *
  * <p><b>Operations</b>
  *
  * <p>The {@code SimplePublisher} supports three simplified operations:
  * <ol>
- *     <li>{@link #write(Object)} for sending messages</li>
+ *     <li>{@link #send(Object)} for sending messages</li>
  *     <li>{@link #complete()} for indicating the successful end of messages</li>
  *     <li>{@link #error(Throwable)} for indicating the unsuccessful end of messages</li>
  * </ol>
  *
  * Each of these operations returns a {@link CompletableFuture} for indicating when the message has been successfully sent.
  *
- * <p>Callers are expected to invoke a series of {@link #write(Object)}s followed by a single {@link #complete()} or
+ * <p>Callers are expected to invoke a series of {@link #send(Object)}s followed by a single {@link #complete()} or
  * {@link #error(Throwable)}. See the documentation on each operation for more details.
  *
  * <p>This publisher will store an unbounded number of messages. It is recommended that callers limit the number of in-flight
- * {@link #write(Object)} operations in order to bound the amount of memory used by this publisher.
+ * {@link #send(Object)} operations in order to bound the amount of memory used by this publisher.
  */
-@SdkPublicApi
+@SdkProtectedApi
 public final class SimplePublisher<T> implements Publisher<T> {
     private static final Logger log = Logger.loggerFor(SimplePublisher.class);
 
@@ -86,7 +86,7 @@ public final class SimplePublisher<T> implements Publisher<T> {
     private final AtomicBoolean processingQueue = new AtomicBoolean(false);
 
     /**
-     * An exception that should be raised to any failed {@link #write(Object)}, {@link #complete()} or {@link #error(Throwable)}
+     * An exception that should be raised to any failed {@link #send(Object)}, {@link #complete()} or {@link #error(Throwable)}
      * operations. This is used to stop accepting messages after the downstream subscription is cancelled or after the
      * caller sends a {@code complete()} or {@code #error()}.
      *
@@ -100,25 +100,25 @@ public final class SimplePublisher<T> implements Publisher<T> {
     private Subscriber<? super T> subscriber;
 
     /**
-     * Write a message to this publisher.
+     * Send a message using this publisher.
      *
-     * <p>Messages written to this publisher will eventually be sent to a downstream subscriber, in the order they were
+     * <p>Messages sent using this publisher will eventually be sent to a downstream subscriber, in the order they were
      * written. When the message is sent to the subscriber, the returned future will be completed successfully.
      *
      * <p>This method may be invoked concurrently when the order of messages is not important.
      *
      * <p>In the time between when this method is invoked and the returned future is not completed, this publisher stores the
-     * request message in memory. Callers are recommended to limit the number of writes in progress at a time to bound the
+     * request message in memory. Callers are recommended to limit the number of sends in progress at a time to bound the
      * amount of memory used by this publisher.
      *
      * <p>The returned future will be completed exceptionally if the downstream subscriber cancels the subscription, or
-     * if the write call was performed after a {@link #complete()} or {@link #error(Throwable)} call.
+     * if the {@code send} call was performed after a {@link #complete()} or {@link #error(Throwable)} call.
      *
      * @param value The message to send. Must not be null.
      * @return A future that is completed when the message is sent to the subscriber.
      */
-    public CompletableFuture<Void> write(T value) {
-        log.trace(() -> "Received write() with " + value);
+    public CompletableFuture<Void> send(T value) {
+        log.trace(() -> "Received send() with " + value);
 
         OnNextQueueEntry<T> entry = new OnNextQueueEntry<>(value);
         try {
@@ -133,13 +133,13 @@ public CompletableFuture<Void> write(T value) {
     }
 
     /**
-     * Indicate that no more {@link #write(Object)} calls will be made, and that stream of messages is completed successfully.
+     * Indicate that no more {@link #send(Object)} calls will be made, and that stream of messages is completed successfully.
      *
-     * <p>This can be called before any in-flight {@code write} calls are complete. Such messages will be processed before the
+     * <p>This can be called before any in-flight {@code send} calls are complete. Such messages will be processed before the
      * stream is treated as complete. The returned future will be completed successfully when the {@code complete} is sent to
      * the downstream subscriber.
      *
-     * <p>After this method is invoked, any future {@link #write(Object)}, {@code complete()} or {@link #error(Throwable)}
+     * <p>After this method is invoked, any future {@link #send(Object)}, {@code complete()} or {@link #error(Throwable)}
      * calls will be completed exceptionally and not be processed.
      *
      * <p>The returned future will be completed exceptionally if the downstream subscriber cancels the subscription, or
@@ -164,13 +164,13 @@ public CompletableFuture<Void> complete() {
     }
 
     /**
-     * Indicate that no more {@link #write(Object)} calls will be made, and that streaming of messages has failed.
+     * Indicate that no more {@link #send(Object)} calls will be made, and that streaming of messages has failed.
      *
-     * <p>This can be called before any in-flight {@code write} calls are complete. Such messages will be processed before the
+     * <p>This can be called before any in-flight {@code send} calls are complete. Such messages will be processed before the
      * stream is treated as being in-error. The returned future will be completed successfully when the {@code error} is
      * sent to the downstream subscriber.
      *
-     * <p>After this method is invoked, any future {@link #write(Object)}, {@link #complete()} or {@code #error(Throwable)}
+     * <p>After this method is invoked, any future {@link #send(Object)}, {@link #complete()} or {@code #error(Throwable)}
      * calls will be completed exceptionally and not be processed.
      *
      * <p>The returned future will be completed exceptionally if the downstream subscriber cancels the subscription, or
@@ -270,17 +270,13 @@ private void doProcessQueue() {
                         entryTypesToFail.addAll(asList(ON_NEXT, ON_COMPLETE, ON_ERROR));
                         log.trace(() -> "Calling onComplete()");
                         subscriber.onComplete();
-                        outstandingDemand.set(0);
-                        log.trace(() -> "Set demand to 0");
                         break;
                     case ON_ERROR:
                         OnErrorQueueEntry<T> onErrorEntry = (OnErrorQueueEntry<T>) entry;
 
                         entryTypesToFail.addAll(asList(ON_NEXT, ON_COMPLETE, ON_ERROR));
                         log.trace(() -> "Calling onError() with " + onErrorEntry.failure, onErrorEntry.failure);
                         subscriber.onError(onErrorEntry.failure);
-                        outstandingDemand.set(0);
-                        log.trace(() -> "Set demand to 0");
                         break;
                     case CANCEL:
                         subscriber = null; // Allow subscriber to be garbage collected after cancellation.
@@ -385,10 +381,16 @@ public void request(long n) {
                                                                                 + "amount of data: " + n);
                 rejectException.compareAndSet(null, () -> failure);
                 eventQueue.add(new OnErrorQueueEntry<>(failure));
-                entryTypesToFail.add(ON_NEXT);
+                entryTypesToFail.addAll(asList(ON_NEXT, ON_COMPLETE));
                 processEventQueue();
             } else {
-                long newDemand = outstandingDemand.addAndGet(n);
+                long newDemand = outstandingDemand.updateAndGet(current -> {
+                    if (Long.MAX_VALUE - current < n) {
+                        return Long.MAX_VALUE;
+                    }
+
+                    return current + n;
+                });
                 log.trace(() -> "Increased demand to " + newDemand);
                 processEventQueue();
             }
@@ -412,7 +414,7 @@ public void cancel() {
      */
     abstract static class QueueEntry<T> {
         /**
-         * The future that was returned to a {@link #write(Object)}, {@link #complete()} or {@link #error(Throwable)} message.
+         * The future that was returned to a {@link #send(Object)}, {@link #complete()} or {@link #error(Throwable)} message.
          */
         protected final CompletableFuture<Void> resultFuture = new CompletableFuture<>();
 
@@ -430,7 +432,7 @@ protected enum Type {
     }
 
     /**
-     * An entry added when we get a {@link #write(Object)} call.
+     * An entry added when we get a {@link #send(Object)} call.
      */
     private static final class OnNextQueueEntry<T> extends QueueEntry<T> {
         private final T value;

utils/src/test/java/software/amazon/awssdk/utils/async/SimplePublisherTckTest.java

@@ -16,10 +16,7 @@
 package software.amazon.awssdk.utils.async;
 
 import org.reactivestreams.Publisher;
-import org.reactivestreams.Subscriber;
-import org.reactivestreams.Subscription;
 import org.reactivestreams.tck.PublisherVerification;
-import org.reactivestreams.tck.SubscriberWhiteboxVerification;
 import org.reactivestreams.tck.TestEnvironment;
 
 public class SimplePublisherTckTest extends PublisherVerification<Integer> {
@@ -31,7 +28,7 @@ public SimplePublisherTckTest() {
     public Publisher<Integer> createPublisher(long elements) {
         SimplePublisher<Integer> publisher = new SimplePublisher<>();
         for (int i = 0; i < elements; i++) {
-            publisher.write(i);
+            publisher.send(i);
         }
         publisher.complete();
         return publisher;

utils/src/test/java/software/amazon/awssdk/utils/async/SimplePublisherTest.java

@@ -81,14 +81,14 @@ public void immediateFailureWorks() {
     public void writeAfterCompleteFails() {
         SimplePublisher<Integer> publisher = new SimplePublisher<>();
         publisher.complete();
-        assertThat(publisher.write(5)).isCompletedExceptionally();
+        assertThat(publisher.send(5)).isCompletedExceptionally();
     }
 
     @Test
     public void writeAfterErrorFails() {
         SimplePublisher<Integer> publisher = new SimplePublisher<>();
         publisher.error(new Throwable());
-        assertThat(publisher.write(5)).isCompletedExceptionally();
+        assertThat(publisher.send(5)).isCompletedExceptionally();
     }
 
     @Test
@@ -125,8 +125,8 @@ public void oneDemandWorks() {
         StoringSubscriber<Integer> subscriber = new StoringSubscriber<>(1);
         publisher.subscribe(subscriber);
 
-        publisher.write(1);
-        publisher.write(2);
+        publisher.send(1);
+        publisher.send(2);
         publisher.complete();
 
         assertThat(subscriber.peek().get().type()).isEqualTo(EventType.ON_NEXT);
@@ -146,28 +146,32 @@ public void oneDemandWorks() {
     @Test
     public void highDemandWorks() {
         SimplePublisher<Integer> publisher = new SimplePublisher<>();
-        StoringSubscriber<Integer> subscriber = new StoringSubscriber<>(Integer.MAX_VALUE);
+        ControllableSubscriber<Integer> subscriber = new ControllableSubscriber<>();
         publisher.subscribe(subscriber);
+        subscriber.subscription.request(Long.MAX_VALUE);
 
-        publisher.write(1);
-        publisher.write(2);
+        publisher.send(1);
+        subscriber.subscription.request(Long.MAX_VALUE);
+        publisher.send(2);
+        subscriber.subscription.request(Long.MAX_VALUE);
         publisher.complete();
+        subscriber.subscription.request(Long.MAX_VALUE);
 
-        assertThat(subscriber.peek().get().type()).isEqualTo(EventType.ON_NEXT);
-        assertThat(subscriber.peek().get().value()).isEqualTo(1);
+        assertThat(subscriber.eventQueue.peek().get().type()).isEqualTo(EventType.ON_NEXT);
+        assertThat(subscriber.eventQueue.peek().get().value()).isEqualTo(1);
 
-        subscriber.poll();
+        subscriber.eventQueue.poll();
 
-        assertThat(subscriber.peek().get().type()).isEqualTo(EventType.ON_NEXT);
-        assertThat(subscriber.peek().get().value()).isEqualTo(2);
+        assertThat(subscriber.eventQueue.peek().get().type()).isEqualTo(EventType.ON_NEXT);
+        assertThat(subscriber.eventQueue.peek().get().value()).isEqualTo(2);
 
-        subscriber.poll();
+        subscriber.eventQueue.poll();
 
-        assertThat(subscriber.peek().get().type()).isEqualTo(EventType.ON_COMPLETE);
+        assertThat(subscriber.eventQueue.peek().get().type()).isEqualTo(EventType.ON_COMPLETE);
 
-        subscriber.poll();
+        subscriber.eventQueue.poll();
 
-        assertThat(subscriber.poll()).isNotPresent();
+        assertThat(subscriber.eventQueue.poll()).isNotPresent();
     }
 
     @Test
@@ -176,7 +180,7 @@ public void writeFuturesDoNotCompleteUntilAfterOnNext() {
         ControllableSubscriber<Integer> subscriber = new ControllableSubscriber<>();
         publisher.subscribe(subscriber);
 
-        CompletableFuture<Void> writeFuture = publisher.write(5);
+        CompletableFuture<Void> writeFuture = publisher.send(5);
 
         assertThat(subscriber.eventQueue.peek()).isNotPresent();
         assertThat(writeFuture).isNotCompleted();
@@ -194,7 +198,7 @@ public void completeFuturesDoNotCompleteUntilAfterOnComplete() {
         ControllableSubscriber<Integer> subscriber = new ControllableSubscriber<>();
 
         publisher.subscribe(subscriber);
-        publisher.write(5);
+        publisher.send(5);
         CompletableFuture<Void> completeFuture = publisher.complete();
 
         assertThat(subscriber.eventQueue.peek()).isNotPresent();
@@ -215,7 +219,7 @@ public void errorFuturesDoNotCompleteUntilAfterOnError() {
         ControllableSubscriber<Integer> subscriber = new ControllableSubscriber<>();
 
         publisher.subscribe(subscriber);
-        publisher.write(5);
+        publisher.send(5);
         CompletableFuture<Void> errorFuture = publisher.error(error);
 
         assertThat(subscriber.eventQueue.peek()).isNotPresent();
@@ -256,7 +260,7 @@ public void writeBeforeSubscribeIsDeliveredOnSubscribe() {
         SimplePublisher<Integer> publisher = new SimplePublisher<>();
         StoringSubscriber<Integer> subscriber = new StoringSubscriber<>(Integer.MAX_VALUE);
 
-        publisher.write(5);
+        publisher.send(5);
         publisher.subscribe(subscriber);
         assertThat(subscriber.peek().get().type()).isEqualTo(EventType.ON_NEXT);
         assertThat(subscriber.peek().get().value()).isEqualTo(5);
@@ -268,7 +272,7 @@ public void cancelFailsAnyInFlightFutures() {
         ControllableSubscriber<Integer> subscriber = new ControllableSubscriber<>();
 
         publisher.subscribe(subscriber);
-        CompletableFuture<Void> writeFuture = publisher.write(5);
+        CompletableFuture<Void> writeFuture = publisher.send(5);
         CompletableFuture<Void> completeFuture = publisher.complete();
 
         subscriber.subscription.cancel();
@@ -285,18 +289,33 @@ public void newCallsAfterCancelFail() {
         publisher.subscribe(subscriber);
         subscriber.subscription.cancel();
 
-        assertThat(publisher.write(5)).isCompletedExceptionally();
+        assertThat(publisher.send(5)).isCompletedExceptionally();
         assertThat(publisher.complete()).isCompletedExceptionally();
         assertThat(publisher.error(new Throwable())).isCompletedExceptionally();
     }
 
+    @Test
+    public void negativeDemandSkipsOutstandingMessages() {
+        SimplePublisher<Integer> publisher = new SimplePublisher<>();
+        ControllableSubscriber<Integer> subscriber = new ControllableSubscriber<>();
+
+        publisher.subscribe(subscriber);
+        CompletableFuture<Void> sendFuture = publisher.send(0);
+        CompletableFuture<Void> completeFuture = publisher.complete();
+        subscriber.subscription.request(-1);
+
+        assertThat(sendFuture).isCompletedExceptionally();
+        assertThat(completeFuture).isCompletedExceptionally();
+        assertThat(subscriber.eventQueue.poll().get().type()).isEqualTo(EventType.ON_ERROR);
+    }
+
     @Test
     public void evilDownstreamPublisherThrowingInOnNextStillCancelsInFlightFutures() {
         SimplePublisher<Integer> publisher = new SimplePublisher<>();
         ControllableSubscriber<Integer> subscriber = new ControllableSubscriber<>();
         subscriber.failureInOnNext = new RuntimeException();
 
-        CompletableFuture<Void> writeFuture = publisher.write(5);
+        CompletableFuture<Void> writeFuture = publisher.send(5);
         CompletableFuture<Void> completeFuture = publisher.complete();
 
         publisher.subscribe(subscriber);
@@ -343,7 +362,7 @@ public void stochastic_completeAndError_seemThreadSafe() throws Exception {
 
                 Future<?> writeCall = executor.submit(() -> {
                     waitForStart.run();
-                    publisher.write(0).join();
+                    publisher.send(0).join();
                 });
 
                 Future<?> completeCall = executor.submit(() -> {
@@ -404,7 +423,7 @@ private void seemsThreadSafeWithProducerCount(int producerCount) throws Interrup
                 producers.add(executor.submit(() -> {
                     while (runProducers.get()) {
                         productionLimiter.acquire();
-                        publisher.write(messageCount.getAndIncrement());
+                        publisher.send(messageCount.getAndIncrement());
                     }
                     publisher.complete(); // All but one producer sending this will fail.
                     return null;