Cancel the subscription after receiving all of the pertinent emissions (reactive-streams#259).

ldaley · ldaley · commit 755e4bf5fbb2 · 2015-04-06T09:30:47.000+10:00
diff --git a/tck/src/main/java/org/reactivestreams/tck/PublisherVerification.java b/tck/src/main/java/org/reactivestreams/tck/PublisherVerification.java
@@ -1044,6 +1044,8 @@ public void onNext(T element) {
               if (callsCounter > 0) {
                 subscription.value().request(Long.MAX_VALUE - 1);
                 callsCounter--;
+              } else {
+                  subscription.value().cancel();
               }
             } else {
               env.flop(String.format("Subscriber::onNext(%s) called before Subscriber::onSubscribe", element));

Original file line number	Diff line number	Diff line change
`@@ -1044,6 +1044,8 @@ public void onNext(T element) {`
`1044`	`1044`	`if (callsCounter > 0) {`
`1045`	`1045`	`subscription.value().request(Long.MAX_VALUE - 1);`
`1046`	`1046`	`callsCounter--;`
	`1047`	`+ } else {`
	`1048`	`+ subscription.value().cancel();`
`1047`	`1049`	`}`
`1048`	`1050`	`} else {`
`1049`	`1051`	`env.flop(String.format("Subscriber::onNext(%s) called before Subscriber::onSubscribe", element));`