GitHub PullRequestReviewCommentEvent at 2014-04-14 12:41:46
Repository: reactive-streams
Actor: ktoso
Payload: {"comment":{"url":"https://api.github.com/repos/reactive-streams/reactive-streams/pulls/comments/11581878","id":11581878,"diff_hunk":"@@ -0,0 +1,573 @@\n+package org.reactivestreams.tck;\n+\n+import org.reactivestreams.api.Processor;\n+import org.reactivestreams.spi.Publisher;\n+import org.reactivestreams.spi.Subscriber;\n+import org.reactivestreams.spi.Subscription;\n+import org.reactivestreams.tck.TestEnvironment.ManualPublisher;\n+import org.reactivestreams.tck.TestEnvironment.ManualSubscriber;\n+import org.reactivestreams.tck.TestEnvironment.ManualSubscriberWithSubscriptionSupport;\n+import org.reactivestreams.tck.TestEnvironment.Promise;\n+import org.testng.annotations.Test;\n+\n+import java.util.HashSet;\n+import java.util.Set;\n+\n+public abstract class IdentityProcessorVerification<T> {\n+\n+  private final TestEnvironment env;\n+\n+  ////////////////////// DELEGATED TO SPECS //////////////////////\n+\n+  // for delegating tests\n+  private final SubscriberVerification<T> subscriberVerification;\n+\n+  // for delegating tests\n+  private final PublisherVerification<T> publisherVerification;\n+\n+  ////////////////// END OF DELEGATED TO SPECS //////////////////\n+\n+\n+  private final int testBufferSize;\n+\n+  /**\n+   * Test class must specify the expected time it takes for the publisher to\n+   * shut itself down when the the last downstream Subscription is cancelled.\n+   * Used by `publisherSubscribeWhenInShutDownStateMustTriggerOnErrorAndNotOnSubscribe`.\n+   */\n+  public IdentityProcessorVerification(TestEnvironment env, long publisherShutdownTimeoutMillis) {\n+    this(env, publisherShutdownTimeoutMillis, TestEnvironment.TEST_BUFFER_SIZE);\n+  }\n+\n+  public IdentityProcessorVerification(final TestEnvironment env, long publisherShutdownTimeoutMillis, int testBufferSize) {\n+    this.env = env;\n+    this.testBufferSize = testBufferSize;\n+\n+    this.subscriberVerification = new SubscriberVerification<T>(env) {\n+      @Override\n+      Subscriber<T> createSubscriber(SubscriberProbe<T> probe) {\n+        return IdentityProcessorVerification.this.createSubscriber(probe);\n+      }\n+\n+      @Override\n+      Publisher<T> createHelperPublisher(int elements) {\n+        return IdentityProcessorVerification.this.createHelperPublisher(elements);\n+      }\n+    };\n+\n+    publisherVerification = new PublisherVerification<T>(env, publisherShutdownTimeoutMillis) {\n+      @Override\n+      public Publisher<T> createPublisher(int elements) {\n+        return IdentityProcessorVerification.this.createPublisher(elements);\n+      }\n+\n+      @Override\n+      public Publisher<T> createCompletedStatePublisher() {\n+        return IdentityProcessorVerification.this.createCompletedStatePublisher();\n+      }\n+\n+      @Override\n+      public Publisher<T> createErrorStatePublisher() {\n+        return IdentityProcessorVerification.this.createErrorStatePublisher();\n+      }\n+\n+    };\n+  }\n+\n+  /**\n+   * This is the main method you must implement in your test incarnation.\n+   * It must create a Processor, which simply forwards all stream elements from its upstream\n+   * to its downstream. It must be able to internally buffer the given number of elements.\n+   */\n+  public abstract Processor<T, T> createIdentityProcessor(int bufferSize);\n+\n+  /**\n+   * Helper method required for running the Publisher rules against a Processor.\n+   * It must create a Publisher for a stream with exactly the given number of elements.\n+   * If `elements` is zero the produced stream must be infinite.\n+   * The stream must not produce the same element twice (in case of an infinite stream this requirement\n+   * is relaxed to only apply to the elements that are actually requested during all tests).\n+   */\n+  public abstract Publisher<T> createHelperPublisher(int elements);\n+\n+  /**\n+   * Override in your test if you want to enable completed-state verification.\n+   * If you don't override the respective tests will be ignored.\n+   */\n+  public Publisher<T> createCompletedStatePublisher() {\n+    return null;\n+  }\n+\n+  /**\n+   * Override in your test if you want to enable error-state verification.\n+   * If you don't override the respective tests will be ignored.\n+   */\n+  public Publisher<T> createErrorStatePublisher() {\n+    return null;\n+  }\n+\n+  ////////////////////// PUBLISHER RULES VERIFICATION ///////////////////////////\n+\n+  // A Processor\n+  //   must obey all Publisher rules on its producing side\n+  public Publisher<T> createPublisher(int elements) {\n+    Processor<T, T> processor = createIdentityProcessor(testBufferSize);\n+    Publisher<T> pub = createHelperPublisher(elements);\n+    pub.subscribe(processor.getSubscriber());\n+    return processor.getPublisher(); // we run the PublisherVerification against this\n+  }\n+\n+  // A Publisher\n+  //   must support a pending element count up to 2^63-1 (Long.MAX_VALUE) and provide for overflow protection\n+  @Test\n+  public void mustSupportAPendingElementCountUpToLongMaxValue() throws Exception {\n+    new TestSetup(env, testBufferSize) {{\n+      TestEnvironment.ManualSubscriber<T> sub = newSubscriber();\n+      sub.requestMore(Integer.MAX_VALUE);\n+      sub.requestMore(Integer.MAX_VALUE);\n+      sub.requestMore(2); // if the Subscription only keeps an int counter without overflow protection it will now be at zero\n+\n+      final T x = sendNextTFromUpstream();\n+      expectNextElement(sub, x);\n+\n+      final T y = sendNextTFromUpstream();\n+      expectNextElement(sub, y);\n+\n+      // to avoid error messages during test harness shutdown\n+      sendCompletion();\n+      sub.expectCompletion(env.defaultTimeoutMillis());\n+\n+      env.verifyNoAsyncErrors();\n+    }};\n+  }\n+\n+  @Test\n+  public void mustCallOnCompleteOnASubscriberAfterHavingProducedTheFinalStreamElementToIt() throws Exception {\n+    publisherVerification.mustCallOnCompleteOnASubscriberAfterHavingProducedTheFinalStreamElementToIt();\n+  }\n+\n+  @Test\n+  public void mustStartProducingWithTheOldestStillAvailableElementForASubscriber() {\n+    publisherVerification.mustStartProducingWithTheOldestStillAvailableElementForASubscriber();\n+  }\n+\n+  // A Publisher\n+  //   must call `onError` on all its subscribers if it encounters a non-recoverable error\n+  @Test\n+  public void mustCallOnErrorOnAllItsSubscribersIfItEncountersANonRecoverableError() throws Exception {\n+    new TestSetup(env, testBufferSize) {\n+      {\n+        ManualSubscriberWithErrorCollection<T> sub1 = new ManualSubscriberWithErrorCollection<T>(env);\n+        env.subscribe(processor.getPublisher(), sub1);\n+        ManualSubscriberWithErrorCollection<T> sub2 = new ManualSubscriberWithErrorCollection<T>(env);\n+        env.subscribe(processor.getPublisher(), sub2);\n+\n+        sub1.requestMore(1);\n+        expectRequestMore();\n+        final T x = sendNextTFromUpstream();\n+        expectNextElement(sub1, x);\n+        sub1.requestMore(1);\n+\n+        // sub1 now has received and element and has 1 pending\n+        // sub2 has not yet requested anything\n+\n+        Exception ex = new RuntimeException(\"Test exception\");\n+        sendError(ex);\n+        sub1.expectError(ex);\n+        sub2.expectError(ex);\n+\n+        env.verifyNoAsyncErrors();\n+      }\n+    };\n+  }\n+\n+  @Test\n+  public void mustNotCallOnCompleteOrOnErrorMoreThanOncePerSubscriber() {\n+    publisherVerification.mustNotCallOnCompleteOrOnErrorMoreThanOncePerSubscriber();\n+  }\n+\n+  ////////////////////// SUBSCRIBER RULES VERIFICATION ///////////////////////////\n+\n+  // A Processor\n+  //   must obey all Subscriber rules on its consuming side\n+  public Subscriber<T> createSubscriber(final SubscriberVerification.SubscriberProbe<T> probe) {\n+    Processor<T, T> processor = createIdentityProcessor(testBufferSize);\n+    processor.getPublisher().subscribe(\n+        new Subscriber<T>() {\n+          public void onSubscribe(final Subscription subscription) {\n+            probe.registerOnSubscribe(\n+                new SubscriberVerification.SubscriberPuppet() {\n+                  public void triggerShutdown() {\n+                    subscription.cancel();\n+                  }\n+\n+                  public void triggerRequestMore(int elements) {\n+                    subscription.requestMore(elements);\n+                  }\n+\n+                  public void triggerCancel() {\n+                    subscription.cancel();\n+                  }\n+                });\n+          }\n+\n+          public void onNext(T element) {\n+            probe.registerOnNext(element);\n+          }\n+\n+          public void onComplete() {\n+            probe.registerOnComplete();\n+          }\n+\n+          public void onError(Throwable cause) {\n+            probe.registerOnError(cause);\n+          }\n+        });\n+\n+    return processor.getSubscriber(); // we run the SubscriberVerification against this\n+  }\n+\n+  ////////////////////// OTHER SPEC RULE VERIFICATION ///////////////////////////\n+\n+  // A Processor\n+  //   must cancel its upstream Subscription if its last downstream Subscription has been cancelled\n+  @Test\n+  public void mustCancelItsUpstreamSubscriptionIfItsLastDownstreamSubscriptionHasBeenCancelled() throws Exception {\n+    new TestSetup(env, testBufferSize) {{\n+      TestEnvironment.ManualSubscriber<T> sub = newSubscriber();\n+      sub.cancel();\n+      expectCancelling();\n+\n+      env.verifyNoAsyncErrors();\n+    }};\n+  }\n+\n+  // A Processor\n+  //   must immediately pass on `onError` events received from its upstream to its downstream\n+  @Test\n+  public void mustImmediatelyPassOnOnErrorEventsReceivedFromItsUpstreamToItsDownstream() throws Exception {\n+    new TestSetup(env, testBufferSize) {\n+      {\n+        ManualSubscriberWithErrorCollection<T> sub = new ManualSubscriberWithErrorCollection<T>(env);\n+        env.subscribe(processor.getPublisher(), sub);\n+\n+        Exception ex = new RuntimeException(\"Test exception\");\n+        sendError(ex);\n+        sub.expectError(ex); // \"immediately\", i.e. without a preceding requestMore\n+\n+        env.verifyNoAsyncErrors();\n+      }\n+    };\n+  }\n+\n+  // A Processor\n+  //   must be prepared to receive incoming elements from its upstream even if a downstream subscriber has not requested anything yet\n+  @Test\n+  public void mustBePreparedToReceiveIncomingElementsFromItsUpstreamEvenIfADownstreamSubscriberHasNotRequestedYet() throws Exception {\n+    new TestSetup(env, testBufferSize) {\n+      {\n+        ManualSubscriber<T> sub = newSubscriber();\n+        final T x = sendNextTFromUpstream();\n+        sub.expectNone(50);\n+        final T y = sendNextTFromUpstream();\n+        sub.expectNone(50);\n+\n+        sub.requestMore(2);\n+        sub.expectNext(x);\n+        sub.expectNext(y);\n+\n+        // to avoid error messages during test harness shutdown\n+        sendCompletion();\n+        sub.expectCompletion(env.defaultTimeoutMillis());\n+\n+        env.verifyNoAsyncErrors();\n+      }\n+    };\n+  }\n+\n+  /////////////////////// DELEGATED TESTS, A PROCESSOR \"IS A\" SUBSCRIBER //////////////////////\n+\n+  @Test\n+  public void exerciseHappyPath() throws InterruptedException {\n+    subscriberVerification.exerciseHappyPath();\n+  }\n+\n+  @Test\n+  public void onSubscribeAndOnNextMustAsynchronouslyScheduleAnEvent() {\n+    subscriberVerification.onSubscribeAndOnNextMustAsynchronouslyScheduleAnEvent();\n+  }\n+\n+  @Test\n+  public void onCompleteAndOnErrorMustAsynchronouslyScheduleAnEvent() {\n+    subscriberVerification.onCompleteAndOnErrorMustAsynchronouslyScheduleAnEvent();\n+  }\n+\n+  @Test\n+  public void mustNotAcceptAnOnSubscribeEventIfItAlreadyHasAnActiveSubscription() throws InterruptedException {\n+    subscriberVerification.mustNotAcceptAnOnSubscribeEventIfItAlreadyHasAnActiveSubscription();\n+  }\n+\n+  @Test\n+  public void mustCallSubscriptionCancelDuringShutdownIfItStillHasAnActiveSubscription() throws InterruptedException {\n+    subscriberVerification.mustCallSubscriptionCancelDuringShutdownIfItStillHasAnActiveSubscription();\n+  }\n+\n+  @Test\n+  public void mustEnsureThatAllCallsOnASubscriptionTakePlaceFromTheSameThreadOrProvideExternalSync() {\n+    subscriberVerification.mustEnsureThatAllCallsOnASubscriptionTakePlaceFromTheSameThreadOrProvideExternalSync();\n+  }\n+\n+  @Test\n+  public void mustBePreparedToReceiveOneOrMoreOnNextEventsAfterHavingCalledSubscriptionCancel() throws InterruptedException {\n+    subscriberVerification.mustBePreparedToReceiveOneOrMoreOnNextEventsAfterHavingCalledSubscriptionCancel();\n+  }\n+\n+  @Test\n+  public void mustBePreparedToReceiveAnOnCompleteEventWithAPrecedingSubscriptionRequestMore() throws InterruptedException {\n+    subscriberVerification.mustBePreparedToReceiveAnOnCompleteEventWithAPrecedingSubscriptionRequestMore();\n+  }\n+\n+  @Test\n+  public void mustBePreparedToReceiveAnOnCompleteEventWithoutAPrecedingSubscriptionRequestMore() throws InterruptedException {\n+    subscriberVerification.mustBePreparedToReceiveAnOnCompleteEventWithoutAPrecedingSubscriptionRequestMore();\n+  }\n+\n+  @Test\n+  public void mustBePreparedToReceiveAnOnErrorEventWithAPrecedingSubscriptionRequestMore() throws InterruptedException {\n+    subscriberVerification.mustBePreparedToReceiveAnOnErrorEventWithAPrecedingSubscriptionRequestMore();\n+  }\n+\n+  @Test\n+  public void mustBePreparedToReceiveAnOnErrorEventWithoutAPrecedingSubscriptionRequestMore() throws InterruptedException {\n+    subscriberVerification.mustBePreparedToReceiveAnOnErrorEventWithoutAPrecedingSubscriptionRequestMore();\n+  }\n+\n+  @Test\n+  public void mustMakeSureThatAllCallsOnItsMethodsHappenBeforeTheProcessingOfTheRespectiveEvents() {\n+    subscriberVerification.mustMakeSureThatAllCallsOnItsMethodsHappenBeforeTheProcessingOfTheRespectiveEvents();\n+  }\n+","path":"tck/src/main/java/org/reactivestreams/tck/IdentityProcessorVerification.java","position":350,"original_position":349,"commit_id":"b05dbcdca0f1dda9f9012293f68e2b0680fc824f","original_commit_id":"f487b33e53d818acbbd60284c2cc79c269e800a2","user":{"login":"ktoso","id":120979,"avatar_url":"https://avatars.githubusercontent.com/u/120979?","gravatar_id":"03cb20b97f6a14701c24c4e088b6af87","url":"https://api.github.com/users/ktoso","html_url":"https://github.com/ktoso","followers_url":"https://api.github.com/users/ktoso/followers","following_url":"https://api.github.com/users/ktoso/following{/other_user}","gists_url":"https://api.github.com/users/ktoso/gists{/gist_id}","starred_url":"https://api.github.com/users/ktoso/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/ktoso/subscriptions","organizations_url":"https://api.github.com/users/ktoso/orgs","repos_url":"https://api.github.com/users/ktoso/repos","events_url":"https://api.github.com/users/ktoso/events{/privacy}","received_events_url":"https://api.github.com/users/ktoso/received_events","type":"User","site_admin":false},"body":"Added such test in https://github.com/ktoso/reactive-streams/commit/b05dbcdca0f1dda9f9012293f68e2b0680fc824f and actually got one missing delegate(!). Fixed.","created_at":"2014-04-14T12:41:46Z","updated_at":"2014-04-14T12:41:46Z","html_url":"https://github.com/reactive-streams/reactive-streams/pull/15#discussion_r11581878","pull_request_url":"https://api.github.com/repos/reactive-streams/reactive-streams/pulls/15","_links":{"self":{"href":"https://api.github.com/repos/reactive-streams/reactive-streams/pulls/comments/11581878"},"html":{"href":"https://github.com/reactive-streams/reactive-streams/pull/15#discussion_r11581878"},"pull_request":{"href":"https://api.github.com/repos/reactive-streams/reactive-streams/pulls/15"}}}}