PR for discussion, showing Reactive Streams TCK, proposing some

solutions

Author: ktoso

core/pom.xml

@@ -115,11 +115,29 @@
             <artifactId>utils</artifactId>
             <version>${awsjavasdk.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.reactivestreams</groupId>
+            <artifactId>reactive-streams</artifactId>
+            <version>${reactive-streams.version}</version>
+        </dependency>
+
         <dependency>
             <groupId>org.assertj</groupId>
             <artifactId>assertj-core</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>software.amazon.awssdk</groupId>
+            <artifactId>test-utils</artifactId>
+            <version>${awsjavasdk.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.reactivestreams</groupId>
+            <artifactId>reactive-streams-tck</artifactId>
+            <version>${reactive-streams.version}</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
     <build>
         <resources>

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

@@ -44,6 +44,15 @@ public long contentLength() {
 
     @Override
     public void subscribe(Subscriber<? super ByteBuffer> subscriber) {
+        // FIXME missing protection abiding to rule 1.9, proposal:
+        // As per rule 1.09, we need to throw a `java.lang.NullPointerException`
+        // if the `Subscriber` is `null`
+        // if (subscriber == null) throw null;
+
+        // FIXME: onSubscribe is user code, and could be ill behaved, as library we should protect from this,
+        // FIXME: This is covered by spec rule 2.13; proposal:
+        // As per 2.13, this method must return normally (i.e. not throw).
+        // try {
         subscriber.onSubscribe(
                 new Subscription() {
                     @Override
@@ -53,10 +62,25 @@ public void request(long n) {
                             subscriber.onComplete();
                         }
                     }
+                    // FIXME missing required validation code (rule 1.9):
+                    //   "Non-positive requests should be honored with IllegalArgumentException"
+                    // proposal:
+                    // else {
+                    //  subscriber.onError(new IllegalArgumentException("§3.9: non-positive requests are not allowed!"));
+                    // }
 
-                    @Override
-                    public void cancel() {
-                    }
-                });
+                @Override
+                public void cancel() {
+                }
+            }
+        );
+        // end of implementing 2.13 spec requirement
+        //  } catch (Throwable ex) {
+        //  new IllegalStateException(subscriber + " 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
+        //      // subscriber is thus calling onError may cause more crashes.
+        //      .printStackTrace();
+        //    }
     }
 }

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

