/*

 * Copyright (c) 2019, Oracle and/or its affiliates. All rights reserved.

 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.

 *

 * This code is free software; you can redistribute it and/or modify it

 * under the terms of the GNU General Public License version 2 only, as

 * published by the Free Software Foundation.

 *

 * This code is distributed in the hope that it will be useful, but WITHOUT

 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or

 * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License

 * version 2 for more details (a copy is included in the LICENSE file that

 * accompanied this code).

 *

 * You should have received a copy of the GNU General Public License version

 * 2 along with this work; if not, write to the Free Software Foundation,

 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.

 *

 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA

 * or visit www.oracle.com if you need additional information or have any

 * questions.

 */

package org.reactivestreams.tck;

import org.reactivestreams.Processor;

import org.reactivestreams.Publisher;

import org.reactivestreams.Subscriber;

import org.reactivestreams.Subscription;

import org.reactivestreams.tck.TestEnvironment.ManualPublisher;

import org.reactivestreams.tck.TestEnvironment.ManualSubscriber;

import org.reactivestreams.tck.TestEnvironment.ManualSubscriberWithSubscriptionSupport;

import org.reactivestreams.tck.TestEnvironment.Promise;

import org.reactivestreams.tck.flow.support.Function;

import org.reactivestreams.tck.flow.support.SubscriberWhiteboxVerificationRules;

import org.reactivestreams.tck.flow.support.PublisherVerificationRules;

import org.testng.annotations.BeforeMethod;

import org.testng.annotations.Test;

import java.util.HashSet;

import java.util.Set;

