<commit_before>/*
 * Copyright (c) 2011-Present VMware Inc. or its affiliates, 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.Queue;

import org.reactivestreams.Subscriber;

import reactor.core.Disposable;
import reactor.core.Exceptions;
import reactor.core.scheduler.Scheduler;
import reactor.util.annotation.Nullable;
import reactor.util.concurrent.Queues;
import reactor.util.context.Context;

/**
 * Sinks are constructs through which Reactive Streams signals can be programmatically pushed, with {@link Flux} or {@link Mono}
 * semantics. These standalone sinks expose {@code emit} methods that return an {@link Emission} enum, allowing to
 * softly fail in case the attempted signal is inconsistent with the spec and/or the state of the sink.
 * <p>
 * This class exposes a collection of ({@link Sinks.Many} builders and {@link Sinks.One} factories.
 *
 * @author <NAME>
 * @author <NAME>
 */
public final class Sinks {

	private Sinks() {
	}

	/**
	 * A {@link Sinks.Empty} which exclusively produces one terminal signal: error or complete.
	 * It has the following characteristics:
	 * <ul>
	 *     <li>Multicast</li>
	 *     <li>Backpressure : this sink does not need any demand since it can only signal error or completion</li>
	 *     <li>Replaying: Replay the terminal signal (error or complete).</li>
	 * </ul>
	 * Use {@link Sinks.Empty#asMono()} to expose the {@link Mono} view of the sink to downstream consumers.
	 */
	public static <T> Sinks.Empty<T> empty() {
		return new VoidProcessor<T>();
	}

	/**
	 * A {@link Sinks.One} that works like a conceptual promise: it can be completed
	 * with or without a value at any time, but only once. This completion is replayed to late subscribers.
	 * Calling {@link One#emitValue(Object)} (or {@link One#tryEmitValue(Object)}) is enough and will
	 * implicitly produce a {@link Subscriber#onComplete()} signal as well.
	 * <p>
	 * Use {@link One#asMono()} to expose the {@link Mono} view of the sink to downstream consumers.
	 */
	public static <T> Sinks.One<T> one() {
		return new NextProcessor<>(null);
	}

	/**
	 * Help building {@link Sinks.Many} sinks that will broadcast multiple signals to one or more {@link Subscriber}.
	 * <p>
	 * Use {@link Many#asFlux()} to expose the {@link Flux} view of the sink to the downstream consumers.
	 *
	 * @return {@link ManySpec}
	 */
	public static ManySpec many() {
		return SinksSpecs.MANY_SPEC;
	}

	/**
	 * Represents the immediate status of a signal emission. This does not guarantee that a signal is consumed,
	 * it simply refers to the sink state when an emit method is invoked. This is a particularly important
	 * distinction with regards to {@link #FAIL_CANCELLED} which means the sink is -now- interrupted and emission can't
	 * proceed. Consequently, it is possible to emit a signal and obtain an "OK" status even if an in-flight cancellation
	 * is happening. This is due to the async nature of these actions: producer emits while consumer can interrupt independently.
	 */
	public enum Emission {
		/**
		 * Has successfully emitted the signal
		 */
		OK,
		/**
		 * Has failed to emit the signal because the sink was previously terminated successfully or with an error
		 */
		FAIL_TERMINATED,
		/**
		 * Has failed to emit the signal because the sink does not have buffering capacity left
		 */
		FAIL_OVERFLOW,
		/**
		 * Has failed to emit the signal because the sink was previously been interrupted by its consumer
		 */
		FAIL_CANCELLED;

		/**
		 * Has successfully emitted the signal
		 */
		public boolean hasEmitted() {
			return this == OK;
		}

		/**
		 * Has failed to emit the signal because the sink was previously terminated successfully or with an error, or
		 * has been cancelled or has overflowed an underlying fixed size buffer.
		 */
		public boolean hasFailed() {
			return this != OK;
		}

		/**
		 * Has failed to emit the signal because the sink does not have buffering capacity left
		 */
		public boolean hasOverflowed() {
			return this == FAIL_OVERFLOW;
		}

		/**
		 * Has failed to emit the signal because the sink was previously been interrupted by its consumer
		 */
		public boolean wasPreviouslyCancelled() {
			return this == FAIL_CANCELLED;
		}

		/**
		 * Has failed to emit the signal because the sink was previously terminated successfully or with an error
		 */
		public boolean wasPreviouslyTerminated() {
			return this == FAIL_TERMINATED;
		}
	}

	/**
	 * Provides {@link Sinks.Many} specs for sinks which can emit multiple elements
	 */
	public interface ManySpec {<commit_msg>fix #2329 Improve Emission API: polish status helpers, add orThrow()<commit_after>/*
 * Copyright (c) 2011-Present VMware Inc. or its affiliates, 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.Queue;

import org.reactivestreams.Subscriber;

import reactor.core.Disposable;
import reactor.core.Exceptions;
import reactor.core.scheduler.Scheduler;
import reactor.util.annotation.Nullable;
import reactor.util.concurrent.Queues;
import reactor.util.context.Context;

/**
 * Sinks are constructs through which Reactive Streams signals can be programmatically pushed, with {@link Flux} or {@link Mono}
 * semantics. These standalone sinks expose {@code emit} methods that return an {@link Emission} enum, allowing to
 * softly fail in case the attempted signal is inconsistent with the spec and/or the state of the sink.
 * <p>
 * This class exposes a collection of ({@link Sinks.Many} builders and {@link Sinks.One} factories.
 *
 * @author <NAME>
 * @author <NAME>
 */
public final class Sinks {

