reactor-core/src/main/java/reactor/core/publisher/ParallelFlux.java
public final Disposable subscribe(
			@Nullable Consumer<? super T> onNext,
			@Nullable Consumer<? super Throwable> onError,
			@Nullable Runnable onComplete,
			@Nullable Consumer<? super Subscription> onSubscribe){

		CorePublisher<T> publisher = Operators.onLastAssembly(this);
		if (publisher instanceof ParallelFlux) {
			@SuppressWarnings("unchecked")
			LambdaSubscriber<? super T>[] subscribers = new LambdaSubscriber[parallelism()];

			int i = 0;
			while(i < subscribers.length){
				subscribers[i++] =
						new LambdaSubscriber<>(onNext, onError, onComplete, onSubscribe);
			}

			((ParallelFlux<T>) publisher).subscribe(subscribers);

			return Disposables.composite(subscribers);
		}
		else {
			LambdaSubscriber<? super T> subscriber =
					new LambdaSubscriber<>(onNext, onError, onComplete, onSubscribe);

			publisher.subscribe(Operators.toCoreSubscriber(new FluxHide.SuppressFuseableSubscriber<>(subscriber)));

			return subscriber;
		}
	}
public final Disposable subscribe(
			@Nullable Consumer<? super T> onNext,
			@Nullable Consumer<? super Throwable> onError,
			@Nullable Runnable onComplete,
			@Nullable Consumer<? super Subscription> onSubscribe){

		CorePublisher<T> publisher = Operators.onLastAssembly(this);
		if (publisher instanceof ParallelFlux) {
			@SuppressWarnings("unchecked")
			LambdaSubscriber<? super T>[] subscribers = new LambdaSubscriber[parallelism()];

			int i = 0;
			while(i < subscribers.length){
				subscribers[i++] =
						new LambdaSubscriber<>(onNext, onError, onComplete, onSubscribe);
			}

			((ParallelFlux<T>) publisher).subscribe(subscribers);

			return Disposables.composite(subscribers);
		}
		else {
			LambdaSubscriber<? super T> subscriber =
					new LambdaSubscriber<>(onNext, onError, onComplete, onSubscribe);

			publisher.subscribe(Operators.toCoreSubscriber(new FluxHide.SuppressFuseableSubscriber<>(subscriber)));

			return subscriber;
		}
	}
Subscribes to this {@link ParallelFlux} by providing an onNext, onError,
onComplete and onSubscribe callback and triggers the execution chain for all
'rails'.

@param onNext consumer of onNext signals
@param onError consumer of error signal
@param onComplete callback on completion signal
@param onSubscribe consumer of the subscription signal