/*
 * 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.Queue;
import java.util.concurrent.atomic.AtomicLongFieldUpdater;
import java.util.function.BooleanSupplier;

import org.reactivestreams.Subscriber;
import reactor.core.CoreSubscriber;
import reactor.util.annotation.Nullable;

abstract class DrainUtils {

	
Indicates the source completed and the value field is ready to be emitted.

The AtomicLong (this) holds the requested amount in bits 0..62 so there is room
for one signal bit. This also means the standard request accounting helper method doesn't work.
/**
	 * Indicates the source completed and the value field is ready to be emitted.
	 * <p>
	 * The AtomicLong (this) holds the requested amount in bits 0..62 so there is room
	 * for one signal bit. This also means the standard request accounting helper method doesn't work.
	 */
	static final long COMPLETED_MASK = 0x8000_0000_0000_0000L;
	static final long REQUESTED_MASK = 0x7FFF_FFFF_FFFF_FFFFL;

	
Perform a potential post-completion request accounting.
Params:
n – the requested amount
actual – the consumer of values
queue – the queue holding the available values
field – the field updater for the requested amount
instance – the parent instance for the requested field
isCancelled – callback to detect cancellation
Type parameters:
<T> – the output value type
<F> – the field type holding the requested amount
Returns:
true if the state indicates a completion state./**
	 * Perform a potential post-completion request accounting.
	 * 
	 * @param <T> the output value type
	 * @param <F> the field type holding the requested amount
	 * @param n the requested amount
	 * @param actual the consumer of values
	 * @param queue the queue holding the available values
	 * @param field the field updater for the requested amount
	 * @param instance the parent instance for the requested field
	 * @param isCancelled callback to detect cancellation
	 * @return true if the state indicates a completion state.
	 */
	static <T, F> boolean postCompleteRequest(long n,
			Subscriber<? super T> actual,
			Queue<T> queue,
			AtomicLongFieldUpdater<F> field,
			F instance,
			BooleanSupplier isCancelled) {

		for (; ; ) {
			long r = field.get(instance);

			// extract the current request amount
			long r0 = r & REQUESTED_MASK;

			// preserve COMPLETED_MASK and calculate new requested amount
			long u = (r & COMPLETED_MASK) | Operators.addCap(r0, n);

			if (field.compareAndSet(instance, r, u)) {
				// (complete, 0) -> (complete, n) transition then replay
				if (r == COMPLETED_MASK) {

					postCompleteDrain(n | COMPLETED_MASK, actual, queue, field, instance, isCancelled);

					return true;
				}
				// (active, r) -> (active, r + n) transition then continue with requesting from upstream
				return false;
			}
		}

	}

	
Drains the queue either in a pre- or post-complete state.
Params:
n – the requested amount
actual – the consumer of values
queue – the queue holding available values
field – the field updater holding the requested amount
instance – the parent instance of the requested field
isCancelled – callback to detect cancellation
Returns:
true if the queue was completely drained or the drain process was cancelled/**
	 * Drains the queue either in a pre- or post-complete state.
	 *
	 * @param n the requested amount
	 * @param actual the consumer of values
	 * @param queue the queue holding available values
	 * @param field the field updater holding the requested amount
	 * @param instance the parent instance of the requested field
	 * @param isCancelled callback to detect cancellation
	 * @return true if the queue was completely drained or the drain process was cancelled
	 */
	static <T, F> boolean postCompleteDrain(long n,
			Subscriber<? super T> actual,
			Queue<T> queue,
			AtomicLongFieldUpdater<F> field,
			F instance,
			BooleanSupplier isCancelled) {

// TODO enable fast-path
//        if (n == -1 || n == Long.MAX_VALUE) {
//            for (;;) {
//                if (isDisposed.getAsBoolean()) {
//                    break;
//                }
//
//                T v = queue.poll();
//
//                if (v == null) {
//                    actual.onComplete();
//                    break;
//                }
//
//                actual.onNext(v);
//            }
//
//            return true;
//        }

		long e = n & COMPLETED_MASK;

		for (; ; ) {

			while (e != n) {
				if (isCancelled.getAsBoolean()) {
					return true;
				}

				T t = queue.poll();

				if (t == null) {
					actual.onComplete();
					return true;
				}

				actual.onNext(t);
				e++;
			}

			if (isCancelled.getAsBoolean()) {
				return true;
			}

			if (queue.isEmpty()) {
				actual.onComplete();
				return true;
			}

			n = field.get(instance);

			if (n == e) {

				n = field.addAndGet(instance, -(e & REQUESTED_MASK));

				if ((n & REQUESTED_MASK) == 0L) {
					return false;
				}

				e = n & COMPLETED_MASK;
			}
		}

	}

	
