/*
 * Copyright (c) 2011-2018 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.time.Duration;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
import java.util.function.Function;
import java.util.function.Supplier;

import org.reactivestreams.Subscription;

import reactor.core.CoreSubscriber;
import reactor.core.Exceptions;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;
import reactor.util.Logger;
import reactor.util.Loggers;
import reactor.util.annotation.Nullable;
import reactor.util.context.Context;
import reactor.util.context.ContextView;

An operator that caches the value from a source Mono with a TTL, after which the value
expires and the next subscription will trigger a new source subscription.
Author:
Simon Baslé/**
 * An operator that caches the value from a source Mono with a TTL, after which the value
 * expires and the next subscription will trigger a new source subscription.
 *
 * @author Simon Baslé
 */
class MonoCacheTime<T> extends InternalMonoOperator<T, T> implements Runnable {

	private static final Duration DURATION_INFINITE = Duration.ofMillis(Long.MAX_VALUE);

	private static final Logger LOGGER = Loggers.getLogger(MonoCacheTime.class);

	final Function<? super Signal<T>, Duration> ttlGenerator;
	final Scheduler                             clock;

	volatile Signal<T> state;
	static final AtomicReferenceFieldUpdater<MonoCacheTime, Signal> STATE =
			AtomicReferenceFieldUpdater.newUpdater(MonoCacheTime.class, Signal.class, "state");

	static final Signal<?> EMPTY = new ImmutableSignal<>(Context.empty(), SignalType.ON_NEXT, null, null, null);

	MonoCacheTime(Mono<? extends T> source, Duration ttl, Scheduler clock) {
		super(source);
		this.ttlGenerator = ignoredSignal -> ttl;
		this.clock = clock;
		@SuppressWarnings("unchecked")
		Signal<T> state = (Signal<T>) EMPTY;
		this.state = state;
	}

	MonoCacheTime(Mono<? extends T> source) {
		this(source, sig -> DURATION_INFINITE, Schedulers.immediate());
	}

	MonoCacheTime(Mono<? extends T> source, Function<? super Signal<T>, Duration> ttlGenerator,
			Scheduler clock) {
		super(source);
		this.ttlGenerator = ttlGenerator;
		this.clock = clock;
		@SuppressWarnings("unchecked")
		Signal<T> state = (Signal<T>) EMPTY;
		this.state = state;
	}

	MonoCacheTime(Mono<? extends T> source,
			Function<? super T, Duration> valueTtlGenerator,
			Function<Throwable, Duration> errorTtlGenerator,
			Supplier<Duration> emptyTtlGenerator,
			Scheduler clock) {
		super(source);
		this.ttlGenerator = sig -> {
			if (sig.isOnNext()) return valueTtlGenerator.apply(sig.get());
			if (sig.isOnError()) return errorTtlGenerator.apply(sig.getThrowable());
			return emptyTtlGenerator.get();
		};
		this.clock = clock;
		@SuppressWarnings("unchecked")
		Signal<T> emptyState = (Signal<T>) EMPTY;
		this.state = emptyState;
	}

	public void run() {
		LOGGER.debug("expired {}", state);
		@SuppressWarnings("unchecked")
		Signal<T> emptyState = (Signal<T>) EMPTY;
		state = emptyState;
	}

	@Override
	public CoreSubscriber<? super T> subscribeOrReturn(CoreSubscriber<? super T> actual) {
		CacheMonoSubscriber<T> inner = new CacheMonoSubscriber<>(actual);
		actual.onSubscribe(inner);
		for(;;){
			Signal<T> state = this.state;
			if (state == EMPTY || state instanceof CoordinatorSubscriber) {
				boolean subscribe = false;
				CoordinatorSubscriber<T> coordinator;
				if (state == EMPTY) {
					coordinator = new CoordinatorSubscriber<>(this);
					if (!STATE.compareAndSet(this, EMPTY, coordinator)) {
						continue;
					}
					subscribe = true;
				}
				else {
					coordinator = (CoordinatorSubscriber<T>) state;
				}

				if (coordinator.add(inner)) {
					if (inner.isCancelled()) {
						coordinator.remove(inner);
					}
					else {
						inner.coordinator = coordinator;
					}

					if (subscribe) {
						source.subscribe(coordinator);
					}
					break;
				}
			}
			else {
				//state is an actual signal, cached
				if (state.isOnNext()) {
					inner.complete(state.get());
				}
				else if (state.isOnComplete()) {
					inner.onComplete();
				}
				else {
					inner.onError(state.getThrowable());
				}
				break;
			}
		}
		return null;
	}

