Add TCK tests for reactive streams impls

Fixes #405
Fixes #406

Author: dagnir

.changes/next-release/feature-AWSSDKforJavav2-61af032.json

@@ -0,0 +1,5 @@
+{
+    "category": "AWS SDK for Java v2", 
+    "type": "feature", 
+    "description": "Incorporate the [Reactive Streams Technology Compatibility Kit](https://github.com/reactive-streams/reactive-streams-jvm/tree/master/tck) and ensure current implementations are compliant. [#519](https://github.com/aws/aws-sdk-java-v2/issues/519)"
+}

CONTRIBUTING.md

@@ -128,6 +128,18 @@ checklist below:
   Commit the new file created by the script in `.changes/next-release` with
   your changes.
 
+__Note__: Some changes have additional requirements. Refer to the section below
+to see if your change will require additional work to be accepted.
+
+#### Additional Pull Request Requirements
+##### Reactive Streams
+If the change includes implementations of the [Reactive Streams
+interfaces](https://github.com/reactive-streams/reactive-streams-jvm), the
+change must also contain verification tests using the [Reactive Streams
+Technology Compatibility
+Kit](https://github.com/reactive-streams/reactive-streams-jvm/tree/master/tck)
+to ensure specificiation compliance.
+
 ### Getting Your Pull Request Merged
 All Pull Requests must be approved by at least one member of the SDK team
 before it can be merged in. The members only have limited bandwitdth to review

core/pom.xml

@@ -118,7 +118,6 @@
         <dependency>
             <groupId>org.reactivestreams</groupId>
             <artifactId>reactive-streams</artifactId>
-            <version>${reactive-streams.version}</version>
         </dependency>
 
         <dependency>
@@ -135,13 +134,11 @@
         <dependency>
             <groupId>org.reactivestreams</groupId>
             <artifactId>reactive-streams-tck</artifactId>
-            <version>${reactive-streams.version}</version>
             <scope>test</scope>
         </dependency>
         <dependency>
             <groupId>com.google.jimfs</groupId>
             <artifactId>jimfs</artifactId>
-            <version>1.1</version>
             <scope>test</scope>
         </dependency>
     </dependencies>

core/src/main/java/software/amazon/awssdk/core/async/AsyncResponseTransformer.java

@@ -21,6 +21,8 @@
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscription;
 import software.amazon.awssdk.core.ResponseBytes;
+import software.amazon.awssdk.core.internal.async.ByteArrayAsyncResponseTransformer;
+import software.amazon.awssdk.core.internal.async.FileAsyncResponseTransformer;
 
 /**
  * Callback interface to handle a streaming asynchronous response.

core/src/main/java/software/amazon/awssdk/core/async/ByteArrayAsyncRequestBody.java

@@ -19,6 +19,7 @@
 
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
+
 import software.amazon.awssdk.annotations.SdkInternalApi;
 
 /**
@@ -32,51 +33,54 @@
 @SdkInternalApi
 final class ByteArrayAsyncRequestBody implements AsyncRequestBody {
 
-  private final byte[] bytes;
+    private final byte[] bytes;
 
     ByteArrayAsyncRequestBody(byte[] bytes) {
         this.bytes = bytes.clone();
     }
 
-  @Override
-  public long contentLength() {
-    return bytes.length;
-  }
+    @Override
+    public long contentLength() {
+        return bytes.length;
+    }
+
+    @Override
+    public void subscribe(Subscriber<? super ByteBuffer> s) {
+        // As per rule 1.9 we must throw NullPointerException if the subscriber parameter is null
+        if (s == null) {
+            throw new NullPointerException("Subscription MUST NOT be null.");
+        }
 
-  @Override
-  public void subscribe(Subscriber<? super ByteBuffer> s) {
-    // As per rule 2.13, we need to throw a `java.lang.NullPointerException` if the `Subscription` is `null`
-    if (s == null) throw new NullPointerException("Subscription MUST NOT be null.");
+        // As per 2.13, this method must return normally (i.e. not throw).
+        try {
+            s.onSubscribe(
+                    new Subscription() {
+                        boolean done = false;
 
-    // As per 2.13, this method must return normally (i.e. not throw).
-    try {
-      s.onSubscribe(
-          new Subscription() {
-            boolean done = false;
-            @Override
-            public void request(long n) {
-              if (n > 0) {
-                if (!done) {
-                  s.onNext(ByteBuffer.wrap(bytes));
-                  done = true;
-                  s.onComplete();
-                }
-              } else {
-                s.onError(new IllegalArgumentException("§3.9: non-positive requests are not allowed!"));
-              }
-            }
+                        @Override
+                        public void request(long n) {
+                            if (n > 0) {
+                                if (!done) {
+                                    s.onNext(ByteBuffer.wrap(bytes));
+                                    done = true;
+                                    s.onComplete();
+                                }
+                            } else {
+                                s.onError(new IllegalArgumentException("§3.9: non-positive requests are not allowed!"));
+                            }
+                        }
 
-            @Override
-            public void cancel() {
-            }
-          }
-      );
-    } catch (Throwable ex) {
-      new IllegalStateException(s + " violated the Reactive Streams rule 2.13 " +
-          "by throwing an exception from onSubscribe.", ex)
-          // When onSubscribe fails this way, we don't know what state the
-          // s is thus calling onError may cause more crashes.
-          .printStackTrace();
+                        @Override
+                        public void cancel() {
+                        }
+                    }
+            );
+        } catch (Throwable ex) {
+            new IllegalStateException(s + " violated the Reactive Streams rule 2.13 " +
+                    "by throwing an exception from onSubscribe.", ex)
+                    // When onSubscribe fails this way, we don't know what state the
+                    // s is thus calling onError may cause more crashes.
+                    .printStackTrace();
+        }
     }
-  }
 }

core/src/main/java/software/amazon/awssdk/core/async/FileAsyncRequestBody.java

@@ -16,17 +16,16 @@
 package software.amazon.awssdk.core.async;
 
 import java.io.IOException;
-import java.io.UncheckedIOException;
 import java.nio.ByteBuffer;
 import java.nio.channels.AsynchronousFileChannel;
 import java.nio.channels.CompletionHandler;
-import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.StandardOpenOption;
 import java.util.concurrent.atomic.AtomicLong;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
 import software.amazon.awssdk.annotations.SdkInternalApi;
+import software.amazon.awssdk.core.util.async.NoopSubscription;
 import software.amazon.awssdk.utils.builder.SdkBuilder;
 
 /**
@@ -43,7 +42,7 @@ final class FileAsyncRequestBody implements AsyncRequestBody {
     /**
      * File to read.
      */
-    private final Path file;
+    private final Path path;
 
     /**
      * Size (in bytes) of ByteBuffer chunks read from the file and delivered to the subscriber.
@@ -52,18 +51,33 @@ final class FileAsyncRequestBody implements AsyncRequestBody {
 
 
     private FileAsyncRequestBody(DefaultBuilder builder) {
-        this.file = builder.path;
+        this.path = builder.path;
         this.chunkSizeInBytes = builder.chunkSizeInBytes == null ? DEFAULT_CHUNK_SIZE : builder.chunkSizeInBytes;
     }
 
     @Override
     public long contentLength() {
-        return file.toFile().length();
+        return path.toFile().length();
     }
 
     @Override
     public void subscribe(Subscriber<? super ByteBuffer> s) {
-        s.onSubscribe(new FileSubscription(file, s, chunkSizeInBytes));
+        try {
+            AsynchronousFileChannel channel = openInputChannel(this.path);
+
+            // We need to synchronize here because the subscriber could call
+            // request() from within onSubscribe which would potentially
+            // trigger onNext before onSubscribe is finished.
+            Subscription subscription = new FileSubscription(channel, s, chunkSizeInBytes);
+            synchronized (subscription) {
+                s.onSubscribe(subscription);
+            }
+        } catch (IOException e) {
+            // subscribe() must return normally, so we need to signal the
+            // failure to open via onError() once onSubscribe() is signaled.
+            s.onSubscribe(new NoopSubscription(s));
+            s.onError(e);
+        }
     }
 
     /**
@@ -135,37 +149,43 @@ public FileAsyncRequestBody build() {
      * Reads the file for one subscriber.
      */
     private static class FileSubscription implements Subscription {
-
         private final AsynchronousFileChannel inputChannel;
         private final Subscriber<? super ByteBuffer> subscriber;
         private final int chunkSize;
 
         private long position = 0;
         private AtomicLong outstandingDemand = new AtomicLong(0);
         private boolean writeInProgress = false;
+        private volatile boolean done = false;
 
-        private FileSubscription(Path file, Subscriber<? super ByteBuffer> subscriber, int chunkSize) {
-            this.inputChannel = openInputChannel(file);
+        private FileSubscription(AsynchronousFileChannel inputChannel, Subscriber<? super ByteBuffer> subscriber, int chunkSize) {
+            this.inputChannel = inputChannel;
             this.subscriber = subscriber;
             this.chunkSize = chunkSize;
         }
 
         @Override
         public void request(long n) {
+            if (done) {
+                return;
+            }
+
             if (n < 1) {
                 IllegalArgumentException ex =
-                    new IllegalArgumentException(subscriber + " violated the Reactive Streams rule 3.9 by requesting a non-positive number of elements.");
-                subscriber.onError(ex);
-            } else
+                    new IllegalArgumentException(subscriber + " violated the Reactive Streams rule 3.9 by requesting a "
+                            + "non-positive number of elements.");
+                signalOnError(ex);
+            } else {
                 try {
-                    long initialDemand = outstandingDemand.get();
-                    long newDemand = initialDemand + n;
-                    if (newDemand < 1) {
-                        // As governed by rule 3.17, when demand overflows `Long.MAX_VALUE` we treat the signalled demand as "effectively unbounded"
-                        outstandingDemand.set(Long.MAX_VALUE);
-                    } else {
-                        outstandingDemand.set(newDemand);
-                    }
+                    // As governed by rule 3.17, when demand overflows `Long.MAX_VALUE` we treat the signalled demand as
+                    // "effectively unbounded"
+                    outstandingDemand.getAndUpdate(initialDemand -> {
+                        if (Long.MAX_VALUE - initialDemand < n) {
+                            return Long.MAX_VALUE;
+                        } else {
+                            return initialDemand + n;
+                        }
+                    });
 
                     synchronized (this) {
                         if (!writeInProgress) {
@@ -174,47 +194,54 @@ public void request(long n) {
                         }
                     }
                 } catch (Exception e) {
-                    subscriber.onError(e);
+                    signalOnError(e);
                 }
+            }
         }
 
         @Override
         public void cancel() {
-            closeFile();
+            synchronized (this) {
+                if (!done) {
+                    done = true;
+                    closeFile();
+                }
+            }
         }
 
         private void readData() {
             // It's possible to have another request for data come in after we've closed the file.
             if (!inputChannel.isOpen()) {
                 return;
             }
+
             final ByteBuffer buffer = ByteBuffer.allocate(chunkSize);
             inputChannel.read(buffer, position, buffer, new CompletionHandler<Integer, ByteBuffer>() {
                 @Override
                 public void completed(Integer result, ByteBuffer attachment) {
                     if (result > 0) {
                         attachment.flip();
                         position += attachment.remaining();
-                        subscriber.onNext(attachment);
+                        signalOnNext(attachment);
                         // If we have more permits, queue up another read.
                         if (outstandingDemand.decrementAndGet() > 0) {
                             readData();
                             return;
                         }
                     } else {
                         // Reached the end of the file, notify the subscriber and cleanup
-                        subscriber.onComplete();
+                        signalOnComplete();
                         closeFile();
                     }
 
-                    synchronized (FileSubscription.this) {
+                    synchronized (this) {
                         writeInProgress = false;
                     }
                 }
 
                 @Override
                 public void failed(Throwable exc, ByteBuffer attachment) {
-                    subscriber.onError(exc);
+                    signalOnError(exc);
                     closeFile();
                 }
             });
@@ -224,20 +251,38 @@ private void closeFile() {
             try {
                 inputChannel.close();
             } catch (IOException e) {
-                throw new UncheckedIOException(e);
+                signalOnError(e);
             }
         }
 
-    }
+        private void signalOnNext(ByteBuffer bb) {
+            synchronized (this) {
+                if (!done) {
+                    subscriber.onNext(bb);
+                }
+            }
+        }
 
-    private static AsynchronousFileChannel openInputChannel(Path path) {
-        try {
-            if (!Files.exists(path)) Files.createFile(path);
-            return AsynchronousFileChannel.open(path, StandardOpenOption.READ);
-        } catch (IOException e) {
-            throw new UncheckedIOException(e);
+        private void signalOnComplete() {
+            synchronized (this) {
+                if (!done) {
+                    subscriber.onComplete();
+                    done = true;
+                }
+            }
         }
-    }
 
+        private void signalOnError(Throwable t) {
+            synchronized (this) {
+                if (!done) {
+                    subscriber.onError(t);
+                    done = true;
+                }
+            }
+        }
+    }
 
+    private static AsynchronousFileChannel openInputChannel(Path path) throws IOException {
+        return AsynchronousFileChannel.open(path, StandardOpenOption.READ);
+    }
 }