<commit_before>/*
 * Copyright (c) 2011-2016 Pivotal Software Inc, 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.
 * You may obtain a copy of the License at
 *
 *       http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License 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 reactor.test;

import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.atomic.AtomicLongFieldUpdater;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
import java.util.function.BiFunction;
import java.util.function.Consumer;
import java.util.function.Predicate;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import reactor.core.Exceptions;
import reactor.core.Fuseable;
import reactor.core.Receiver;
import reactor.core.Trackable;
import reactor.core.publisher.Operators;
import reactor.core.publisher.Signal;
import reactor.test.scheduler.VirtualTimeScheduler;
import reactor.util.Logger;
import reactor.util.Loggers;

/**
 * Default implementation of {@link StepVerifier.Step} and
 * {@link StepVerifier.LastStep}.
 *
 * @author <NAME>
 * @since 1.0
 */
final class DefaultStepVerifierBuilder<T>
		implements StepVerifier.FirstStep<T> {

	static void checkPositive(long n) {
		if (n < 0) {
			throw new IllegalArgumentException("'n' should be >= 0 but was " + n);
		}
	}

	static void checkStrictlyPositive(long n) {
		if (n <= 0) {
			throw new IllegalArgumentException("'n' should be > 0 but was " + n);
		}
	}

	static <T> StepVerifier.FirstStep<T> newVerifier(long n,
			Supplier<? extends Publisher<? extends T>> scenarioSupplier,
			Supplier<? extends VirtualTimeScheduler> vtsLookup){
		DefaultStepVerifierBuilder.checkPositive(n);
		Objects.requireNonNull(scenarioSupplier, "scenarioSupplier");

		return new DefaultStepVerifierBuilder<>
				(n, scenarioSupplier, vtsLookup);
	}

	@SuppressWarnings("unchecked")
	static <T> SignalEvent<T> defaultFirstStep() {
		return (SignalEvent<T>) DEFAULT_ONSUBSCRIBE_STEP;
	}

	final List<Event<T>>                             script;
	final long                                       initialRequest;
	final Supplier<? extends VirtualTimeScheduler>   vtsLookup;
	final Supplier<? extends Publisher<? extends T>> sourceSupplier;

	long hangCheckRequested;
	int  requestedFusionMode = -1;
	int  expectedFusionMode  = -1;

	DefaultStepVerifierBuilder(long initialRequest,
			Supplier<? extends Publisher<? extends T>> sourceSupplier,
			Supplier<? extends VirtualTimeScheduler> vtsLookup) {
		this.initialRequest = initialRequest;
		this.vtsLookup = vtsLookup;
		this.sourceSupplier = sourceSupplier;
		this.script = new ArrayList<>();
		this.script.add(defaultFirstStep());

		this.hangCheckRequested = initialRequest;
	}

	@Override
	public DefaultStepVerifierBuilder<T> as(String description) {
		this.script.add(new DescriptionEvent<>(description));
		return this;
	}

	@Override
	public DefaultStepVerifier<T> consumeErrorWith(Consumer<Throwable> consumer) {
		Objects.requireNonNull(consumer, "consumer");
		SignalEvent<T> event = new SignalEvent<>((signal, se) -> {
			if (!signal.isOnError()) {
				return fail(se, "expected: onError(); actual: %s", signal);
			}
			else {
				consumer.accept(signal.getThrowable());
				return Optional.empty();
			}
		}, "consumeErrorWith");
		this.script.add(event);
		return build();
	}

	@Override
	public DefaultStepVerifierBuilder<T> assertNext(Consumer<? super T> consumer) {
		return consumeNextWith(consumer, "assertNext");
	}

	@Override
	public DefaultStepVerifierBuilder<T> consumeNextWith(
			Consumer<? super T> consumer) {
		return consumeNextWith(consumer, "consumeNextWith");
	}

	private DefaultStepVerifierBuilder<T> consumeNextWith(Consumer<? super T> consumer, String description) {
		Objects.requireNonNull(consumer, "consumer");
		checkPotentialHang(1, description);
		SignalEvent<T> event = new SignalEvent<>((signal, se) -> {
			if (!signal.isOnNext()) {
				return fail(se, "expected: onNext(); actual: %s", signal);
			}
			else {
				consumer.accept(signal.get());
				return Optional.empty();
			}
		}, description);
		this.script.add(event);
		return this;
	}

	@Override
	public DefaultStepVerifierBuilder<T> consumeRecordedWith(
			Consumer<? super Collection<T>> consumer) {
		Objects.requireNonNull(consumer, "consumer");
		this.script.add(new CollectEvent<>(consumer, "consumeRecordedWith"));
		return this;
	}

	@Override
	public DefaultStepVerifierBuilder<T> consumeSubscriptionWith(
			Consumer<? super Subscription> consumer) {
		Objects.requireNonNull(consumer, "consumer");
		if(script.isEmpty()) {
			this.script.set(0, new SignalEvent<>((signal, se) -> {
				if (!signal.isOnSubscribe()) {
					return fail(se, "expected: onSubscribe(); actual: %s", signal);
				}
				else {
					consumer.accept(signal.getSubscription());
					return Optional.empty();
				}
			}, "consumeSubscriptionWith"));
		}
		else {
			this.script.add(new SubscriptionConsumerEvent<>(consumer,
					"consumeSubscriptionWith"));
		}
		return this;
	}

	@Override<commit_msg>retain onSubscribe signal check on first consumeSubscriptionWith<commit_after>/*
 * Copyright (c) 2011-2016 Pivotal Software Inc, 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.
 * You may obtain a copy of the License at
 *
 *       http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License 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 reactor.test;

import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.atomic.AtomicLongFieldUpdater;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
import java.util.function.BiFunction;
import java.util.function.Consumer;
import java.util.function.Predicate;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import reactor.core.Exceptions;
import reactor.core.Fuseable;
import reactor.core.Receiver;
import reactor.core.Trackable;
import reactor.core.publisher.Operators;
import reactor.core.publisher.Signal;
import reactor.test.scheduler.VirtualTimeScheduler;
import reactor.util.Logger;
import reactor.util.Loggers;

/**
 * Default implementation of {@link StepVerifier.Step} and
 * {@link StepVerifier.LastStep}.
 *
 * @author <NAME>
 * @since 1.0
 */
final class DefaultStepVerifierBuilder<T>
		implements StepVerifier.FirstStep<T> {

	static void checkPositive(long n) {
		if (n < 0) {
			throw new IllegalArgumentException("'n' should be >= 0 but was " + n);
		}
	}

	static void checkStrictlyPositive(long n) {
		if (n <= 0) {
			throw new IllegalArgumentException("'n' should be > 0 but was " + n);
		}
	}

	static <T> StepVerifier.FirstStep<T> newVerifier(long n,
			Supplier<? extends Publisher<? extends T>> scenarioSupplier,
			Supplier<? extends VirtualTimeScheduler> vtsLookup){
		DefaultStepVerifierBuilder.checkPositive(n);
		Objects.requireNonNull(scenarioSupplier, "scenarioSupplier");

		return new DefaultStepVerifierBuilder<>
				(n, scenarioSupplier, vtsLookup);
	}

	@SuppressWarnings("unchecked")
	static <T> SignalEvent<T> defaultFirstStep() {
		return (SignalEvent<T>) DEFAULT_ONSUBSCRIBE_STEP;
	}

	final List<Event<T>>                             script;
	final long                                       initialRequest;
	final Supplier<? extends VirtualTimeScheduler>   vtsLookup;
	final Supplier<? extends Publisher<? extends T>> sourceSupplier;

	long hangCheckRequested;
	int  requestedFusionMode = -1;
	int  expectedFusionMode  = -1;

	DefaultStepVerifierBuilder(long initialRequest,
			Supplier<? extends Publisher<? extends T>> sourceSupplier,
			Supplier<? extends VirtualTimeScheduler> vtsLookup) {
		this.initialRequest = initialRequest;
		this.vtsLookup = vtsLookup;
		this.sourceSupplier = sourceSupplier;
		this.script = new ArrayList<>();
		this.script.add(defaultFirstStep());

		this.hangCheckRequested = initialRequest;
	}

	@Override
	public DefaultStepVerifierBuilder<T> as(String description) {
		this.script.add(new DescriptionEvent<>(description));
		return this;
	}

	@Override
	public DefaultStepVerifier<T> consumeErrorWith(Consumer<Throwable> consumer) {
		Objects.requireNonNull(consumer, "consumer");
		SignalEvent<T> event = new SignalEvent<>((signal, se) -> {
			if (!signal.isOnError()) {
				return fail(se, "expected: onError(); actual: %s", signal);
			}
			else {
				consumer.accept(signal.getThrowable());
				return Optional.empty();
			}
		}, "consumeErrorWith");
		this.script.add(event);
		return build();
	}

	@Override
	public DefaultStepVerifierBuilder<T> assertNext(Consumer<? super T> consumer) {
		return consumeNextWith(consumer, "assertNext");
	}

	@Override
	public DefaultStepVerifierBuilder<T> consumeNextWith(
			Consumer<? super T> consumer) {
		return consumeNextWith(consumer, "consumeNextWith");
	}

	private DefaultStepVerifierBuilder<T> consumeNextWith(Consumer<? super T> consumer, String description) {
		Objects.requireNonNull(consumer, "consumer");
		checkPotentialHang(1, description);
		SignalEvent<T> event = new SignalEvent<>((signal, se) -> {
			if (!signal.isOnNext()) {
				return fail(se, "expected: onNext(); actual: %s", signal);
			}
			else {
				consumer.accept(signal.get());
				return Optional.empty();
			}
		}, description);
		this.script.add(event);
		return this;
	}

	@Override
	public DefaultStepVerifierBuilder<T> consumeRecordedWith(
			Consumer<? super Collection<T>> consumer) {
		Objects.requireNonNull(consumer, "consumer");
		this.script.add(new CollectEvent<>(consumer, "consumeRecordedWith"));
		return this;
	}

	@Override
	public DefaultStepVerifierBuilder<T> consumeSubscriptionWith(
			Consumer<? super Subscription> consumer) {
		Objects.requireNonNull(consumer, "consumer");
		if(script.isEmpty() || (script.size() == 1 && script.get(0) == DEFAULT_ONSUBSCRIBE_STEP)) {
			this.script.set(0, new SignalEvent<>((signal, se) -> {
				if (!signal.isOnSubscribe()) {
					return fail(se, "expected: onSubscribe(); actual: %s", signal);
				}
				else {
					consumer.accept(signal.getSubscription());
					return Optional.empty();
				}
			}, "consumeSubscriptionWith"));
		}
		else {
			this.script.add(new SubscriptionConsumerEvent<>(consumer,
					"consumeSubscriptionWith"));
		}
		return this;
	}

	@Override
{'fraction_non_alphanumeric': 0.1154099440237076, 'fraction_numerical': 0.0037866315442871253, 'mean_word_length': 4.782484531175631, 'pattern_counts': {'":': 0, '<': 93, '<?xml version=': 0, '>': 99, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1566728', 'n_tokens_mistral': 3951, 'n_tokens_neox': 3589, 'n_words': 879}