Tries draining the queue if the source just completed.
Params:
actual – the consumer of values
queue – the queue holding available values
field – the field updater holding the requested amount
instance – the parent instance of the requested field
isCancelled – callback to detect cancellation
Type parameters:
<T> – the output value type
<F> – the field type holding the requested amount/**
	 * Tries draining the queue if the source just completed.
	 *
     * @param <T> the output value type
     * @param <F> the field type holding the requested amount
	 * @param actual the consumer of values
	 * @param queue the queue holding available values
	 * @param field the field updater holding the requested amount
	 * @param instance the parent instance of the requested field
	 * @param isCancelled callback to detect cancellation
	 */
	public static <T, F> void postComplete(CoreSubscriber<? super T> actual,
			Queue<T> queue,
			AtomicLongFieldUpdater<F> field,
			F instance,
			BooleanSupplier isCancelled) {

		if (queue.isEmpty()) {
			actual.onComplete();
			return;
		}

		if (postCompleteDrain(field.get(instance), actual, queue, field, instance, isCancelled)) {
			return;
		}

		for (; ; ) {
			long r = field.get(instance);

			if ((r & COMPLETED_MASK) != 0L) {
				return;
			}

			long u = r | COMPLETED_MASK;
			// (active, r) -> (complete, r) transition
			if (field.compareAndSet(instance, r, u)) {
				// if the requested amount was non-zero, drain the queue
				if (r != 0L) {
					postCompleteDrain(u, actual, queue, field, instance, isCancelled);
				}

				return;
			}
		}
	}

    
Perform a potential post-completion request accounting.
Params:
n – the request amount
actual – the consumer of values
queue – the queue of available values
field – the field updater for the requested amount
instance – the parent instance of the requested field
isCancelled – callback to detect cancellation
error – if not null, the error to signal after the queue has been drained
Type parameters:
<T> – the output value type
<F> – the field type holding the requested amount
Returns:
true if the state indicates a completion state./**
     * Perform a potential post-completion request accounting.
     *
     * @param <T> the output value type
     * @param <F> the field type holding the requested amount
     * @param n the request amount
     * @param actual the consumer of values
     * @param queue the queue of available values
     * @param field the field updater for the requested amount
     * @param instance the parent instance of the requested field 
     * @param isCancelled callback to detect cancellation
     * @param error if not null, the error to signal after the queue has been drained
     * @return true if the state indicates a completion state.
     */
    public static <T, F> boolean postCompleteRequestDelayError(long n,
            Subscriber<? super T> actual,
            Queue<T> queue,
            AtomicLongFieldUpdater<F> field,
            F instance,
            BooleanSupplier isCancelled, Throwable error) {

        for (; ; ) {
            long r = field.get(instance);

            // extract the current request amount
            long r0 = r & REQUESTED_MASK;

            // preserve COMPLETED_MASK and calculate new requested amount
            long u = (r & COMPLETED_MASK) | Operators.addCap(r0, n);

            if (field.compareAndSet(instance, r, u)) {
                // (complete, 0) -> (complete, n) transition then replay
                if (r == COMPLETED_MASK) {

                    postCompleteDrainDelayError(n | COMPLETED_MASK, actual, queue, field, instance, isCancelled, error);

                    return true;
                }
                // (active, r) -> (active, r + n) transition then continue with requesting from upstream
                return false;
            }
        }

    }

    
