/*
 * Copyright (c) 2011-2017 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
 *
 *       https://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.core.publisher;

import java.util.Objects;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
import java.util.function.Function;
import java.util.stream.Stream;

import org.reactivestreams.Publisher;
import org.reactivestreams.Subscription;

import reactor.core.CorePublisher;
import reactor.core.CoreSubscriber;
import reactor.core.Exceptions;
import reactor.core.Scannable;
import reactor.util.annotation.Nullable;
import reactor.util.context.Context;

Waits for a Mono source to terminate or produce a value, in which case the value is
mapped to a Publisher used as a delay: its termination will trigger the actual emission
of the value downstream. If the Mono source didn't produce a value, terminate with the
same signal (empty completion or error).
Author:
Simon Baslé
Type parameters:
<T> – the value type/**
 * Waits for a Mono source to terminate or produce a value, in which case the value is
 * mapped to a Publisher used as a delay: its termination will trigger the actual emission
 * of the value downstream. If the Mono source didn't produce a value, terminate with the
 * same signal (empty completion or error).
 *
 * @param <T> the value type
 *
 * @author Simon Baslé
 */
final class MonoDelayUntil<T> extends Mono<T> implements Scannable,
                                                         OptimizableOperator<T, T> {

	final Mono<T> source;

	Function<? super T, ? extends Publisher<?>>[] otherGenerators;

	@Nullable
	final OptimizableOperator<?, T> optimizableOperator;

	@SuppressWarnings("unchecked")
	MonoDelayUntil(Mono<T> monoSource,
			Function<? super T, ? extends Publisher<?>> triggerGenerator) {
		this.source = Objects.requireNonNull(monoSource, "monoSource");
		this.otherGenerators = new Function[] { Objects.requireNonNull(triggerGenerator, "triggerGenerator")};
		this.optimizableOperator = source instanceof OptimizableOperator ? (OptimizableOperator) source : null;
	}

	MonoDelayUntil(Mono<T> monoSource,
			Function<? super T, ? extends Publisher<?>>[] triggerGenerators) {
		this.source = Objects.requireNonNull(monoSource, "monoSource");
		this.otherGenerators = triggerGenerators;
		if (source instanceof OptimizableOperator) {
			@SuppressWarnings("unchecked")
			OptimizableOperator<?, T> optimSource = (OptimizableOperator<?, T>) source;
			this.optimizableOperator = optimSource;
		}
		else {
			this.optimizableOperator = null;
		}
	}

	
Add a trigger generator to wait for.
Params:
delayError – the delayError parameter for the trigger being added, ignored if already true
triggerGenerator – the new trigger to add to the copy of the operator
Returns:
a new MonoDelayUntil instance with same source but additional trigger generator/**
	 * Add a trigger generator to wait for.
	 * @param delayError the delayError parameter for the trigger being added, ignored if already true
	 * @param triggerGenerator the new trigger to add to the copy of the operator
	 * @return a new {@link MonoDelayUntil} instance with same source but additional trigger generator
	 */
	@SuppressWarnings("unchecked")
	MonoDelayUntil<T> copyWithNewTriggerGenerator(boolean delayError,
			Function<? super T, ? extends Publisher<?>> triggerGenerator) {
		Objects.requireNonNull(triggerGenerator, "triggerGenerator");
		Function<? super T, ? extends Publisher<?>>[] oldTriggers = this.otherGenerators;
		Function<? super T, ? extends Publisher<?>>[] newTriggers = new Function[oldTriggers.length + 1];
		System.arraycopy(oldTriggers, 0, newTriggers, 0, oldTriggers.length);
		newTriggers[oldTriggers.length] = triggerGenerator;
		return new MonoDelayUntil<>(this.source, newTriggers);
	}

	@Override
	public void subscribe(CoreSubscriber<? super T> actual) {
		try {
			source.subscribe(subscribeOrReturn(actual));
		}
		catch (Throwable e) {
			Operators.error(actual, Operators.onOperatorError(e, actual.currentContext()));
			return;
		}
	}

	@Override
	public final CoreSubscriber<? super T> subscribeOrReturn(CoreSubscriber<? super T> actual) throws Throwable {
		DelayUntilCoordinator<T> parent = new DelayUntilCoordinator<>(actual, otherGenerators);
		actual.onSubscribe(parent);

		return parent;
	}

	@Override
	public final CorePublisher<? extends T> source() {
		return source;
	}

	@Override
	public final OptimizableOperator<?, ? extends T> nextOptimizableSource() {
		return optimizableOperator;
	}

	@Override
	public Object scanUnsafe(Attr key) {
		if (key == Attr.RUN_STYLE) return Attr.RunStyle.SYNC;
		return null; //no particular key to be represented, still useful in hooks
	}

	static final class DelayUntilCoordinator<T>
			extends Operators.MonoSubscriber<T, T> {

		static final DelayUntilTrigger[] NO_TRIGGER = new DelayUntilTrigger[0];

		final int                                                n;
		final Function<? super T, ? extends Publisher<?>>[] otherGenerators;

		volatile int done;
		static final AtomicIntegerFieldUpdater<DelayUntilCoordinator> DONE =
				AtomicIntegerFieldUpdater.newUpdater(DelayUntilCoordinator.class, "done");

		volatile Subscription s;
		@SuppressWarnings("rawtypes")
		static final AtomicReferenceFieldUpdater<DelayUntilCoordinator, Subscription> S =
				AtomicReferenceFieldUpdater.newUpdater(DelayUntilCoordinator.class, Subscription.class, "s");

		DelayUntilTrigger[] triggerSubscribers;

		DelayUntilCoordinator(CoreSubscriber<? super T> subscriber,
				Function<? super T, ? extends Publisher<?>>[] otherGenerators) {
			super(subscriber);
			this.otherGenerators = otherGenerators;
			//don't consider the source as this is only used from when there is a value
			this.n = otherGenerators.length;
			triggerSubscribers = NO_TRIGGER;
		}

		@Override
		public void onSubscribe(Subscription s) {
			if (Operators.setOnce(S, this, s)) {
				s.request(Long.MAX_VALUE);
			} else {
				s.cancel();
			}
		}

		@Override
		public void onNext(T t) {
			if (this.value == null) {
				setValue(t);
				subscribeNextTrigger(t, done);
			}
		}

		@Override
		public void onError(Throwable t) {
			actual.onError(t);
		}

		@Override
		public void onComplete() {
			if (this.value == null && state < HAS_REQUEST_HAS_VALUE) {
				actual.onComplete();
			}
		}

		@Override
		@Nullable
		public Object scanUnsafe(Attr key) {
			if (key == Attr.TERMINATED) return done == n;
			if (key == Attr.RUN_STYLE) return Attr.RunStyle.SYNC;

			return super.scanUnsafe(key);
		}

		@Override
		public Stream<? extends Scannable> inners() {
			return Stream.of(triggerSubscribers);
		}


		@SuppressWarnings("unchecked")
		void subscribeNextTrigger(@Nullable T value, int triggerIndex) {
			if (value == null) {
				return; //we've been cancelled
			}
			if (triggerSubscribers == NO_TRIGGER) {
				triggerSubscribers = new DelayUntilTrigger[otherGenerators.length];
			}

			Function<? super T, ? extends Publisher<?>> generator = otherGenerators[triggerIndex];

			Publisher<?> p;

			try {
				p = generator.apply(value);
				Objects.requireNonNull(p, "mapper returned null value");
			}
			catch (Throwable t) {
				onError(t);
				return;
			}

			DelayUntilTrigger triggerSubscriber = new DelayUntilTrigger<>(this);

			this.triggerSubscribers[triggerIndex] = triggerSubscriber;
			p.subscribe(triggerSubscriber);
		}

		void signal() {
			int nextIndex = DONE.incrementAndGet(this);
			if (nextIndex != n) {
				subscribeNextTrigger(this.value, nextIndex);
				return;
			}

			Throwable error = null;
			Throwable compositeError = null;

			//check for errors in the triggers
			for (int i = 0; i < n; i++) {
				DelayUntilTrigger mt = triggerSubscribers[i];
				Throwable e = mt.error;
				if (e != null) {
					if (compositeError != null) {
						//this is ok as the composite created by multiple is never a singleton
						compositeError.addSuppressed(e);
					} else
					if (error != null) {
						compositeError = Exceptions.multiple(error, e);
					} else {
						error = e;
					}
					//else the trigger publisher was empty, but we'll ignore that
				}
			}

			if (compositeError != null) {
				actual.onError(compositeError);
			}
			else if (error != null) {
				actual.onError(error);
			} else {
				//emit the value downstream
				complete(this.value);
			}
		}

		@Override
		public void cancel() {
			if (!isCancelled()) {
				super.cancel();
				//source is always cancellable...
				Operators.terminate(S, this);
				//...but triggerSubscribers could be partially initialized
				for (int i = 0; i < triggerSubscribers.length; i++) {
					DelayUntilTrigger ts = triggerSubscribers[i];
					if (ts != null) ts.cancel();
				}
			}
		}
	}

	static final class DelayUntilTrigger<T> implements InnerConsumer<T> {

		final DelayUntilCoordinator<?> parent;

		volatile Subscription s;
		@SuppressWarnings("rawtypes")
		static final AtomicReferenceFieldUpdater<DelayUntilTrigger, Subscription> S =
				AtomicReferenceFieldUpdater.newUpdater(DelayUntilTrigger.class, Subscription.class, "s");

		boolean done;
		volatile Throwable error;

		DelayUntilTrigger(DelayUntilCoordinator<?> parent) {
			this.parent = parent;
		}

		@Override
		public Context currentContext() {
			return parent.currentContext();
		}

		@Override
		@Nullable
		public Object scanUnsafe(Attr key) {
			if (key == Attr.CANCELLED) return s == Operators.cancelledSubscription();
			if (key == Attr.PARENT) return s;
			if (key == Attr.ACTUAL) return parent;
			if (key == Attr.ERROR) return error;
			if (key == Attr.PREFETCH) return Integer.MAX_VALUE;
			if (key == Attr.RUN_STYLE) return Attr.RunStyle.SYNC;

			return null;
		}

		@Override
		public void onSubscribe(Subscription s) {
			if (Operators.setOnce(S, this, s)) {
				s.request(Long.MAX_VALUE);
			} else {
				s.cancel();
			}
		}

		@Override
		public void onNext(Object t) {
			//NO-OP
		}

		@Override
		public void onError(Throwable t) {
			error = t;
			if (DelayUntilCoordinator.DONE.getAndSet(parent, parent.n) != parent.n) {
				parent.cancel();
				parent.actual.onError(t);
			}
		}

		@Override
		public void onComplete() {
			if (!done) {
				done = true;
				parent.signal();
			}
		}

		void cancel() {
			Operators.terminate(S, this);
		}
	}
}