	private Sinks() {
	}

	/**
	 * A {@link Sinks.Empty} which exclusively produces one terminal signal: error or complete.
	 * It has the following characteristics:
	 * <ul>
	 *     <li>Multicast</li>
	 *     <li>Backpressure : this sink does not need any demand since it can only signal error or completion</li>
	 *     <li>Replaying: Replay the terminal signal (error or complete).</li>
	 * </ul>
	 * Use {@link Sinks.Empty#asMono()} to expose the {@link Mono} view of the sink to downstream consumers.
	 */
	public static <T> Sinks.Empty<T> empty() {
		return new VoidProcessor<T>();
	}

	/**
	 * A {@link Sinks.One} that works like a conceptual promise: it can be completed
	 * with or without a value at any time, but only once. This completion is replayed to late subscribers.
	 * Calling {@link One#emitValue(Object)} (or {@link One#tryEmitValue(Object)}) is enough and will
	 * implicitly produce a {@link Subscriber#onComplete()} signal as well.
	 * <p>
	 * Use {@link One#asMono()} to expose the {@link Mono} view of the sink to downstream consumers.
	 */
	public static <T> Sinks.One<T> one() {
		return new NextProcessor<>(null);
	}

	/**
	 * Help building {@link Sinks.Many} sinks that will broadcast multiple signals to one or more {@link Subscriber}.
	 * <p>
	 * Use {@link Many#asFlux()} to expose the {@link Flux} view of the sink to the downstream consumers.
	 *
	 * @return {@link ManySpec}
	 */
	public static ManySpec many() {
		return SinksSpecs.MANY_SPEC;
	}

	/**
	 * Represents the immediate status of a signal emission. This does not guarantee that a signal is consumed,
	 * it simply refers to the sink state when an emit method is invoked. This is a particularly important
	 * distinction with regards to {@link #FAIL_CANCELLED} which means the sink is -now- interrupted and emission can't
	 * proceed. Consequently, it is possible to emit a signal and obtain an "OK" status even if an in-flight cancellation
	 * is happening. This is due to the async nature of these actions: producer emits while consumer can interrupt independently.
	 */
	public enum Emission {
		/**
		 * Has successfully emitted the signal
		 */
		OK,
		/**
		 * Has failed to emit the signal because the sink was previously terminated successfully or with an error
		 */
		FAIL_TERMINATED,
		/**
		 * Has failed to emit the signal because the sink does not have buffering capacity left
		 */
		FAIL_OVERFLOW,
		/**
		 * Has failed to emit the signal because the sink was previously interrupted by its consumer
		 */
		FAIL_CANCELLED;

		/**
		 * Has successfully emitted the signal
		 */
		public boolean hasSucceeded() {
			return this == OK;
		}

		/**
		 * Has failed to emit the signal because the sink was previously terminated successfully or with an error, or
		 * has been cancelled or has overflowed its buffering capacity in terms of backpressure.
		 */
		public boolean hasFailed() {
			return this != OK;
		}

		/**
		 * Easily convert from an {@link Emission} to throwing an exception on {@link #hasFailed() failure cases}.
		 * This is useful if throwing is the most relevant way of dealing with a failed emission attempt.
		 * See also {@link #orThrowWithCause(Throwable)} in case of an {@link One#emitError(Throwable) emitError}
		 * failure for which you want to propagate the originally pushed {@link Exception}.
		 *
		 * @see #orThrowWithCause(Throwable)
		 */
		public void orThrow() {
			if (this == OK) return;

			throw new EmissionException(this);
		}

		/**
		 * Easily convert from an {@link Emission} to throwing an exception on {@link #hasFailed() failure cases}.
		 * This is useful if throwing is the most relevant way of dealing with failed {@link One#emitError(Throwable)}
		 * attempt, in which case you probably wants to propagate the originally pushed {@link Exception}.
		 *
		 * @see #orThrow()
		 */
		public void orThrowWithCause(Throwable cause) {
			if (this == OK) return;

			throw new EmissionException(cause, this);
		}
	}

	/**
	 * An exception representing a {@link Emission#hasFailed() failed} {@link Emission}.
	 * The exact type of failure can be found via {@link #getReason()}.
	 */
	public static final class EmissionException extends IllegalStateException {

		final Emission reason;

		public EmissionException(Emission reason) {
			super("Sink emission failed with " + reason);
			this.reason = reason;
		}

		public EmissionException(Throwable cause, Emission reason) {
			super("Sink emission failed with " + reason, cause);
			this.reason = reason;
		}

		/**
		 * Get the failure {@link Emission} code that is represented by this exception.
		 *
		 * @return the {@link Emission}
		 */
		public Emission getReason() {
			return this.reason;
		}
	}

	/**
	 * Provides {@link Sinks.Many} specs for sinks which can emit multiple elements
	 */
	public interface ManySpec {
{'fraction_non_alphanumeric': 0.09007844151366262, 'fraction_numerical': 0.0017239893112662701, 'mean_word_length': 3.6333865814696487, 'pattern_counts': {'":': 0, '<': 41, '<?xml version=': 0, '>': 41, 'https://': 2, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3954249', 'n_tokens_mistral': 3434, 'n_tokens_neox': 3054, 'n_words': 1430}