/*
 * 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
 *
 *       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.core.publisher;

import java.util.Objects;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
import java.util.function.BiFunction;

import org.reactivestreams.Subscription;
import reactor.core.CoreSubscriber;
import reactor.core.Fuseable;
import reactor.core.Scannable;
import reactor.util.annotation.Nullable;
import reactor.util.context.Context;

Reduces all 'rails' into a single value which then gets reduced into a single
Publisher sequence.
Type parameters:
<T> – the value type/**
 * Reduces all 'rails' into a single value which then gets reduced into a single
 * Publisher sequence.
 *
 * @param <T> the value type
 */
final class ParallelMergeReduce<T> extends Mono<T> implements Scannable, Fuseable {

	final ParallelFlux<? extends T> source;

	final BiFunction<T, T, T> reducer;

	ParallelMergeReduce(ParallelFlux<? extends T> source,
			BiFunction<T, T, T> reducer) {
		this.source = source;
		this.reducer = reducer;
	}

	@Override
	@Nullable
	public Object scanUnsafe(Attr key) {
		if (key == Attr.PARENT) return source;

		return null;
	}

	@Override
	public void subscribe(CoreSubscriber<? super T> actual) {
		MergeReduceMain<T> parent =
				new MergeReduceMain<>(actual, source.parallelism(), reducer);
		actual.onSubscribe(parent);

		source.subscribe(parent.subscribers);
	}

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

		final MergeReduceInner<T>[] subscribers;

		final BiFunction<T, T, T> reducer;

		volatile SlotPair<T> current;
		@SuppressWarnings("rawtypes")
		static final AtomicReferenceFieldUpdater<MergeReduceMain, SlotPair>
				CURRENT = AtomicReferenceFieldUpdater.newUpdater(
				MergeReduceMain.class,
				SlotPair.class,
				"current");

		volatile int remaining;
		@SuppressWarnings("rawtypes")
		static final AtomicIntegerFieldUpdater<MergeReduceMain>
				REMAINING = AtomicIntegerFieldUpdater.newUpdater(
				MergeReduceMain.class,
				"remaining");

		volatile Throwable error;
		@SuppressWarnings("rawtypes")
		static final AtomicReferenceFieldUpdater<MergeReduceMain, Throwable>
				ERROR = AtomicReferenceFieldUpdater.newUpdater(
				MergeReduceMain.class,
				Throwable.class,
				"error");

		MergeReduceMain(CoreSubscriber<? super T> subscriber,
				int n,
				BiFunction<T, T, T> reducer) {
			super(subscriber);
			@SuppressWarnings("unchecked") MergeReduceInner<T>[] a =
					new MergeReduceInner[n];
			for (int i = 0; i < n; i++) {
				a[i] = new MergeReduceInner<>(this, reducer);
			}
			this.subscribers = a;
			this.reducer = reducer;
			REMAINING.lazySet(this, n);
		}

		@Override
		@Nullable
		public Object scanUnsafe(Attr key) {
			if (key == Attr.ERROR) return error;

			return super.scanUnsafe(key);
		}

		@Nullable
		SlotPair<T> addValue(T value) {
			for (; ; ) {
				SlotPair<T> curr = current;

				if (curr == null) {
					curr = new SlotPair<>();
					if (!CURRENT.compareAndSet(this, null, curr)) {
						continue;
					}
				}

				int c = curr.tryAcquireSlot();
				if (c < 0) {
					CURRENT.compareAndSet(this, curr, null);
					continue;
				}
				if (c == 0) {
					curr.first = value;
				}
				else {
					curr.second = value;
				}

				if (curr.releaseSlot()) {
					CURRENT.compareAndSet(this, curr, null);
					return curr;
				}
				return null;
			}
		}

		@Override
		public void cancel() {
			for (MergeReduceInner<T> inner : subscribers) {
				inner.cancel();
			}
			super.cancel();
		}

		void innerError(Throwable ex) {
			if(ERROR.compareAndSet(this, null, ex)){
				cancel();
				actual.onError(ex);
			}
			else if(error != ex) {
				Operators.onErrorDropped(ex, actual.currentContext());
			}
		}

		void innerComplete(@Nullable T value) {
			if (value != null) {
				for (; ; ) {
					SlotPair<T> sp = addValue(value);

					if (sp != null) {

						try {
							value = Objects.requireNonNull(reducer.apply(sp.first,
									sp.second), "The reducer returned a null value");
						}
						catch (Throwable ex) {
							innerError(Operators.onOperatorError(this, ex,
									actual.currentContext()));
							return;
						}
					}
					else {
						break;
					}
				}
			}

			if (REMAINING.decrementAndGet(this) == 0) {
				SlotPair<T> sp = current;
				CURRENT.lazySet(this, null);

				if (sp != null) {
					complete(sp.first);
				}
				else {
					actual.onComplete();
				}
			}
		}
	}

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

		final MergeReduceMain<T> parent;

		final BiFunction<T, T, T> reducer;

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

		T value;

		boolean done;

		MergeReduceInner(MergeReduceMain<T> parent,
				BiFunction<T, T, T> reducer) {
			this.parent = parent;
			this.reducer = reducer;
		}

		@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.TERMINATED) return done;
			if (key == Attr.ACTUAL) return parent;
			if (key == Attr.BUFFERED) return value != null ? 1 : 0;
			if (key == Attr.PREFETCH) return Integer.MAX_VALUE;

			return null;
		}

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

		@Override
		public void onNext(T t) {
			if (done) {
				Operators.onNextDropped(t, currentContext());
				return;
			}
			T v = value;

			if (v == null) {
				value = t;
			}
			else {

				try {
					v = Objects.requireNonNull(reducer.apply(v, t), "The reducer returned a null value");
				}
				catch (Throwable ex) {
					onError(Operators.onOperatorError(s, ex, t, currentContext()));
					return;
				}

				value = v;
			}
		}

		@Override
		public void onError(Throwable t) {
			if (done) {
				Operators.onErrorDropped(t, parent.currentContext());
				return;
			}
			done = true;
			parent.innerError(t);
		}

		@Override
		public void onComplete() {
			if (done) {
				return;
			}
			done = true;
			parent.innerComplete(value);
		}

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

	static final class SlotPair<T> {

		T first;

		T second;

		volatile int acquireIndex;
		@SuppressWarnings("rawtypes")
		static final AtomicIntegerFieldUpdater<SlotPair> ACQ =
				AtomicIntegerFieldUpdater.newUpdater(SlotPair.class, "acquireIndex");

		volatile int releaseIndex;
		@SuppressWarnings("rawtypes")
		static final AtomicIntegerFieldUpdater<SlotPair> REL =
				AtomicIntegerFieldUpdater.newUpdater(SlotPair.class, "releaseIndex");

		int tryAcquireSlot() {
			for (; ; ) {
				int acquired = acquireIndex;
				if (acquired >= 2) {
					return -1;
				}

				if (ACQ.compareAndSet(this, acquired, acquired + 1)) {
					return acquired;
				}
			}
		}

		boolean releaseSlot() {
			return REL.incrementAndGet(this) == 2;
		}
	}
}