	@Override
	public Object scanUnsafe(Attr key) {
		if (key == Attr.RUN_STYLE) return Attr.RunStyle.SYNC;
		return super.scanUnsafe(key);
	}

	static final class CoordinatorSubscriber<T> implements InnerConsumer<T>, Signal<T> {

		final MonoCacheTime<T> main;

		volatile Subscription subscription;
		static final AtomicReferenceFieldUpdater<CoordinatorSubscriber, Subscription> S =
				AtomicReferenceFieldUpdater.newUpdater(CoordinatorSubscriber.class, Subscription.class, "subscription");

		volatile Operators.MonoSubscriber<T, T>[] subscribers;
		static final AtomicReferenceFieldUpdater<CoordinatorSubscriber, Operators.MonoSubscriber[]> SUBSCRIBERS =
				AtomicReferenceFieldUpdater.newUpdater(CoordinatorSubscriber.class, Operators.MonoSubscriber[].class, "subscribers");

		@SuppressWarnings("unchecked")
		CoordinatorSubscriber(MonoCacheTime<T> main) {
			this.main = main;
			this.subscribers = EMPTY;
		}

		
unused in this context as the Signal interface is only implemented for use in the main's STATE compareAndSet. /**
		 * unused in this context as the {@link Signal} interface is only
		 * implemented for use in the main's STATE compareAndSet.
		 */
		@Override
		public Throwable getThrowable() {
			throw new UnsupportedOperationException("illegal signal use");
		}

		
unused in this context as the Signal interface is only implemented for use in the main's STATE compareAndSet. /**
		 * unused in this context as the {@link Signal} interface is only
		 * implemented for use in the main's STATE compareAndSet.
		 */
		@Override
		public Subscription getSubscription() {
			throw new UnsupportedOperationException("illegal signal use");
		}

		
unused in this context as the Signal interface is only implemented for use in the main's STATE compareAndSet. /**
		 * unused in this context as the {@link Signal} interface is only
		 * implemented for use in the main's STATE compareAndSet.
		 */
		@Override
		public T get() {
			throw new UnsupportedOperationException("illegal signal use");
		}

		
unused in this context as the Signal interface is only implemented for use in the main's STATE compareAndSet. /**
		 * unused in this context as the {@link Signal} interface is only
		 * implemented for use in the main's STATE compareAndSet.
		 */
		@Override
		public SignalType getType() {
			throw new UnsupportedOperationException("illegal signal use");
		}

		
unused in this context as the Signal interface is only implemented for use in the main's STATE compareAndSet. /**
		 * unused in this context as the {@link Signal} interface is only
		 * implemented for use in the main's STATE compareAndSet.
		 */
		@Override
		public ContextView getContextView() {
			throw new UnsupportedOperationException("illegal signal use: getContextView");
		}

		final boolean add(Operators.MonoSubscriber<T, T> toAdd) {
			for (; ; ) {
				Operators.MonoSubscriber<T, T>[] a = subscribers;
				if (a == TERMINATED) {
					return false;
				}
				int n = a.length;
				@SuppressWarnings("unchecked")
				Operators.MonoSubscriber<T, T>[] b = new Operators.MonoSubscriber[n + 1];
				System.arraycopy(a, 0, b, 0, n);
				b[n] = toAdd;
				if (SUBSCRIBERS.compareAndSet(this, a, b)) {
					return true;
				}
			}
		}