Drains the queue either in a pre- or post-complete state, delaying an
optional error to the end of the drain operation.
Params:
n – the requested amount
actual – the consumer of values
queue – the queue holding available values
field – the field updater holding the requested amount
instance – the parent instance of the requested field
isCancelled – callback to detect cancellation
error – the delayed error
Returns:
true if the queue was completely drained or the drain process was cancelled/**
     * Drains the queue either in a pre- or post-complete state, delaying an
     * optional error to the end of the drain operation.
     *
     * @param n the requested amount
     * @param actual the consumer of values
     * @param queue the queue holding available values
     * @param field the field updater holding the requested amount
     * @param instance the parent instance of the requested field
     * @param isCancelled callback to detect cancellation
     * @param error the delayed error
     * @return true if the queue was completely drained or the drain process was cancelled
     */
    static <T, F> boolean postCompleteDrainDelayError(long n,
            Subscriber<? super T> actual,
            Queue<T> queue,
            AtomicLongFieldUpdater<F> field,
            F instance,
            BooleanSupplier isCancelled,
		    @Nullable Throwable error) {

        long e = n & COMPLETED_MASK;

        for (; ; ) {

            while (e != n) {
                if (isCancelled.getAsBoolean()) {
                    return true;
                }

                T t = queue.poll();

                if (t == null) {
                    if (error == null) {
                        actual.onComplete();
                    } else {
                        actual.onError(error);
                    }
                    return true;
                }

                actual.onNext(t);
                e++;
            }

            if (isCancelled.getAsBoolean()) {
                return true;
            }

            if (queue.isEmpty()) {
                if (error == null) {
                    actual.onComplete();
                } else {
                    actual.onError(error);
                }
                return true;
            }

            n = field.get(instance);

            if (n == e) {

                n = field.addAndGet(instance, -(e & REQUESTED_MASK));

                if ((n & REQUESTED_MASK) == 0L) {
                    return false;
                }

                e = n & COMPLETED_MASK;
            }
        }

    }

    
Tries draining the queue if the source just completed.
Params:
actual – the consumer of values
queue – the queue of available values
field – the field updater for the requested amount
instance – the parent instance of the requested field
isCancelled – callback to detect cancellation
error – if not null, the error to signal after the queue has been drained
Type parameters:
<T> – the output value type
<F> – the field type holding the requested amount/**
     * Tries draining the queue if the source just completed.
     *
     * @param <T> the output value type
     * @param <F> the field type holding the requested amount
     * @param actual the consumer of values
     * @param queue the queue of available values
     * @param field the field updater for the requested amount
     * @param instance the parent instance of the requested field 
     * @param isCancelled callback to detect cancellation
     * @param error if not null, the error to signal after the queue has been drained
     */
    public static <T, F> void postCompleteDelayError(CoreSubscriber<? super T> actual,
            Queue<T> queue,
            AtomicLongFieldUpdater<F> field,
            F instance,
            BooleanSupplier isCancelled,
		    @Nullable Throwable error) {

        if (queue.isEmpty()) {
            if (error == null) {
                actual.onComplete();
            } else {
                actual.onError(error);
            }
            return;
        }

        if (postCompleteDrainDelayError(field.get(instance), actual, queue, field, instance, isCancelled, error)) {
            return;
        }

        for (; ; ) {
            long r = field.get(instance);

            if ((r & COMPLETED_MASK) != 0L) {
                return;
            }

            long u = r | COMPLETED_MASK;
            // (active, r) -> (complete, r) transition
            if (field.compareAndSet(instance, r, u)) {
                // if the requested amount was non-zero, drain the queue
                if (r != 0L) {
                    postCompleteDrainDelayError(u, actual, queue, field, instance, isCancelled, error);
                }

                return;
            }
        }
    }

	DrainUtils(){}
}