Add TCK tests for reactive streams implementations #519

dagnir · 2018-06-04T18:53:22Z

Add TCK tests for reactive streams implementations.

Builds off #407 from @ktoso.

Description

Add TCK tests for reactive streams implementations. This PR also adds a snippet in CONTRIBUTING.md to make it a requirement for all future PR's to include necessary TCK tests if it includes reactive streams implementations.

Motivation and Context

In order to ensure proper interoperability with other reactive streams libraries, we should ensure that our implementations of the interfaces follow the spec.

Testing

Add TCK tests for all current implementations.

Screenshots (if appropriate)

Types of changes

Bug fix (non-breaking change which fixes an issue)
New feature (non-breaking change which adds functionality)

Checklist

I have read the CONTRIBUTING document
Local run of mvn install succeeds
My code follows the code style of this project
My change requires a change to the Javadoc documentation
I have updated the Javadoc documentation accordingly
I have read the README document
I have added tests to cover my changes
All new and existing tests passed
A short description of the change has been added to the CHANGELOG

License

I confirm that this pull request can be released under the Apache 2 license

varunnvs92 · 2018-06-04T19:40:33Z

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

-                            subscriber.onComplete();
+    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) {


Here 's' is Subscriber?

I think the comment is wrong, this is the wording for 2.13:

Calling onSubscribe, onNext, onError or onComplete MUST return normally except when any provided parameter is null in which case it MUST throw a java.lang.NullPointerException to the caller, for all other situations the only legal way for a Subscriber to signal failure is by cancelling its Subscription. In the case that this rule is violated, any associated Subscription to the Subscriber MUST be considered as cancelled, and the caller MUST raise this error condition in a fashion that is adequate for the runtime environment.

I'll update it

And 1.9:

Publisher.subscribe MUST call onSubscribe on the provided Subscriber prior to any other signals to that Subscriber and MUST return normally, except when the provided Subscriber is null in which case it MUST throw a java.lang.NullPointerException to the caller, for all other situations the only legal way to signal failure (or reject the Subscriber) is by calling onError (after calling onSubscribe).

shorea · 2018-06-04T19:16:12Z

core/pom.xml

+        <dependency>
+            <groupId>org.reactivestreams</groupId>
+            <artifactId>reactive-streams-tck</artifactId>
+            <version>${reactive-streams.version}</version>


Can we just have the version in our dependency management like we do for other deps?

shorea · 2018-06-04T19:16:40Z

core/pom.xml

@@ -115,11 +115,35 @@
            <artifactId>utils</artifactId>
            <version>${awsjavasdk.version}</version>
        </dependency>
+        <dependency>


Huh how did we not have this before?

Looks like it was being pulled in transitively via the SPI:

mvn dependency:tree:

... [INFO] +- software.amazon.awssdk:http-client-spi:jar:2.0.0-preview-11-SNAPSHOT:compile [INFO] | +- software.amazon.awssdk:annotations:jar:2.0.0-preview-11-SNAPSHOT:compile [INFO] | \- org.reactivestreams:reactive-streams:jar:1.0.2:compile ...

shorea · 2018-06-04T19:41:20Z

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

+
+                        @Override
+                        public void request(long n) {
+                            if (n > 0) {


Is there opportunity to extract some of these checks into a decorator?

I think that might be a little hard to do in a general way. For example like with the async file publisher, we need to synchronize the interactions with the subscriber but for this one we don't need to do that.

shorea · 2018-06-04T19:46:41Z

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

+                    "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();


This seems weird. So we just log the error?

I don't think we need to log it but since onSubscribe must always return normally unless s is null, then we have to swallow it here.

Since 2.13 requires the Subscriber implementer to make onSubscribe return normally to be in spec, I don't think this will be a big issue.

shorea · 2018-06-04T19:51:53Z

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

                            readData();
                            return;
                        }
                    } else {
                        // Reached the end of the file, notify the subscriber and cleanup
-                        subscriber.onComplete();
+                        synchronized (subscriber) {


Why does this need to be synchronized?

Interactions with the subscriber need external synchrnozation. Without it, it's possible for a race condition between request(-1) triggering onError and this call to onComplete from the handler thread for example

Do we have any logic that would prevent both onError and onComplete from being called on the same subscriber? Even with synchronization that can still happen right?

Yeah I think that can with this implementation but I think that's a separate issue.

The purpose of this synchronization is 1.3:

onSubscribe, onNext, onError and onComplete signaled to a Subscriber MUST be signaled in a thread-safe manner—and if performed by multiple threads—use external synchronization.

I think the rule you're referring to is 1.7:

Once a terminal state has been signaled (onError, onComplete) it is REQUIRED that no further signals occur.

We'd need additional state tracking to comply with 1.7

I see. So does the TCK not test the 1.7 requirement?

I guess not. I'll go ahead and add some state tracking here

Found these: https://github.com/reactive-streams/reactive-streams-jvm/blob/43fd6d18d45484575facf42b82f5e6bffe35a486/tck/src/main/java/org/reactivestreams/tck/PublisherVerification.java#L428-L447

But it looks like one is not implemented at the moment, and there's no test that checks signaling onError after onComplete or vice versa

shorea · 2018-06-04T19:52:27Z

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

    }

+    private static class NoopSubscription implements Subscription {


Would this be more useful in a util package?

Moved to util package

zoewangg · 2018-06-04T20:41:02Z

core/pom.xml

+        <dependency>
+            <groupId>com.google.jimfs</groupId>
+            <artifactId>jimfs</artifactId>
+            <version>1.1</version>


minor, can we have all versions defined in parent pom properties? it's easier to manage versions that way.

zoewangg · 2018-06-04T20:43:59Z

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

-                });
+            );
+        } catch (Throwable ex) {
+            new IllegalStateException(s + " violated the Reactive Streams rule 2.13 " +


do we want to log Subscriber here? the implementations might not override toString

I think it's okay, it's just a diagnostic. Correct implementations should return normally from onSubscribe so it should be it should be rare that we hit this

zoewangg · 2018-06-04T20:44:39Z

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

@@ -45,26 +41,35 @@
    /**
     * File to read.
     */
-    private final File file;
+    private final Path file;


minor, can we rename it to path?

zoewangg · 2018-06-04T20:50:56Z

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

+
+public class SingleByteArrayAsyncRequestProviderTckTest extends org.reactivestreams.tck.PublisherVerification<ByteBuffer> {
+
+  public SingleByteArrayAsyncRequestProviderTckTest() {


nit: formatting seems a bit off in this file

zoewangg · 2018-06-04T20:52:18Z

pom.xml

@@ -257,6 +260,17 @@
                <artifactId>rxjava</artifactId>
                <version>2.1.9</version>
            </dependency>
+            <dependency>


do we need the reactive-streams dependencies in here?

Not sure what you mean, this is a entry in our <dependencyManagement> block

shorea · 2018-06-05T00:58:11Z

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

                }
            }
        }

+        private void signalOnNext(ByteBuffer bb) {
+            if (!done.get()) {


Is there still a race condition here if onComplete signals after we check done?

Ah yeah, It could happen for onError though if we signal for the non positive demand case. onComplete and onNext are only signaled within the same thread, so I don't think that could happen for that case.

Update so all signaling to synchronizes on this

shorea

Thanks for making new commits! Makes it so much easier to review. Feel free to rebase away.

ktoso · 2018-06-05T07:50:25Z

Cool, thanks for picking this up. I'm on vacation now-ish but will try to have a look soon :-)
I see you continued the work a bit on #519 :)

solutions

filesystem

Fixes aws#405 Fixes aws#406

dagnir · 2018-06-05T18:12:04Z

@ktoso Yup, I continued the work from #407 here in #519. Feel free to update #407 if you see anything that I might have missed!

Regression from aws#519

Regression from #519

dagnir requested a review from zoewangg June 4, 2018 18:53

varunnvs92 reviewed Jun 4, 2018

View reviewed changes

shorea reviewed Jun 4, 2018

View reviewed changes

zoewangg reviewed Jun 4, 2018

View reviewed changes

zoewangg approved these changes Jun 4, 2018

View reviewed changes

shorea reviewed Jun 5, 2018

View reviewed changes

shorea approved these changes Jun 5, 2018

View reviewed changes

ktoso and others added 5 commits June 5, 2018 11:08

PR for discussion, showing Reactive Streams TCK, proposing some

7598c3a

solutions

include sample content in test file, FIXME: these should use a virtual

98c8851

filesystem

SimpleSubscriber is now spec compliant

455c116

SingleByteArrayAsyncRequestProvider is now compliant

dcfd7bc

Add TCK tests for reactive streams impls

bc78da8

Fixes aws#405 Fixes aws#406

dagnir force-pushed the pr-407-dongie branch from 96d8dfd to bc78da8 Compare June 5, 2018 18:10

dagnir merged commit 12aaaae into aws:master Jun 5, 2018

dagnir added a commit to dagnir/aws-sdk-java-v2 that referenced this pull request Jun 7, 2018

Synchronize on FileSubscription.this

221ec5c

Regression from aws#519

dagnir mentioned this pull request Jun 7, 2018

Synchronize on FileSubscription.this #531

Merged

dagnir deleted the pr-407-dongie branch June 7, 2018 04:54

dagnir added a commit that referenced this pull request Jun 7, 2018

Synchronize on FileSubscription.this

c0497c8

Regression from #519

This was referenced Jun 7, 2018

ReceiveMessageResponse.messages is null when no message is returned by SQS #454

Closed

[PR for discussion] Showing Reactive Streams TCK, proposing some #407

Closed

shorea pushed a commit that referenced this pull request Aug 3, 2018

Synchronize on FileSubscription.this

1065296

Regression from #519

		}

		private static class NoopSubscription implements Subscription {


		public class SingleByteArrayAsyncRequestProviderTckTest extends org.reactivestreams.tck.PublisherVerification<ByteBuffer> {

		public SingleByteArrayAsyncRequestProviderTckTest() {

Add TCK tests for reactive streams implementations #519

Add TCK tests for reactive streams implementations #519

Uh oh!

Conversation

dagnir commented Jun 4, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

Motivation and Context

Testing

Screenshots (if appropriate)

Types of changes

Checklist

License

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dagnir Jun 4, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dagnir commented Jun 4, 2018 •

edited

Loading

dagnir Jun 4, 2018 •

edited

Loading