content: /*
   * Copyright (C) 2020 ActiveJ LLC.
   *
   * 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 io.activej.csp.queue;
  
  import io.activej.async.process.AsyncCloseable;
  import io.activej.common.Checks;
  import io.activej.common.recycle.Recyclers;
  import io.activej.promise.Promise;
  import io.activej.promise.SettablePromise;
  import io.activej.reactor.ImplicitlyReactive;
  import org.jetbrains.annotations.Nullable;
  
  import static io.activej.common.Checks.checkState;
  import static io.activej.reactor.Reactive.checkInReactorThread;
  import static java.lang.Integer.numberOfLeadingZeros;
  import static java.lang.Math.max;
  
  /**
   * Represents a queue of elements which you can {@code put} and {@code take}.
   * In order to mark if an object is pending put or take to/from the queue,
   * there are corresponding {@code put} and {@code take} {@link SettablePromise}s.
   *
   * @param <T> the type of values that are stored in the buffer
   */
  public final class ChannelBuffer<T> extends ImplicitlyReactive implements ChannelQueue<T> {
  	private static final boolean CHECKS = Checks.isEnabled(ChannelBuffer.class);
  
  	private Exception exception;
  
  	private Object[] elements;
  	private int tail;
  	private int head;
  
  	private final int bufferMinSize;
  	private final int bufferMaxSize;
  
  	private @Nullable SettablePromise<Void> put;
  	private @Nullable SettablePromise<T> take;
  
  	private @Nullable AsyncCloseable closeable;
  
  	/**
  	 * @see #ChannelBuffer(int, int)
  	 */
  	public ChannelBuffer(int bufferSize) {
  		this(0, bufferSize);
  	}
  
  	/**
  	 * Creates a ChannelBuffer with the buffer size of the next highest
  	 * power of 2 (for example, if {@code bufferMinxSize = 113}, a buffer
  	 * of 128 elements will be created).
  	 *
  	 * @param bufferMinSize a minimal amount of elements in the buffer
  	 * @param bufferMaxSize a max amount of elements in the buffer
  	 */
  	public ChannelBuffer(int bufferMinSize, int bufferMaxSize) {
  		this.bufferMinSize = bufferMinSize + 1;
  		this.bufferMaxSize = bufferMaxSize;
  		this.elements = new Object[1 << (32 - numberOfLeadingZeros(max(16, this.bufferMinSize) - 1))];
  	}
  
  	public void setCloseable(@Nullable AsyncCloseable closeable) {
  		this.closeable = closeable;
  	}
  
  	/**
  	 * Checks if this buffer is saturated by comparing
  	 * its current size with {@code bufferMaxSize}.
  	 *
  	 * @return {@code true} if this buffer size is greater
  	 * than {@code bufferMaxSize}, otherwise returns {@code false}
  	 */
  	@Override
  	public boolean isSaturated() {
  		return size() > bufferMaxSize;
  	}
  
  	/**
  	 * Checks if this buffer will be saturated if at
  	 * least one more element will be added, by comparing
  	 * its current size with {@code bufferMaxSize}.
  	 *
  	 * @return {@code true} if this buffer size is
  	 * bigger or equal to the {@code bufferMaxSize},
  	 * otherwise returns {@code false}
  	 */
  	public boolean willBeSaturated() {
  		return size() >= bufferMaxSize;
  	}
  
  	/**
  	 * Checks if this buffer has fewer
  	 * elements than {@code bufferMinSize}.
  	 *
  	 * @return {@code true} if this buffer size is
  	 * smaller than {@code bufferMinSize},
  	 * otherwise returns {@code false}
  	 */
  	@Override
  	public boolean isExhausted() {
  		return size() < bufferMinSize;
  	}
  
  	/**
  	 * Checks if this buffer will have fewer elements
  	 * than {@code bufferMinSize}, if at least one
  	 * more element will be taken, by comparing its
  	 * current size with {@code bufferMinSize}.
  	 *
  	 * @return {@code true} if this buffer size is
  	 * smaller or equal to the {@code bufferMinSize},
  	 * otherwise returns {@code false}
  	 */
  	public boolean willBeExhausted() {
  		return size() <= bufferMinSize;
  	}
  
  	/**
  	 * Checks if this buffer contains elements.
  	 *
  	 * @return {@code true} if {@code tail}
  	 * and {@code head} values are equal,
  	 * otherwise {@code false}
  	 */
  	public boolean isEmpty() {
  		return tail == head;
  	}
  
  	/**
  	 * Returns amount of elements in this buffer.
  	 */
  	public int size() {
  		return tail - head;
  	}
  
  	/**
  	 * Adds provided item to the buffer and resets current {@code take}.
  	 */
  	public void add(@Nullable T item) throws Exception {
  		if (CHECKS) checkInReactorThread(this);
  		if (exception == null) {
  			if (take != null) {
  				assert isEmpty();
  				SettablePromise<T> take = this.take;
  				this.take = null;
  				take.set(item);
  				if (exception != null) throw exception;
  				return;
  			}
  
  			doAdd(item);
  		} else {
  			Recyclers.recycle(item);
  			throw exception;
  		}
  	}
  
  	private void doAdd(@Nullable T value) {
  		elements[tail++ & (elements.length - 1)] = value;
  	}
  
  	/**
  	 * Returns the head of the buffer if it is not empty,
  	 * otherwise returns {@code null}. Increases the value of {@code head}.
  	 * <p>
  	 * If the buffer will have fewer elements than {@code bufferMinSize}
  	 * after this poll and {@code put} promise is not {@code null},
  	 * {@code put} will be set {@code null} after the poll.
  	 * <p>
  	 * If current {@code exception} is not {@code null},
  	 * it will be thrown.
  	 *
  	 * @return head element of this buffer
  	 * index if the buffer is not empty, otherwise {@code null}
  	 * @throws Exception if current {@code exception}
  	 *                   is not {@code null}
  	 */
  	public @Nullable T poll() throws Exception {
  		if (CHECKS) checkInReactorThread(this);
  		if (exception != null) throw exception;
  
  		if (put != null && willBeExhausted()) {
  			T item = doPoll();
  			SettablePromise<Void> put = this.put;
  			this.put = null;
  			put.set(null);
  			return item;
  		}
  
  		return !isEmpty() ? doPoll() : null;
  	}
  
  	private T doPoll() {
  		assert head != tail;
  		int pos = head++ & (elements.length - 1);
  		@SuppressWarnings("unchecked")
  		T result = (T) elements[pos];
  		elements[pos] = null;     // Must null out slot
  		return result;
  	}
  
  	/**
  	 * Puts {@code value} in this buffer and increases {@code tail} value.
  	 * <p>
  	 * Current {@code put} must be {@code null}. If
  	 * current {@code exception} is not {@code null},
  	 * a promise of this exception will be returned and
  	 * the {@code value} will be recycled.
  	 * <p>
  	 * If this {@code take} is not {@code null}, the value
  	 * will be set directly to the {@code set}, without
  	 * adding to the buffer.
  	 *
  	 * @param item a value passed to the buffer
  	 * @return promise of {@code null} or {@code exception}
  	 * as a marker of completion
  	 */
  	@Override
  	public Promise<Void> put(@Nullable T item) {
  		if (CHECKS) {
  			checkInReactorThread(this);
  			checkState(put == null, "Previous put() has not finished yet");
  		}
  		if (exception == null) {
  			if (take != null) {
  				assert isEmpty();
  				SettablePromise<T> take = this.take;
  				this.take = null;
  				take.set(item);
  				return Promise.complete();
  			}
  
  			doAdd(item);
  
  			if (isSaturated()) {
  				put = new SettablePromise<>();
  				return put;
  			} else {
  				return Promise.complete();
  			}
  		} else {
  			Recyclers.recycle(item);
  			return Promise.ofException(exception);
  		}
  	}
  
  	/**
  	 * Returns a promise of the head of
  	 * the {@code buffer} if it is not empty.
  	 * <p>
  	 * If this buffer will be exhausted after this
  	 * take and {@code put} promise is not {@code null},
  	 * {@code put} will be set {@code null} after the poll.
  	 * <p>
  	 * Current {@code take} must be {@code null}. If
  	 * current {@code exception} is not {@code null},
  	 * a promise of this exception will be returned.
  	 *
  	 * @return promise of element taken from the buffer
  	 */
  	@Override
  	public Promise<T> take() {
  		if (CHECKS) {
  			checkInReactorThread(this);
  			checkState(take == null, "Previous take() has not finished yet");
  		}
  		if (exception == null) {
  			if (put != null && willBeExhausted()) {
  				assert !isEmpty();
  				T item = doPoll();
  				SettablePromise<Void> put = this.put;
  				this.put = null;
  				put.set(null);
  				return Promise.of(item);
  			}
  
  			if (!isEmpty()) {
  				return Promise.of(doPoll());
  			}
  
  			take = new SettablePromise<>();
  			return take;
  		} else {
  			return Promise.ofException(exception);
  		}
  	}
  
  	/**
  	 * Closes the buffer if this {@code exception} is not
  	 * {@code null}. Recycles all elements of the buffer and
  	 * sets {@code elements}, {@code put} and {@code take} to
  	 * {@code null}.
  	 *
  	 * @param e exception that is used to close buffer with
  	 */
  	@Override
  	public void closeEx(Exception e) {
  		checkInReactorThread(this);
  		if (exception != null) return;
  		exception = e;
  		if (put != null) {
  			put.setException(e);
  			put = null;
  		}
  		if (take != null) {
  			take.setException(e);
  			take = null;
  		}
  		for (int i = head; i != tail; i = (i + 1) & (elements.length - 1)) {
  			Recyclers.recycle(elements[i]);
  		}
  		//noinspection AssignmentToNull - resource release
  		elements = null;
  		if (closeable != null) {
  			closeable.closeEx(e);
  			closeable = null;
  		}
  	}
  
  	public @Nullable Exception getException() {
  		return exception;
  	}
  }
  