<commit_before>/*
 * Copyright (c) 2011-2016 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.concurrent.Callable;

import org.reactivestreams.Processor;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import reactor.core.flow.Producer;
import reactor.core.flow.Receiver;
import reactor.core.state.Backpressurable;
import reactor.core.subscriber.SignalEmitter;
import reactor.core.util.Assert;
import reactor.core.util.BackpressureUtils;
import reactor.core.util.EmptySubscription;
import reactor.core.util.Exceptions;
import reactor.fn.Consumer;
import reactor.fn.Function;

/**
 * A base processor that expose {@link Flux} API, {@link Processor} and generic {@link Consumer} for {@link Runnable}
 * {@link SchedulerGroup scheduling contract}.
 *
 * Factories available allow arbitrary {@link FluxProcessor} creation from blackboxed and external reactive components.
 *
 * @author <NAME>
 * @since 2.0.2, 2.5
 */
public abstract class FluxProcessor<IN, OUT> extends Flux<OUT>
		implements Processor<IN, OUT>, Backpressurable, Receiver, Consumer<IN> {

	/**
	 * Create an asynchronously {@link Flux#dispatchOn(Callable) dispatched} {@link FluxProcessor} multicast/topic
	 * relay.
	 * Like {@link Flux#dispatchOn(Callable)} the scheduler resources will be cleaned accordingly to the {@link Runnable} {@literal null} protocol.
	 * Unlike {@link TopicProcessor} or {@link WorkQueueProcessor}, the threading resources are not dedicated nor
	 * mandatory.
	 *
	 * @param <IN> The relayed data type
	 *
	 * @return a new asynchronous {@link FluxProcessor}
	 */
	public static <IN> FluxProcessor<IN, IN> async(final Callable<? extends Consumer<Runnable>> schedulerFactory) {
		FluxProcessor<IN, IN> emitter = EmitterProcessor.create();
		return create(emitter, emitter.dispatchOn(schedulerFactory));
	}

	/**
	 * Blackbox a given
	 * {@link Subscriber} receiving type with a transforming function returning the producing side {@link Publisher}.
	 *
	 * <pre>
	 * {@code
	 *   Processor<String, String> asyncLowerCase =
	 *      FluxProcessor.blackbox(TopicProcessor.create(), input -> input.map(String::toLowerCase));
	 * }
	 *</pre>
	 *
	 * @param input an input {@link Subscriber}
	 * @param  blackboxFunction the
	 * {@link Function} given the input subscriber to compose on and return {@link Publisher}
	 * @param <IN> the reified received type
	 * @param <OUT> the reified produced type
	 *
	 * @return a blackboxed chain as {@link FluxProcessor}
	 */
	public static <IN, OUT, E extends Subscriber<IN>> FluxProcessor<IN, OUT> blackbox(
			final E input,
			final Function<E, ? extends Publisher<OUT>> blackboxFunction) {
		return create(input, blackboxFunction.apply(input));
	}

	/**
	 * Blackbox an arbitrary {@link Flux} operation chain into a {@link FluxProcessor} that can be subscribed once
	 * only.
	 * <p>
	 * <pre>
	 * {@code
	 *  Processor<String, Integer> stringToInt =
	 *      FluxProcessor.blackbox(input -> input.map(Integer::parseString));
	 * }
	 *</pre>
	 *
	 * @param blackboxFunction the {@link Function} given a {@link Flux} to compose on and return {@link Publisher}
	 * @param <IN> the reified received type
	 * @param <OUT> the reified produced type
	 *
	 * @return a blackboxed chain as {@link FluxProcessor}
	 */
	public static <IN, OUT> FluxProcessor<IN, OUT> blackbox(final Function<Flux<IN>, ? extends Publisher<OUT>> blackboxFunction) {
		FluxPassthrough<IN> passthrough = new FluxPassthrough<>();
		return create(passthrough, blackboxFunction.apply(passthrough));
	}

	/**
	 * Create a passthrough {@link FluxProcessor} relay blocking when overrun.
	 * <p>
	 * It will use a deferred blocking {@link SignalEmitter} and pass it to {@link #onSubscribe(Subscription)}.
	 * Multiple producer can share the returned reference IF and only IF they don't publish concurrently. In this
	 * very case, implementor must take care of using a multiproducer capable receiver downstream e.g.
	 * {@link TopicProcessor#share()} or {@link WorkQueueProcessor#share()}.
	 *
	 * @return a new {@link FluxProcessor}
	 */
	public static <IN> FluxProcessor<IN, IN> blocking() {
		FluxPassthrough<IN> passthrough = new FluxPassthrough<>();
		return create(SignalEmitter.blocking(passthrough), passthrough);
	}

	/**
	 * Transform a receiving {@link Subscriber} and a producing {@link Publisher} in a logical {@link FluxProcessor}.
	 * The link between the passed upstream and returned downstream will not be created automatically, e.g. not
	 * subscribed together. A {@link Processor} might choose to have orthogonal sequence input and output.
	 *
	 * @param <IN> the receiving type
	 * @param <OUT> the producing type
	 *
	 * @return a new blackboxed {@link FluxProcessor}
	 */
	public static <IN, OUT> FluxProcessor<IN, OUT> create(final Subscriber<IN> upstream, final Publisher<OUT> downstream) {
		return new DelegateProcessor<>(downstream, upstream);
	}

	Subscription upstreamSubscription;

	protected FluxProcessor() {
	}

	/**
	 * Trigger onSubscribe with a stateless subscription to signal this subscriber it can start receiving
	 * onNext, onComplete and onError calls.
	 * <p>
	 * Doing so MAY allow direct UNBOUNDED onXXX calls and MAY prevent {@link org.reactivestreams.Publisher} to subscribe this
	 * subscriber.
	 *
	 * Note that {@link org.reactivestreams.Processor} can extend this behavior to effectively start its subscribers.
	 */
	public FluxProcessor<IN, OUT> start() {
		onSubscribe(EmptySubscription.INSTANCE);
		return this;
	}

	/**
	 * Create a {@link SignalEmitter} and attach it via {@link #onSubscribe(Subscription)}.
	 *
	 * @return a new subscribed {@link SignalEmitter}
	 */
	public SignalEmitter<IN> startEmitter() {
		return bindEmitter(true);
	}<commit_msg>polish, javadoc<commit_after>/*
 * Copyright (c) 2011-2016 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.concurrent.Callable;

import org.reactivestreams.Processor;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import reactor.core.flow.Producer;
import reactor.core.flow.Receiver;
import reactor.core.state.Backpressurable;
import reactor.core.subscriber.SignalEmitter;
import reactor.core.util.Assert;
import reactor.core.util.BackpressureUtils;
import reactor.core.util.EmptySubscription;
import reactor.core.util.Exceptions;
import reactor.fn.Consumer;
import reactor.fn.Function;

/**
 * A base processor that expose {@link Flux} API, {@link Processor} and generic {@link Consumer} for {@link Runnable}
 * {@link SchedulerGroup scheduling contract}.
 *
 * Factories available allow arbitrary {@link FluxProcessor} creation from blackboxed and external reactive components.
 *
 * @author <NAME>
 * @since 2.0.2, 2.5
 */
public abstract class FluxProcessor<IN, OUT> extends Flux<OUT>
		implements Processor<IN, OUT>, Backpressurable, Receiver, Consumer<IN> {

	/**
	 * Create an asynchronously {@link Flux#dispatchOn(Callable) dispatched} {@link FluxProcessor} multicast/topic
	 * relay.
	 * Like {@link Flux#dispatchOn(Callable)} the scheduler resources will be cleaned accordingly to the {@link Runnable} {@literal null} protocol.
	 * Unlike {@link TopicProcessor} or {@link WorkQueueProcessor}, the threading resources are not dedicated nor
	 * mandatory.
	 *
	 * @param <IN> The relayed data type
	 *
	 * @return a new asynchronous {@link FluxProcessor}
	 */
	public static <IN> FluxProcessor<IN, IN> async(final Callable<? extends Consumer<Runnable>> schedulerFactory) {
		FluxProcessor<IN, IN> emitter = EmitterProcessor.create();
		return create(emitter, emitter.dispatchOn(schedulerFactory));
	}

	/**
	 * Blackbox a given
	 * {@link Subscriber} receiving type with a transforming function returning the producing side {@link Publisher}.
	 *
	 * <pre>
	 * {@code
	 *   Processor<String, String> asyncLowerCase =
	 *      FluxProcessor.blackbox(TopicProcessor.create(), input -> input.map(String::toLowerCase));
	 * }
	 *</pre>
	 *
	 * @param input an input {@link Subscriber}
	 * @param  blackboxFunction the
	 * {@link Function} given the input subscriber to compose on and return {@link Publisher}
	 * @param <IN> the reified received type
	 * @param <OUT> the reified produced type
	 *
	 * @return a blackboxed chain as {@link FluxProcessor}
	 */
	public static <IN, OUT, E extends Subscriber<IN>> FluxProcessor<IN, OUT> blackbox(
			final E input,
			final Function<E, ? extends Publisher<OUT>> blackboxFunction) {
		return create(input, blackboxFunction.apply(input));
	}

	/**
	 * Blackbox an arbitrary {@link Flux} operation chain into a {@link FluxProcessor} that can be subscribed once
	 * only.
	 * <p>
	 * <pre>
	 * {@code
	 *  Processor<String, Integer> stringToInt =
	 *      FluxProcessor.blackbox(input -> input.map(Integer::parseString));
	 * }
	 *</pre>
	 *
	 * @param blackboxFunction the {@link Function} given a {@link Flux} to compose on and return {@link Publisher}
	 * @param <IN> the reified received type
	 * @param <OUT> the reified produced type
	 *
	 * @return a blackboxed chain as {@link FluxProcessor}
	 */
	public static <IN, OUT> FluxProcessor<IN, OUT> blackbox(final Function<Flux<IN>, ? extends Publisher<OUT>> blackboxFunction) {
		FluxPassthrough<IN> passthrough = new FluxPassthrough<>();
		return create(passthrough, blackboxFunction.apply(passthrough));
	}

	/**
	 * Create a passthrough {@link FluxProcessor} relay blocking when overrun.
	 * <p>
	 * It will use a deferred blocking {@link SignalEmitter} and pass it to {@link #onSubscribe(Subscription)}.
	 * Multiple producer can share the returned reference IF and only IF they don't publish concurrently. In this
	 * very case, implementor must take care of using a multiproducer capable receiver downstream e.g.
	 * {@link TopicProcessor#share()} or {@link WorkQueueProcessor#share()}.
	 *
	 * @return a new {@link FluxProcessor}
	 */
	public static <IN> FluxProcessor<IN, IN> blocking() {
		FluxPassthrough<IN> passthrough = new FluxPassthrough<>();
		return create(SignalEmitter.blocking(passthrough), passthrough);
	}

	/**
	 * Transform a receiving {@link Subscriber} and a producing {@link Publisher} in a logical {@link FluxProcessor}.
	 * The link between the passed upstream and returned downstream will not be created automatically, e.g. not
	 * subscribed together. A {@link Processor} might choose to have orthogonal sequence input and output.
	 *
	 * @param <IN> the receiving type
	 * @param <OUT> the producing type
	 *
	 * @return a new blackboxed {@link FluxProcessor}
	 */
	public static <IN, OUT> FluxProcessor<IN, OUT> create(final Subscriber<IN> upstream, final Publisher<OUT> downstream) {
		return new DelegateProcessor<>(downstream, upstream);
	}

	Subscription upstreamSubscription;

	protected FluxProcessor() {
	}

	/**
	 * Trigger onSubscribe with a stateless subscription to signal this subscriber it can start receiving
	 * onNext, onComplete and onError calls.
	 * <p>
	 * Doing so MAY allow direct UNBOUNDED onXXX calls and MAY prevent {@link org.reactivestreams.Publisher} to subscribe this
	 * subscriber.
	 *
	 * Note that {@link org.reactivestreams.Processor} can extend this behavior to effectively start its subscribers.
	 *
	 * @return this {@link FluxProcessor}
	 */
	public FluxProcessor<IN, OUT> start() {
		onSubscribe(EmptySubscription.INSTANCE);
		return this;
	}

	/**
	 * Create a {@link SignalEmitter} and attach it via {@link #onSubscribe(Subscription)}.
	 *
	 * @return a new subscribed {@link SignalEmitter}
	 */
	public SignalEmitter<IN> startEmitter() {
		return bindEmitter(true);
	}
{'fraction_non_alphanumeric': 0.10441233893010543, 'fraction_numerical': 0.0026552128074970715, 'mean_word_length': 4.538927335640138, 'pattern_counts': {'":': 0, '<': 101, '<?xml version=': 0, '>': 105, 'https://': 0, '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': '3628976', 'n_tokens_mistral': 3904, 'n_tokens_neox': 3552, 'n_words': 1297}