https://github.com/reactor/reactor-core: Non-Blocking Reactive Foundation for the JVM (reactor)
  • Stephane Maldini
  • David Karnok
  • Simon Baslé
  • Rajini Sivaram
  • Sébastien Deleuze 
/*
 * 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
 *
 *       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.time.Duration;
import java.util.function.Consumer;

import reactor.core.Disposable;
import reactor.core.Fuseable;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;


The abstract base class for connectable publishers that let subscribers pile up
before they connect to their data source.

Type parameters:
  • <T> – the input and output value type
See Also:
/**
 * The abstract base class for connectable publishers that let subscribers pile up
 * before they connect to their data source.
 *
 * @see #publish
 * @see <a href="https://github.com/reactor/reactive-streams-commons">Reactive-Streams-Commons</a>
 * @param <T> the input and output value type
 */
public abstract class ConnectableFlux<T> extends Flux<T> {

	
Connects this ConnectableFlux to the upstream source when the first Subscriber subscribes. 



Returns:a Flux that connects to the upstream source when the first Subscriber subscribes
/**
	 * Connects this {@link ConnectableFlux} to the upstream source when the first {@link org.reactivestreams.Subscriber}
	 * subscribes.
	 *
	 * <p>
	 * <img class="marble" src="https://raw.githubusercontent.com/reactor/reactor-core/v3.1.3.RELEASE/src/docs/marble/autoconnect.png" alt="">
	 *
	 * @return a {@link Flux} that connects to the upstream source when the first {@link org.reactivestreams.Subscriber} subscribes
	 */
	public final Flux<T> autoConnect() {
		return autoConnect(1);
	}

	
Connects this ConnectableFlux to the upstream source when the specified amount of Subscriber subscribes. 

Subscribing and immediately unsubscribing still contributes to the counter that
triggers the connection.




Params:
  • minSubscribers – the minimum number of subscribers
Returns:a Flux that connects to the upstream source when the given amount of Subscribers subscribed
/**
	 * Connects this {@link ConnectableFlux} to the upstream source when the specified amount of
	 * {@link org.reactivestreams.Subscriber} subscribes.
	 * <p>
	 * Subscribing and immediately unsubscribing still contributes to the counter that
	 * triggers the connection.
	 *
	 * <p>
	 * <img class="marble" src="https://raw.githubusercontent.com/reactor/reactor-core/v3.1.3.RELEASE/src/docs/marble/autoconnect.png" alt="">
	 *
	 * @param minSubscribers the minimum number of subscribers
	 *
	 * @return a {@link Flux} that connects to the upstream source when the given amount of Subscribers subscribed
	 */
	public final Flux<T> autoConnect(int minSubscribers) {
		return autoConnect(minSubscribers, NOOP_DISCONNECT);
	}

	
Connects this ConnectableFlux to the upstream source when the specified amount of Subscriber subscribes and calls the supplied consumer with a runnable that allows disconnecting. 
Params:
  • minSubscribers – the minimum number of subscribers
  • cancelSupport – the consumer that will receive the Disposable that allows disconnecting 
    

Returns:a Flux that connects to the upstream source when the given amount of subscribers subscribed
/**
	 * Connects this {@link ConnectableFlux} to the upstream source when the specified amount of
	 * {@link org.reactivestreams.Subscriber} subscribes and calls the supplied consumer with a runnable that allows disconnecting.
	 * @param minSubscribers the minimum number of subscribers
	 * @param cancelSupport the consumer that will receive the {@link Disposable} that allows disconnecting
	 *
	 * <p>
	 * <img class="marble" src="https://raw.githubusercontent.com/reactor/reactor-core/v3.1.3.RELEASE/src/docs/marble/autoconnect.png" alt="">
	 *
	 * @return a {@link Flux} that connects to the upstream source when the given amount of subscribers subscribed
	 */
	public final Flux<T> autoConnect(int minSubscribers, Consumer<? super Disposable> cancelSupport) {
		if (minSubscribers == 0) {
			connect(cancelSupport);
			return this;
		}
		if(this instanceof Fuseable){
			return onAssembly(new FluxAutoConnectFuseable<>(this,
					minSubscribers,
					cancelSupport));
		}
		return onAssembly(new FluxAutoConnect<>(this, minSubscribers,
				cancelSupport));
	}

	
Connect this ConnectableFlux to its source and return a Runnable that can be used for disconnecting. 
Returns:the Disposable that allows disconnecting the connection after.
/**
	 * Connect this {@link ConnectableFlux} to its source and return a {@link Runnable} that
	 * can be used for disconnecting.
	 * @return the {@link Disposable} that allows disconnecting the connection after.
	 */
	public final Disposable connect() {
		final Disposable[] out = { null };
		connect(r -> out[0] = r);
		return out[0];
	}

	
Connects this ConnectableFlux to its source and sends a Disposable to a callback that can be used for disconnecting. 
The call should be idempotent in respect of connecting the first
and subsequent times. In addition the disconnection should be also tied
to a particular connection (so two different connection can't disconnect the other).

Params:
  • cancelSupport – the callback is called with a Disposable instance that can
be called to disconnect the source, even synchronously.
/**
	 * Connects this {@link ConnectableFlux} to its source and sends a {@link Disposable} to a callback that
	 * can be used for disconnecting.
	 *
	 * <p>The call should be idempotent in respect of connecting the first
	 * and subsequent times. In addition the disconnection should be also tied
	 * to a particular connection (so two different connection can't disconnect the other).
	 *
	 * @param cancelSupport the callback is called with a Disposable instance that can
	 * be called to disconnect the source, even synchronously.
	 */
	public abstract void connect(Consumer<? super Disposable> cancelSupport);

	
Connects to the upstream source when the first Subscriber subscribes and disconnects when all Subscribers cancelled or the upstream source completed. 



Returns:a reference counting Flux
/**
	 * Connects to the upstream source when the first {@link org.reactivestreams.Subscriber} subscribes and disconnects
	 * when all Subscribers cancelled or the upstream source completed.
	 *
	 * <p>
	 * <img class="marble" src="https://raw.githubusercontent.com/reactor/reactor-core/v3.1.3.RELEASE/src/docs/marble/refCount.png" alt="">
	 *
	 * @return a reference counting {@link Flux}
	 */
	public final Flux<T> refCount() {
		return refCount(1);
	}

	
Connects to the upstream source when the given number of Subscriber subscribes and disconnects when all Subscribers cancelled or the upstream source completed. 



Params:
  • minSubscribers – the number of subscribers expected to subscribe before connection
Returns:a reference counting Flux
/**
	 * Connects to the upstream source when the given number of {@link org.reactivestreams.Subscriber} subscribes and disconnects
	 * when all Subscribers cancelled or the upstream source completed.
	 *
	 * <p>
	 * <img class="marble" src="https://raw.githubusercontent.com/reactor/reactor-core/v3.1.3.RELEASE/src/docs/marble/refCount.png" alt="">
	 *
	 * @param minSubscribers the number of subscribers expected to subscribe before connection
	 *
	 * @return a reference counting {@link Flux}
	 */
	public final Flux<T> refCount(int minSubscribers) {
		return onAssembly(new FluxRefCount<>(this, minSubscribers));
	}

	
Connects to the upstream source when the given number of Subscriber subscribes. Disconnection can happen in two scenarios: when the upstream source completes (or errors) then there is an immediate disconnection. However, when all subscribers have cancelled, a deferred disconnection is scheduled. If any new subscriber comes in during the gracePeriod that follows, the disconnection is cancelled. 



Params:
  • minSubscribers – the number of subscribers expected to subscribe before connection
  • gracePeriod – the Duration for which to wait for new subscribers before actually disconnecting when all subscribers have cancelled.
Returns:a reference counting Flux with a grace period for disconnection
/**
	 * Connects to the upstream source when the given number of {@link org.reactivestreams.Subscriber} subscribes.
	 * Disconnection can happen in two scenarios: when the upstream source completes (or errors) then
	 * there is an immediate disconnection. However, when all subscribers have cancelled,
	 * a <strong>deferred</strong> disconnection is scheduled. If any new subscriber comes
	 * in during the {@code gracePeriod} that follows, the disconnection is cancelled.
	 *
	 * <p>
	 * <img class="marble" src="https://raw.githubusercontent.com/reactor/reactor-core/v3.1.3.RELEASE/src/docs/marble/refCount.png" alt="">
	 *
	 * @param minSubscribers the number of subscribers expected to subscribe before connection
	 * @param gracePeriod the {@link Duration} for which to wait for new subscribers before actually
	 * disconnecting when all subscribers have cancelled.
	 *
	 * @return a reference counting {@link Flux} with a grace period for disconnection
	 */
	public final Flux<T> refCount(int minSubscribers, Duration gracePeriod) {
		return refCount(minSubscribers, gracePeriod, Schedulers.parallel());
	}

	
Connects to the upstream source when the given number of Subscriber subscribes. Disconnection can happen in two scenarios: when the upstream source completes (or errors) then there is an immediate disconnection. However, when all subscribers have cancelled, a deferred disconnection is scheduled. If any new subscriber comes in during the gracePeriod that follows, the disconnection is cancelled. 



Params:
  • minSubscribers – the number of subscribers expected to subscribe before connection
  • gracePeriod – the Duration for which to wait for new subscribers before actually disconnecting when all subscribers have cancelled.
  • scheduler – the Scheduler on which to run timeouts
Returns:a reference counting Flux with a grace period for disconnection
/**
	 * Connects to the upstream source when the given number of {@link org.reactivestreams.Subscriber} subscribes.
	 * Disconnection can happen in two scenarios: when the upstream source completes (or errors) then
	 * there is an immediate disconnection. However, when all subscribers have cancelled,
	 * a <strong>deferred</strong> disconnection is scheduled. If any new subscriber comes
	 * in during the {@code gracePeriod} that follows, the disconnection is cancelled.
	 *
	 * <p>
	 * <img class="marble" src="https://raw.githubusercontent.com/reactor/reactor-core/v3.1.3.RELEASE/src/docs/marble/refCount.png" alt="">
	 *
	 * @param minSubscribers the number of subscribers expected to subscribe before connection
	 * @param gracePeriod the {@link Duration} for which to wait for new subscribers before actually
	 * disconnecting when all subscribers have cancelled.
	 * @param scheduler the {@link Scheduler} on which to run timeouts
	 *
	 * @return a reference counting {@link Flux} with a grace period for disconnection
	 */
	public final Flux<T> refCount(int minSubscribers, Duration gracePeriod, Scheduler scheduler) {
		return onAssembly(new FluxRefCountGrace<>(this, minSubscribers, gracePeriod, scheduler));
	}

	static final Consumer<Disposable> NOOP_DISCONNECT = runnable -> {

	};
}