public abstract class IdentityProcessorVerification<T> extends WithHelperPublisher<T>

  implements SubscriberWhiteboxVerificationRules, PublisherVerificationRules {

  private final TestEnvironment env;

  ////////////////////// DELEGATED TO SPECS //////////////////////

  // for delegating tests

  private final SubscriberWhiteboxVerification<T> subscriberVerification;

  // for delegating tests

  private final PublisherVerification<T> publisherVerification;

  ////////////////// END OF DELEGATED TO SPECS //////////////////

  // number of elements the processor under test must be able ot buffer,

  // without dropping elements. Defaults to `TestEnvironment.TEST_BUFFER_SIZE`.

  private final int processorBufferSize;

  /**

   * Test class must specify the expected time it takes for the publisher to

   * shut itself down when the the last downstream {@code Subscription} is cancelled.

   *

   * The processor will be required to be able to buffer {@code TestEnvironment.TEST_BUFFER_SIZE} elements.

   */

  @SuppressWarnings("unused")

  public IdentityProcessorVerification(final TestEnvironment env) {

    this(env, PublisherVerification.envPublisherReferenceGCTimeoutMillis(), TestEnvironment.TEST_BUFFER_SIZE);

  }

  /**

   * Test class must specify the expected time it takes for the publisher to

   * shut itself down when the the last downstream {@code Subscription} is cancelled.

   *

   * The processor will be required to be able to buffer {@code TestEnvironment.TEST_BUFFER_SIZE} elements.

   *

   * @param publisherReferenceGCTimeoutMillis used to determine after how much time a reference to a Subscriber should be already dropped by the Publisher.

   */

  @SuppressWarnings("unused")

  public IdentityProcessorVerification(final TestEnvironment env, long publisherReferenceGCTimeoutMillis) {

    this(env, publisherReferenceGCTimeoutMillis, TestEnvironment.TEST_BUFFER_SIZE);

  }

  /**

   * Test class must specify the expected time it takes for the publisher to

   * shut itself down when the the last downstream {@code Subscription} is cancelled.

   *

   * @param publisherReferenceGCTimeoutMillis used to determine after how much time a reference to a Subscriber should be already dropped by the Publisher.

   * @param processorBufferSize            number of elements the processor is required to be able to buffer.

   */

  public IdentityProcessorVerification(final TestEnvironment env, long publisherReferenceGCTimeoutMillis, int processorBufferSize) {

    this.env = env;

    this.processorBufferSize = processorBufferSize;

    this.subscriberVerification = new SubscriberWhiteboxVerification<T>(env) {

      @Override

      public Subscriber<T> createSubscriber(WhiteboxSubscriberProbe<T> probe) {

        return IdentityProcessorVerification.this.createSubscriber(probe);

      }

      @Override public T createElement(int element) {

        return IdentityProcessorVerification.this.createElement(element);

      }

      @Override

      public Publisher<T> createHelperPublisher(long elements) {

        return IdentityProcessorVerification.this.createHelperPublisher(elements);

      }

    };

    publisherVerification = new PublisherVerification<T>(env, publisherReferenceGCTimeoutMillis) {

      @Override

      public Publisher<T> createPublisher(long elements) {

        return IdentityProcessorVerification.this.createPublisher(elements);

      }

      @Override

      public Publisher<T> createFailedPublisher() {

        return IdentityProcessorVerification.this.createFailedPublisher();

      }

      @Override

      public long maxElementsFromPublisher() {

        return IdentityProcessorVerification.this.maxElementsFromPublisher();

      }

      @Override

      public long boundedDepthOfOnNextAndRequestRecursion() {

        return IdentityProcessorVerification.this.boundedDepthOfOnNextAndRequestRecursion();

      }

      @Override

      public boolean skipStochasticTests() {

        return IdentityProcessorVerification.this.skipStochasticTests();

      }

    };

  }

  /**

   * This is the main method you must implement in your test incarnation.

   * It must create a {@link Processor}, which simply forwards all stream elements from its upstream

   * to its downstream. It must be able to internally buffer the given number of elements.

   *

   * @param bufferSize number of elements the processor is required to be able to buffer.

   */

  public abstract Processor<T, T> createIdentityProcessor(int bufferSize);

  /**

   * By implementing this method, additional TCK tests concerning a "failed" publishers will be run.

   *

   * The expected behaviour of the {@link Publisher} returned by this method is hand out a subscription,

   * followed by signalling {@code onError} on it, as specified by Rule 1.9.

   *

   * If you want to ignore these additional tests, return {@code null} from this method.

   */

  public abstract Publisher<T> createFailedPublisher();

  /**

   * Override and return lower value if your Publisher is only able to produce a known number of elements.

   * For example, if it is designed to return at-most-one element, return {@code 1} from this method.

   *

   * Defaults to {@code Long.MAX_VALUE - 1}, meaning that the Publisher can be produce a huge but NOT an unbounded number of elements.

   *

   * To mark your Publisher will *never* signal an {@code onComplete} override this method and return {@code Long.MAX_VALUE},

   * which will result in *skipping all tests which require an onComplete to be triggered* (!).

   */

  public long maxElementsFromPublisher() {

    return Long.MAX_VALUE - 1;

  }

  /**

   * In order to verify rule 3.3 of the reactive streams spec, this number will be used to check if a

   * {@code Subscription} actually solves the "unbounded recursion" problem by not allowing the number of

   * recursive calls to exceed the number returned by this method.

   *

   * @see <a href="https://github.com/reactive-streams/reactive-streams-jvm#3.3">reactive streams spec, rule 3.3</a>

   * @see PublisherVerification#required_spec303_mustNotAllowUnboundedRecursion()

   */

  public long boundedDepthOfOnNextAndRequestRecursion() {

    return 1;

  }

  /**

   * Override and return {@code true} in order to skip executing tests marked as {@code Stochastic}.

   * Stochastic in this case means that the Rule is impossible or infeasible to deterministically verify—

   * usually this means that this test case can yield false positives ("be green") even if for some case,

   * the given implementation may violate the tested behaviour.

   */

  public boolean skipStochasticTests() {

    return false;

  }

  /**

   * Describes the tested implementation in terms of how many subscribers they can support.

   * Some tests require the {@code Publisher} under test to support multiple Subscribers,

   * yet the spec does not require all publishers to be able to do so, thus – if an implementation

   * supports only a limited number of subscribers (e.g. only 1 subscriber, also known as "no fanout")

   * you MUST return that number from this method by overriding it.

   */

  public long maxSupportedSubscribers() {

      return Long.MAX_VALUE;

  }

  /**

   * Override this method and return {@code true} if the {@link Processor} returned by the

   * {@link #createIdentityProcessor(int)} coordinates its {@link Subscriber}s

   * request amounts and only delivers onNext signals if all Subscribers have

   * indicated (via their Subscription#request(long)) they are ready to receive elements.

   */

  public boolean doesCoordinatedEmission() {

    return false;

  }

  ////////////////////// TEST ENV CLEANUP /////////////////////////////////////

  @BeforeMethod

  public void setUp() throws Exception {

    publisherVerification.setUp();

    subscriberVerification.setUp();

  }

  ////////////////////// PUBLISHER RULES VERIFICATION ///////////////////////////

  // A Processor

  //   must obey all Publisher rules on its publishing side

  public Publisher<T> createPublisher(long elements) {

    final Processor<T, T> processor = createIdentityProcessor(processorBufferSize);

    final Publisher<T> pub = createHelperPublisher(elements);

    pub.subscribe(processor);

    return processor; // we run the PublisherVerification against this

  }

  @Override @Test

  public void required_validate_maxElementsFromPublisher() throws Exception {

    publisherVerification.required_validate_maxElementsFromPublisher();

  }

  @Override @Test

  public void required_validate_boundedDepthOfOnNextAndRequestRecursion() throws Exception {

    publisherVerification.required_validate_boundedDepthOfOnNextAndRequestRecursion();

  }

  /////////////////////// DELEGATED TESTS, A PROCESSOR "IS A" PUBLISHER //////////////////////

  // Verifies rule: https://github.com/reactive-streams/reactive-streams-jvm#4.1

  @Test

  public void required_createPublisher1MustProduceAStreamOfExactly1Element() throws Throwable {

    publisherVerification.required_createPublisher1MustProduceAStreamOfExactly1Element();

  }

  @Test

  public void required_createPublisher3MustProduceAStreamOfExactly3Elements() throws Throwable {

    publisherVerification.required_createPublisher3MustProduceAStreamOfExactly3Elements();

  }

  @Override @Test

  public void required_spec101_subscriptionRequestMustResultInTheCorrectNumberOfProducedElements() throws Throwable {

    publisherVerification.required_spec101_subscriptionRequestMustResultInTheCorrectNumberOfProducedElements();

  }

  @Override @Test

  public void required_spec102_maySignalLessThanRequestedAndTerminateSubscription() throws Throwable {

    publisherVerification.required_spec102_maySignalLessThanRequestedAndTerminateSubscription();

  }

  @Override @Test

  public void stochastic_spec103_mustSignalOnMethodsSequentially() throws Throwable {

    publisherVerification.stochastic_spec103_mustSignalOnMethodsSequentially();

  }

  @Override @Test

  public void optional_spec104_mustSignalOnErrorWhenFails() throws Throwable {

    publisherVerification.optional_spec104_mustSignalOnErrorWhenFails();

  }

  @Override @Test

  public void required_spec105_mustSignalOnCompleteWhenFiniteStreamTerminates() throws Throwable {

    publisherVerification.required_spec105_mustSignalOnCompleteWhenFiniteStreamTerminates();

  }

  @Override @Test

  public void optional_spec105_emptyStreamMustTerminateBySignallingOnComplete() throws Throwable {

    publisherVerification.optional_spec105_emptyStreamMustTerminateBySignallingOnComplete();

  }

  @Override @Test

  public void untested_spec106_mustConsiderSubscriptionCancelledAfterOnErrorOrOnCompleteHasBeenCalled() throws Throwable {

    publisherVerification.untested_spec106_mustConsiderSubscriptionCancelledAfterOnErrorOrOnCompleteHasBeenCalled();

  }

  @Override @Test

  public void required_spec107_mustNotEmitFurtherSignalsOnceOnCompleteHasBeenSignalled() throws Throwable {

    publisherVerification.required_spec107_mustNotEmitFurtherSignalsOnceOnCompleteHasBeenSignalled();

  }

  @Override @Test

  public void untested_spec107_mustNotEmitFurtherSignalsOnceOnErrorHasBeenSignalled() throws Throwable {

    publisherVerification.untested_spec107_mustNotEmitFurtherSignalsOnceOnErrorHasBeenSignalled();

  }

  @Override @Test

  public void untested_spec108_possiblyCanceledSubscriptionShouldNotReceiveOnErrorOrOnCompleteSignals() throws Throwable {

    publisherVerification.untested_spec108_possiblyCanceledSubscriptionShouldNotReceiveOnErrorOrOnCompleteSignals();

  }

  @Override @Test

  public void untested_spec109_subscribeShouldNotThrowNonFatalThrowable() throws Throwable {

    publisherVerification.untested_spec109_subscribeShouldNotThrowNonFatalThrowable();

  }

  @Override @Test

  public void required_spec109_subscribeThrowNPEOnNullSubscriber() throws Throwable {

    publisherVerification.required_spec109_subscribeThrowNPEOnNullSubscriber();

  }

  @Override @Test

  public void required_spec109_mayRejectCallsToSubscribeIfPublisherIsUnableOrUnwillingToServeThemRejectionMustTriggerOnErrorAfterOnSubscribe() throws Throwable {

    publisherVerification.required_spec109_mayRejectCallsToSubscribeIfPublisherIsUnableOrUnwillingToServeThemRejectionMustTriggerOnErrorAfterOnSubscribe();

  }

  @Override @Test

  public void required_spec109_mustIssueOnSubscribeForNonNullSubscriber() throws Throwable {

    publisherVerification.required_spec109_mustIssueOnSubscribeForNonNullSubscriber();

  }

  @Override @Test

  public void untested_spec110_rejectASubscriptionRequestIfTheSameSubscriberSubscribesTwice() throws Throwable {

    publisherVerification.untested_spec110_rejectASubscriptionRequestIfTheSameSubscriberSubscribesTwice();

  }

  @Override @Test

  public void optional_spec111_maySupportMultiSubscribe() throws Throwable {

    publisherVerification.optional_spec111_maySupportMultiSubscribe();

  }

  @Override @Test

  public void optional_spec111_registeredSubscribersMustReceiveOnNextOrOnCompleteSignals() throws Throwable {

    publisherVerification.optional_spec111_registeredSubscribersMustReceiveOnNextOrOnCompleteSignals();

  }

  @Override @Test

  public void optional_spec111_multicast_mustProduceTheSameElementsInTheSameSequenceToAllOfItsSubscribersWhenRequestingOneByOne() throws Throwable {

    publisherVerification.optional_spec111_multicast_mustProduceTheSameElementsInTheSameSequenceToAllOfItsSubscribersWhenRequestingOneByOne();

  }

  @Override @Test

  public void optional_spec111_multicast_mustProduceTheSameElementsInTheSameSequenceToAllOfItsSubscribersWhenRequestingManyUpfront() throws Throwable {

    publisherVerification.optional_spec111_multicast_mustProduceTheSameElementsInTheSameSequenceToAllOfItsSubscribersWhenRequestingManyUpfront();

  }

  @Override @Test

  public void optional_spec111_multicast_mustProduceTheSameElementsInTheSameSequenceToAllOfItsSubscribersWhenRequestingManyUpfrontAndCompleteAsExpected() throws Throwable {

    publisherVerification.optional_spec111_multicast_mustProduceTheSameElementsInTheSameSequenceToAllOfItsSubscribersWhenRequestingManyUpfrontAndCompleteAsExpected();

  }

  @Override @Test

  public void required_spec302_mustAllowSynchronousRequestCallsFromOnNextAndOnSubscribe() throws Throwable {

    publisherVerification.required_spec302_mustAllowSynchronousRequestCallsFromOnNextAndOnSubscribe();

  }

  @Override @Test

  public void required_spec303_mustNotAllowUnboundedRecursion() throws Throwable {

    publisherVerification.required_spec303_mustNotAllowUnboundedRecursion();

  }

  @Override @Test

  public void untested_spec304_requestShouldNotPerformHeavyComputations() throws Exception {

    publisherVerification.untested_spec304_requestShouldNotPerformHeavyComputations();

  }

  @Override @Test

  public void untested_spec305_cancelMustNotSynchronouslyPerformHeavyComputation() throws Exception {

    publisherVerification.untested_spec305_cancelMustNotSynchronouslyPerformHeavyComputation();

  }

  @Override @Test

  public void required_spec306_afterSubscriptionIsCancelledRequestMustBeNops() throws Throwable {

    publisherVerification.required_spec306_afterSubscriptionIsCancelledRequestMustBeNops();

  }

  @Override @Test

  public void required_spec307_afterSubscriptionIsCancelledAdditionalCancelationsMustBeNops() throws Throwable {

    publisherVerification.required_spec307_afterSubscriptionIsCancelledAdditionalCancelationsMustBeNops();

  }

  @Override @Test

  public void required_spec309_requestZeroMustSignalIllegalArgumentException() throws Throwable {

    publisherVerification.required_spec309_requestZeroMustSignalIllegalArgumentException();

  }

  @Override @Test

  public void required_spec309_requestNegativeNumberMustSignalIllegalArgumentException() throws Throwable {

    publisherVerification.required_spec309_requestNegativeNumberMustSignalIllegalArgumentException();

  }

  @Override @Test

  public void optional_spec309_requestNegativeNumberMaySignalIllegalArgumentExceptionWithSpecificMessage() throws Throwable {

    publisherVerification.optional_spec309_requestNegativeNumberMaySignalIllegalArgumentExceptionWithSpecificMessage();

  }

  @Override @Test

  public void required_spec312_cancelMustMakeThePublisherToEventuallyStopSignaling() throws Throwable {

    publisherVerification.required_spec312_cancelMustMakeThePublisherToEventuallyStopSignaling();

  }

  @Override @Test

  public void required_spec313_cancelMustMakeThePublisherEventuallyDropAllReferencesToTheSubscriber() throws Throwable {

    publisherVerification.required_spec313_cancelMustMakeThePublisherEventuallyDropAllReferencesToTheSubscriber();

  }

  @Override @Test

  public void required_spec317_mustSupportAPendingElementCountUpToLongMaxValue() throws Throwable {

    publisherVerification.required_spec317_mustSupportAPendingElementCountUpToLongMaxValue();

  }

  @Override @Test

  public void required_spec317_mustSupportACumulativePendingElementCountUpToLongMaxValue() throws Throwable {

    publisherVerification.required_spec317_mustSupportACumulativePendingElementCountUpToLongMaxValue();

  }

  @Override @Test

  public void required_spec317_mustNotSignalOnErrorWhenPendingAboveLongMaxValue() throws Throwable {

    publisherVerification.required_spec317_mustNotSignalOnErrorWhenPendingAboveLongMaxValue();

  }

  /**

   * Asks for a {@code Processor} that supports at least 2 {@code Subscriber}s at once and checks if two {@code Subscriber}s

   * receive the same items and a terminal {@code Exception}.

   * <p>

   * If the {@code Processor} requests and/or emits items only when all of its {@code Subscriber}s have requested,

   * override {@link #doesCoordinatedEmission()} and return {@code true} to indicate this property.

   * <p>

   * <b>Verifies rule:</b> <a href='https://github.com/reactive-streams/reactive-streams-jvm#1.4'>1.4</a> with multiple

   * {@code Subscriber}s.

   * <p>

   * The test is not executed if {@link IdentityProcessorVerification#maxSupportedSubscribers()} is less than 2.

   * <p>

   * If this test fails, the following could be checked within the {@code Processor} implementation:

   * <ul>

   * <li>The {@code TestEnvironment} has large enough timeout specified in case the {@code Processor} has some time-delay behavior.</li>

   * <li>The {@code Processor} is able to fulfill requests of its {@code Subscriber}s independently of each other's requests or

   * else override {@link #doesCoordinatedEmission()} and return {@code true} to indicate the test {@code Subscriber}s

   * both have to request first.</li>

   * </ul>

   */

  @Test

  public void required_spec104_mustCallOnErrorOnAllItsSubscribersIfItEncountersANonRecoverableError() throws Throwable {

    optionalMultipleSubscribersTest(2, new Function<Long,TestSetup>() {

      @Override

      public TestSetup apply(Long aLong) throws Throwable {

        return new TestSetup(env, processorBufferSize) {{

          final ManualSubscriberWithErrorCollection<T> sub1 = new ManualSubscriberWithErrorCollection<T>(env);

          env.subscribe(processor, sub1);

          final ManualSubscriberWithErrorCollection<T> sub2 = new ManualSubscriberWithErrorCollection<T>(env);

          env.subscribe(processor, sub2);

          final Exception ex = new RuntimeException("Test exception");

          if (doesCoordinatedEmission()) {

            sub1.request(1);

            sub2.request(1);

            expectRequest();

            final T x = sendNextTFromUpstream();

            expectNextElement(sub1, x);

            expectNextElement(sub2, x);

            sub1.request(1);

            sub2.request(1);

          } else {

            sub1.request(1);

            expectRequest(env.defaultTimeoutMillis(),

                    "If the Processor coordinates requests/emissions when having multiple Subscribers"

                    + " at once, please override doesCoordinatedEmission() to return true in this "

                    + "IdentityProcessorVerification to allow this test to pass.");

            final T x = sendNextTFromUpstream();

            expectNextElement(sub1, x,

                    "If the Processor coordinates requests/emissions when having multiple Subscribers"

                            + " at once, please override doesCoordinatedEmission() to return true in this "

                            + "IdentityProcessorVerification to allow this test to pass.");

            sub1.request(1);

            // sub1 has received one element, and has one demand pending

            // sub2 has not yet requested anything

          }

          sendError(ex);

          sub1.expectError(ex);

          sub2.expectError(ex);

          env.verifyNoAsyncErrorsNoDelay();

        }};

      }

    });

  }

  ////////////////////// SUBSCRIBER RULES VERIFICATION ///////////////////////////

  // Verifies rule: https://github.com/reactive-streams/reactive-streams-jvm#4.1

  // A Processor

  //   must obey all Subscriber rules on its consuming side

  public Subscriber<T> createSubscriber(final SubscriberWhiteboxVerification.WhiteboxSubscriberProbe<T> probe) {

    final Processor<T, T> processor = createIdentityProcessor(processorBufferSize);

    processor.subscribe(

        new Subscriber<T>() {

          private final Promise<Subscription> subs = new Promise<Subscription>(env);

          @Override

          public void onSubscribe(final Subscription subscription) {

            if (env.debugEnabled()) {

              env.debug(String.format("whiteboxSubscriber::onSubscribe(%s)", subscription));

            }

            if (subs.isCompleted()) subscription.cancel(); // the Probe must also pass subscriber verification

            probe.registerOnSubscribe(new SubscriberWhiteboxVerification.SubscriberPuppet() {

              @Override

              public void triggerRequest(long elements) {

                subscription.request(elements);

              }