SimpleSubscriber is now spec compliant

Author: ktoso

core/pom.xml

@@ -138,6 +138,12 @@
             <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>
     <build>
         <resources>

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

@@ -15,15 +15,13 @@
 
 package software.amazon.awssdk.core.async;
 
-import java.io.File;
 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.Paths;
 import java.nio.file.StandardOpenOption;
 import java.util.concurrent.atomic.AtomicLong;
 import org.reactivestreams.Subscriber;
@@ -45,21 +43,22 @@ final class FileAsyncRequestBody implements AsyncRequestBody {
     /**
      * File to read.
      */
-    private final File file;
+    private final Path file;
 
     /**
      * Size (in bytes) of ByteBuffer chunks read from the file and delivered to the subscriber.
      */
     private final int chunkSizeInBytes;
 
+
     private FileAsyncRequestBody(DefaultBuilder builder) {
-        this.file = builder.path.toFile();
+        this.file = builder.path;
         this.chunkSizeInBytes = builder.chunkSizeInBytes == null ? DEFAULT_CHUNK_SIZE : builder.chunkSizeInBytes;
     }
 
     @Override
     public long contentLength() {
-        return file.length();
+        return file.toFile().length();
     }
 
     @Override
@@ -142,29 +141,41 @@ private static class FileSubscription implements Subscription {
         private final int chunkSize;
 
         private long position = 0;
-        private AtomicLong outstandingRequests = new AtomicLong(0);
+        private AtomicLong outstandingDemand = new AtomicLong(0);
         private boolean writeInProgress = false;
 
-        private FileSubscription(File file, Subscriber<? super ByteBuffer> subscriber, int chunkSize) {
+        private FileSubscription(Path file, Subscriber<? super ByteBuffer> subscriber, int chunkSize) {
             this.inputChannel = openInputChannel(file);
             this.subscriber = subscriber;
             this.chunkSize = chunkSize;
         }
 
         @Override
         public void request(long n) {
-            try {
-                outstandingRequests.addAndGet(n);
+            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
+                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);
+                    }
 
-                synchronized (this) {
-                    if (!writeInProgress) {
-                        writeInProgress = true;
-                        readData();
+                    synchronized (this) {
+                        if (!writeInProgress) {
+                            writeInProgress = true;
+                            readData();
+                        }
                     }
+                } catch (Exception e) {
+                    subscriber.onError(e);
                 }
-            } catch (Exception e) {
-                subscriber.onError(e);
-            }
         }
 
         @Override
@@ -186,7 +197,7 @@ public void completed(Integer result, ByteBuffer attachment) {
                         position += attachment.remaining();
                         subscriber.onNext(attachment);
                         // If we have more permits, queue up another read.
-                        if (outstandingRequests.decrementAndGet() > 0) {
+                        if (outstandingDemand.decrementAndGet() > 0) {
                             readData();
                             return;
                         }
@@ -219,12 +230,9 @@ private void closeFile() {
 
     }
 
-    private static AsynchronousFileChannel openInputChannel(File file) {
+    private static AsynchronousFileChannel openInputChannel(Path path) {
         try {
-            final Path path = Paths.get(file.getAbsolutePath());
-            if (!Files.exists(path)) {
-                Files.createFile(path);
-            }
+            if (!Files.exists(path)) Files.createFile(path);
             return AsynchronousFileChannel.open(path, StandardOpenOption.READ);
         } catch (IOException e) {
             throw new UncheckedIOException(e);

core/src/test/java/software/amazon/awssdk/core/async/FileAsyncRequestPublisherTckTest.java

@@ -29,37 +29,51 @@
 import java.nio.charset.StandardCharsets;
 import java.nio.file.FileSystem;
 import java.nio.file.Files;
+import java.nio.file.Path;
 import java.nio.file.Paths;
+import java.util.Arrays;
+import java.util.List;
 
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.mockito.Matchers.any;
-import static org.mockito.Mockito.*;
+import com.google.common.jimfs.Configuration;
+import com.google.common.jimfs.Jimfs;
 
 public class FileAsyncRequestPublisherTckTest extends org.reactivestreams.tck.PublisherVerification<ByteBuffer> {
 
-    final File example = File.createTempFile("example", ".tmp");
-    final File fileDoesNotExist = new File(example.getPath() + "-does-not-exist");
+    // same as `FileAsyncRequestProvider.DEFAULT_CHUNK_SIZE`:
+    final int DEFAULT_CHUNK_SIZE = 16 * 1024;
+    final int ELEMENTS = 1000;
+
+    // mock file system:
+    final FileSystem fs = Jimfs.newFileSystem(Configuration.unix());
+
+    final Path testFile;
+    final Path doestNotExist;
 
     public FileAsyncRequestPublisherTckTest() throws IOException {
         super(new TestEnvironment());
+        testFile = Files.createFile(fs.getPath("/test-file.tmp"));
+
+        doestNotExist = new File("does-not-exist").toPath();
+
+        final BufferedWriter writer = Files.newBufferedWriter(testFile);
+
+        final char[] chars = new char[DEFAULT_CHUNK_SIZE];
+        Arrays.fill(chars, 'A');
 
-        BufferedWriter writer = new BufferedWriter(new FileWriter(example));
-        writer.write("Hello world\n");
-        writer.write("Hello world\n");
-        writer.write("Hello world\n");
-        writer.write("Hello world\n");
-        writer.write("Hello world\n");
-        writer.flush();
-        writer.close();
+        for (int i = 0; i < ELEMENTS; i++) {
+            writer.write(chars); // write one chunk
+        }
     }
 
     @Override
-    public Publisher<ByteBuffer> createPublisher(long l) {
-        return AsyncRequestProvider.fromFile(example.toPath());
+    public Publisher<ByteBuffer> createPublisher(long elements) {
+        if (elements < ELEMENTS) return AsyncRequestProvider.fromFile(testFile);
+        else return null; // we don't support more elements
     }
 
     @Override
     public Publisher<ByteBuffer> createFailedPublisher() {
-        return AsyncRequestProvider.fromFile(fileDoesNotExist.toPath());
+        // tests properly failing on non existing files:
+        return AsyncRequestProvider.fromFile(doestNotExist);
     }
 }

http-client-spi/src/main/java/software/amazon/awssdk/http/async/SimpleSubscriber.java

@@ -16,38 +16,59 @@
 package software.amazon.awssdk.http.async;
 
 import java.nio.ByteBuffer;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
 
 /**
  * Simple subscriber that does no backpressure and doesn't care about errors or completion.
  */
-// TODO provided example how to cover it using the TCK, you may want to put those tests to a separate project,
-// TODO if you'd prefer to do so
 public class SimpleSubscriber implements Subscriber<ByteBuffer> {
 
     private final Consumer<ByteBuffer> consumer;
+    private final AtomicReference<Subscription> subscription = new AtomicReference<>();
 
     public SimpleSubscriber(Consumer<ByteBuffer> consumer) {
         this.consumer = consumer;
     }
 
     @Override
     public void onSubscribe(Subscription s) {
-        s.request(Long.MAX_VALUE);
+        // 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.");
+
+        if (subscription.get() == null) {
+            if (subscription.compareAndSet(null, s)) {
+                s.request(Long.MAX_VALUE);
+            } else onSubscribe(s); // lost race, retry (will cancel in the else branch below)
+        } else {
+            try {
+                s.cancel(); // Cancel the additional subscription
+            } catch (final Throwable t) {
+                // Subscription.cancel is not allowed to throw an exception, according to rule 3.15
+                (new IllegalStateException(s + " violated the Reactive Streams rule 3.15 by throwing an exception from cancel.", t))
+                    .printStackTrace(System.err);
+            }
+        }
     }
 
     @Override
     public void onNext(ByteBuffer byteBuffer) {
+        // Rule 2.13, null arguments must be failed on eagerly
+        if (byteBuffer == null) throw new NullPointerException("Element passed to onNext MUST NOT be null.");
+
         consumer.accept(byteBuffer);
     }
 
     @Override
     public void onError(Throwable t) {
+        if (t == null) throw new NullPointerException("Throwable passed to onError MUST NOT be null.");
+        // else, ignore
     }
 
     @Override
     public void onComplete() {
+        // ignore
     }
 }