		final void remove(Operators.MonoSubscriber<T, T> toRemove) {
			for (; ; ) {
				Operators.MonoSubscriber<T, T>[] a = subscribers;
				if (a == TERMINATED || a == EMPTY) {
					return;
				}
				int n = a.length;
				int j = -1;
				for (int i = 0; i < n; i++) {
					if (a[i] == toRemove) {
						j = i;
						break;
					}
				}

				if (j < 0) {
					return;
				}

				Operators.MonoSubscriber<?, ?>[] b;
				if (n == 1) {
					b = EMPTY;
				}
				else {
					b = new Operators.MonoSubscriber<?, ?>[n - 1];
					System.arraycopy(a, 0, b, 0, j);
					System.arraycopy(a, j + 1, b, j, n - j - 1);
				}
				if (SUBSCRIBERS.compareAndSet(this, a, b)) {
					//no particular cleanup here for the EMPTY case, we don't cancel the
					// source because a new subscriber could come in before the coordinator
					// is terminated.

					return;
				}
			}
		}

		@Override
		public void onSubscribe(Subscription s) {
			if (Operators.validate(subscription, s)) {
				subscription = s;
				s.request(Long.MAX_VALUE);
			}
		}

		@SuppressWarnings("unchecked")
		private void signalCached(Signal<T> signal) {
			Signal<T> signalToPropagate = signal;
			if (STATE.compareAndSet(main, this, signal)) {
				Duration ttl = null;
				try {
					ttl = main.ttlGenerator.apply(signal);
				}
				catch (Throwable generatorError) {
					signalToPropagate = Signal.error(generatorError);
					STATE.set(main, signalToPropagate);
					if (signal.isOnError()) {
						//noinspection ThrowableNotThrown
						Exceptions.addSuppressed(generatorError, signal.getThrowable());
					}
				}

				if (ttl != null) {
					if (ttl.isZero()) {
						//immediate cache clear
						main.run();
					}
					else if (!ttl.equals(DURATION_INFINITE)) {
						main.clock.schedule(main, ttl.toNanos(), TimeUnit.NANOSECONDS);
					}
					//else TTL is Long.MAX_VALUE, schedule nothing but still cache
				}
				else {
					//error during TTL generation, signal != updatedSignal, aka dropped
					if (signal.isOnNext()) {
						Operators.onNextDropped(signal.get(), currentContext());
					}
					//if signal.isOnError(), avoid dropping the error. it is not really dropped but already suppressed
					//in all cases, unless nextDropped hook throws, immediate cache clear
					main.run();
				}
			}

			for (Operators.MonoSubscriber<T, T> inner : SUBSCRIBERS.getAndSet(this, TERMINATED)) {
				if (signalToPropagate.isOnNext()) {
					inner.complete(signalToPropagate.get());
				}
				else if (signalToPropagate.isOnError()) {
					inner.onError(signalToPropagate.getThrowable());
				}
				else {
					inner.onComplete();
				}
			}
		}

		@Override
		public void onNext(T t) {
			if (main.state != this) {
				Operators.onNextDroppedMulticast(t, subscribers);
				return;
			}
			signalCached(Signal.next(t));
		}

		@Override
		public void onError(Throwable t) {
			if (main.state != this) {
				Operators.onErrorDroppedMulticast(t, subscribers);
				return;
			}
			signalCached(Signal.error(t));
		}

		@Override
		public void onComplete() {
			signalCached(Signal.complete());
		}

		@Override
		public Context currentContext() {
			return Operators.multiSubscribersContext(subscribers);
		}

		@Nullable
		@Override
		public Object scanUnsafe(Attr key) {
			if (key == Attr.RUN_STYLE) return Attr.RunStyle.SYNC;
			return null;
		}

		private static final Operators.MonoSubscriber[] TERMINATED        = new Operators.MonoSubscriber[0];
		private static final Operators.MonoSubscriber[] EMPTY             = new Operators.MonoSubscriber[0];
	}

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

		CoordinatorSubscriber<T> coordinator;

		CacheMonoSubscriber(CoreSubscriber<? super T> actual) {
			super(actual);
		}

		@Override
		public void cancel() {
			super.cancel();
			CoordinatorSubscriber<T> coordinator = this.coordinator;
			if (coordinator != null) {
				coordinator.remove(this);
			}
		}

		@Override
		public Object scanUnsafe(Attr key) {
			if (key == Attr.RUN_STYLE) return Attr.RunStyle.SYNC;
			return super.scanUnsafe(key);
		}
	}

}