@@ -69,6 +69,7 @@ public ResponseBytes<ResponseT> complete() {
     /**
      * Requests chunks sequentially and dumps them into a {@link ByteArrayOutputStream}.
      */
+    // TODO cover with Reactive Streams TCK, it's mostly ok, just a few edge cases / sanity checks should be covered AFAICS
     private class BaosSubscriber implements Subscriber<ByteBuffer> {
 
         private Subscription subscription;

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

@@ -82,6 +82,7 @@ public ResponseT complete() {
     /**
      * {@link Subscriber} implementation that writes chunks to a file.
      */
+    // FIXME cover with Reactive Streams TCK, looks ok from a first brief look, but could be missing some edge cases
     private class FileSubscriber implements Subscriber<ByteBuffer> {
 
         private volatile boolean writeInProgress = false;

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

@@ -0,0 +1,43 @@
+/*
+ * Copyright 2010-2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+package software.amazon.awssdk.core.async;
+
+import org.reactivestreams.Publisher;
+import org.reactivestreams.tck.TestEnvironment;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+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");
+
+    public FileAsyncRequestPublisherTckTest() throws IOException {
+        super(new TestEnvironment());
+    }
+
+    @Override
+    public Publisher<ByteBuffer> createPublisher(long l) {
+        return AsyncRequestProvider.fromFile(example.toPath());
+    }
+
+    @Override
+    public Publisher<ByteBuffer> createFailedPublisher() {
+        return AsyncRequestProvider.fromFile(fileDoesNotExist.toPath());
+    }
+}

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

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2010-2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+package software.amazon.awssdk.core.async;
+
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.tck.TestEnvironment;
+import software.amazon.awssdk.http.async.SimpleSubscriber;
+
+import java.nio.ByteBuffer;
+
+public class SimpleSubscriberTckTest extends org.reactivestreams.tck.SubscriberBlackboxVerification<ByteBuffer> {
+
+  public SimpleSubscriberTckTest() {
+    super(new TestEnvironment());
+  }
+
+  @Override
+  public Subscriber<ByteBuffer> createSubscriber() {
+    return new SimpleSubscriber(buffer -> {
+      // ignore
+    });
+  }
+
+  @Override
+  public ByteBuffer createElement(int i) {
+    return ByteBuffer.wrap(String.valueOf(i).getBytes());
+  }
+
+}

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

@@ -0,0 +1,42 @@
+/*
+ * Copyright 2010-2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+package software.amazon.awssdk.core.async;
+
+import org.reactivestreams.Publisher;
+import org.reactivestreams.tck.TestEnvironment;
+
+import java.nio.ByteBuffer;
+
+public class SingleByteArrayAsyncRequestProviderTckTest extends org.reactivestreams.tck.PublisherVerification<ByteBuffer> {
+
+  public SingleByteArrayAsyncRequestProviderTckTest() {
+    super(new TestEnvironment());
+  }
+
+  @Override
+  public long maxElementsFromPublisher() {
+    return super.maxElementsFromPublisher();
+  }
+
+  @Override
+  public Publisher<ByteBuffer> createPublisher(long n) {
+    return AsyncRequestProvider.fromString("Hello world");
+  }
+
+  @Override
+  public Publisher<ByteBuffer> createFailedPublisher() {
+    return null;
+  }
+}

http-client-spi/pom.xml

@@ -41,7 +41,7 @@
         <dependency>
             <groupId>org.reactivestreams</groupId>
             <artifactId>reactive-streams</artifactId>
-            <version>1.0.0.final</version>
+            <version>${reactive-streams.version}</version>
         </dependency>
         <dependency>
             <groupId>junit</groupId>
@@ -53,6 +53,12 @@
             <artifactId>assertj-core</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.reactivestreams</groupId>
+            <artifactId>reactive-streams-tck</artifactId>
+            <version>${reactive-streams.version}</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
 </project>

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

@@ -23,6 +23,8 @@
 /**
  * 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;

http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/ResponseHandler.java

@@ -170,6 +170,7 @@ private static ByteBuffer copyToByteBuffer(ByteBuf byteBuf) {
         return bb;
     }
 
+    // TODO cover using Reactive Streams TCK
     private static class PublisherAdapter implements Publisher<ByteBuffer> {
         private final StreamedHttpResponse response;
         private final ChannelHandlerContext channelContext;
@@ -220,6 +221,7 @@ public void onComplete() {
         }
     }
 
+    // TODO cover using Reactive Streams TCK (glad you seem to already have looked into the rules! :-))
     private static class FullResponseContentPublisher implements Publisher<ByteBuffer> {
         private final ChannelHandlerContext channelContext;
         private final ByteBuffer fullContent;

pom.xml

@@ -103,13 +103,16 @@
         <spring.version>3.0.7.RELEASE</spring.version>
         <freemarker.version>2.3.9</freemarker.version>
         <aspectj.version>1.8.2</aspectj.version>
-        
+
         <jre.version>1.8</jre.version>
         <httpcomponents.httpclient.version>4.5.4</httpcomponents.httpclient.version>
 
         <!-- These properties are used by cucumber tests related code -->
         <unitils.version>3.3</unitils.version>
 
+        <!-- Reactive Streams version -->
+        <reactive-streams.version>1.0.2</reactive-streams.version>
+
         <!-- Sonar -->
         <root.offset>..</root.offset>
         <sonar.java.coveragePlugin>jacoco</sonar.java.coveragePlugin>
@@ -257,6 +260,16 @@
                 <artifactId>rxjava</artifactId>
                 <version>2.1.9</version>
             </dependency>
+                <groupId>org.reactivestreams</groupId>
+                <artifactId>reactive-streams</artifactId>
+                <version>${reactive-streams.version}</version>
+            </dependency>
+            <dependency>
+                <groupId>org.reactivestreams</groupId>
+                <artifactId>reactive-streams-tck</artifactId>
+                <version>${reactive-streams.version}</version>
+                <scope>test</scope>
+            </dependency>
         </dependencies>
     </dependencyManagement>