reactor.core.publisher

Class TopicProcessor<E>

java.lang.Object

reactor.core.publisher.Flux<OUT>

reactor.core.publisher.FluxProcessor<IN,IN>

reactor.core.publisher.TopicProcessor<E>

Type Parameters:

E - Type of dispatched signal

All Implemented Interfaces:

java.lang.Runnable, org.reactivestreams.Processor<E,E>, org.reactivestreams.Publisher<E>, org.reactivestreams.Subscriber<E>, MultiProducer, Receiver, Trackable

public final class TopicProcessor<E>
extends FluxProcessor<IN,IN>

An implementation of a RingBuffer backed message-passing Processor implementing publish-subscribe with async event loops.

Created from share(boolean), the TopicProcessor will authorize concurrent publishing (multi-producer) from its receiving side Subscriber.onNext(Object). Additionally, any of the TopicProcessor will stop the event loop thread if an error occurs.

The processor respects the Reactive Streams contract and must not be signalled concurrently on any onXXXX method if share(boolean) has not been used. Each subscriber will be assigned a unique thread that will only stop on terminal event: Complete, Error or Cancel. If Auto-Cancel is enabled, when all subscribers are unregistered, a cancel signal is sent to the upstream Publisher if any. Executor can be customized and will define how many concurrent subscribers are allowed (fixed thread). When a Subscriber requests Long.MAX, there won't be any backpressure applied and the producer will run at risk of being throttled if the subscribers don't catch up. With any other strictly positive demand, a subscriber will stop reading new Next signals (Complete and Error will still be read) as soon as the demand has been fully consumed by the publisher.

When more than 1 subscriber listens to that processor, they will all receive the exact same events if their respective demand is still strictly positive, very much like a Fan-Out scenario.

When the backlog has been completely booked and no subscribers is draining the signals, the publisher will start throttling. In effect the smaller the backlog size is defined, the smaller the difference in processing rate between subscribers must remain. Since the sequence for each subscriber will point to various ringBuffer locations, the processor knows when a backlog can't override the previously occupied slot.

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static class	reactor.core.publisher.EventLoopProcessor.Slot<T> A simple reusable data container.

Field Summary

Fields

Modifier and Type	Field and Description
static boolean	TRACEABLE_RING_BUFFER_PROCESSOR Whether the RingBuffer*Processor can be graphed by wrapping the individual Sequence with the target downstream

Fields inherited from interface reactor.core.Trackable

UNSPECIFIED

Method Summary

Modifier and Type	Method and Description
boolean	alive() Determine whether this Processor can be used.
boolean	awaitAndShutdown() Block until all submitted tasks have completed, then do a normal EventLoopProcessor.shutdown().
boolean	awaitAndShutdown(long timeout, java.util.concurrent.TimeUnit timeUnit) Block until all submitted tasks have completed, then do a normal EventLoopProcessor#shutdown().
static <E> TopicProcessor<E>	create() Create a new TopicProcessor using QueueSupplier.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy and auto-cancel.
static <E> TopicProcessor<E>	create(boolean autoCancel) Create a new TopicProcessor using QueueSupplier.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy and the passed auto-cancel setting.
static <E> TopicProcessor<E>	create(java.util.concurrent.ExecutorService service) Create a new TopicProcessor using QueueSupplier.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy and auto-cancel.
static <E> TopicProcessor<E>	create(java.util.concurrent.ExecutorService service, boolean autoCancel) Create a new TopicProcessor using QueueSupplier.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy and the passed auto-cancel setting.
static <E> TopicProcessor<E>	create(java.util.concurrent.ExecutorService service, int bufferSize) Create a new TopicProcessor using passed backlog size, blockingWait Strategy and will auto-cancel.
static <E> TopicProcessor<E>	create(java.util.concurrent.ExecutorService service, int bufferSize, boolean autoCancel) Create a new TopicProcessor using passed backlog size, blockingWait Strategy and the auto-cancel argument.
static <E> TopicProcessor<E>	create(java.util.concurrent.ExecutorService service, int bufferSize, WaitStrategy strategy) Create a new TopicProcessor using passed backlog size, wait strategy and will auto-cancel.
static <E> TopicProcessor<E>	create(java.util.concurrent.ExecutorService service, int bufferSize, WaitStrategy strategy, boolean autoCancel) Create a new TopicProcessor using passed backlog size, wait strategy and auto-cancel settings.
static <E> TopicProcessor<E>	create(java.lang.String name) Create a new TopicProcessor using QueueSupplier.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy and auto-cancel.
static <E> TopicProcessor<E>	create(java.lang.String name, int bufferSize) Create a new TopicProcessor using QueueSupplier.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy and the passed auto-cancel setting.
static <E> TopicProcessor<E>	create(java.lang.String name, int bufferSize, boolean autoCancel) Create a new TopicProcessor using the blockingWait Strategy, passed backlog size, and auto-cancel settings.
static <E> TopicProcessor<E>	create(java.lang.String name, int bufferSize, WaitStrategy strategy) Create a new TopicProcessor using passed backlog size, wait strategy and will auto-cancel.
static <E> TopicProcessor<E>	create(java.lang.String name, int bufferSize, WaitStrategy strategy, boolean autoCancel) Create a new TopicProcessor using passed backlog size, wait strategy and auto-cancel settings.
static <E> TopicProcessor<E>	create(java.lang.String name, int bufferSize, WaitStrategy strategy, java.util.function.Supplier<E> signalSupplier) Create a new TopicProcessor using passed backlog size, wait strategy, signal supplier.
protected void	doComplete()
protected void	doError(java.lang.Throwable t)
long	downstreamCount() the number of downstream receivers
java.util.Iterator<?>	downstreams() the connected data receivers
Flux<E>	drain() Drain is a hot replication of the current buffer delivered if supported.
Flux<IN>	forceShutdown() Shutdown this Processor, forcibly halting any work currently executing and discarding any tasks that have not yet been executed.
long	getAvailableCapacity()
long	getCapacity() Return defined element capacity
java.lang.Throwable	getError() Current error if any, default to null
long	getPending() Return current used space in buffer
int	hashCode()
boolean	isCancelled()
boolean	isStarted() Has this upstream started or "onSubscribed" ?
boolean	isTerminated() Has this upstream finished or "completed" / "failed" ?
void	onComplete()
void	onError(java.lang.Throwable t)
void	onNext(IN o)
void	onSubscribe(org.reactivestreams.Subscription s)
protected void	requestTask(org.reactivestreams.Subscription s)
void	run()
static <E> TopicProcessor<E>	share(boolean autoCancel) Create a new TopicProcessor using QueueSupplier.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy and the passed auto-cancel setting.
static <E> TopicProcessor<E>	share(java.util.concurrent.ExecutorService service) Create a new TopicProcessor using QueueSupplier.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy and auto-cancel.
static <E> TopicProcessor<E>	share(java.util.concurrent.ExecutorService service, boolean autoCancel) Create a new TopicProcessor using QueueSupplier.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy and the passed auto-cancel setting.
static <E> TopicProcessor<E>	share(java.util.concurrent.ExecutorService service, int bufferSize) Create a new TopicProcessor using passed backlog size, blockingWait Strategy and will auto-cancel.
static <E> TopicProcessor<E>	share(java.util.concurrent.ExecutorService service, int bufferSize, boolean autoCancel) Create a new TopicProcessor using passed backlog size, blockingWait Strategy and the auto-cancel argument.
static <E> TopicProcessor<E>	share(java.util.concurrent.ExecutorService service, int bufferSize, WaitStrategy strategy) Create a new TopicProcessor using passed backlog size, wait strategy and will auto-cancel.
static <E> TopicProcessor<E>	share(java.util.concurrent.ExecutorService service, int bufferSize, WaitStrategy strategy, boolean autoCancel) Create a new TopicProcessor using passed backlog size, wait strategy and auto-cancel settings.
static <E> TopicProcessor<E>	share(java.lang.String name, int bufferSize) Create a new TopicProcessor using QueueSupplier.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy and the passed auto-cancel setting.
static <E> TopicProcessor<E>	share(java.lang.String name, int bufferSize, boolean autoCancel) Create a new TopicProcessor using the blockingWait Strategy, passed backlog size, and auto-cancel settings.
static <E> TopicProcessor<E>	share(java.lang.String name, int bufferSize, java.util.function.Supplier<E> signalSupplier) Create a new TopicProcessor using passed backlog size, wait strategy and signal supplier.
static <E> TopicProcessor<E>	share(java.lang.String name, int bufferSize, WaitStrategy strategy) Create a new TopicProcessor using passed backlog size, wait strategy and will auto-cancel.
static <E> TopicProcessor<E>	share(java.lang.String name, int bufferSize, WaitStrategy strategy, boolean autoCancel) Create a new TopicProcessor using passed backlog size, wait strategy and auto-cancel settings.
static <E> TopicProcessor<E>	share(java.lang.String name, int bufferSize, WaitStrategy waitStrategy, java.util.function.Supplier<E> signalSupplier) Create a new TopicProcessor using passed backlog size, wait strategy and signal supplier.
void	shutdown() Shutdown this active Processor such that it can no longer be used.
void	subscribe(org.reactivestreams.Subscriber<? super E> subscriber)
java.lang.String	toString()
org.reactivestreams.Subscription	upstream() Return the direct source of data, Supports reference.

Methods inherited from class reactor.core.publisher.FluxProcessor

connect, connectSink, connectSink, serialize, switchOnNext, wrap

Methods inherited from class reactor.core.publisher.Flux

all, any, as, awaitOnSubscribe, blockFirst, blockFirst, blockFirstMillis, blockLast, blockLast, blockLastMillis, buffer, buffer, buffer, buffer, buffer, buffer, buffer, buffer, buffer, buffer, buffer, buffer, buffer, bufferMillis, bufferMillis, bufferMillis, bufferMillis, bufferMillis, bufferMillis, bufferMillis, cache, cache, cache, cache, cancelOn, cast, collect, collect, collectList, collectMap, collectMap, collectMap, collectMultimap, collectMultimap, collectMultimap, collectSortedList, collectSortedList, combineLatest, combineLatest, combineLatest, combineLatest, combineLatest, combineLatest, combineLatest, combineLatest, combineLatest, compose, concat, concat, concat, concat, concatDelayError, concatDelayError, concatDelayError, concatDelayError, concatMap, concatMap, concatMapDelayError, concatMapDelayError, concatMapDelayError, concatMapIterable, concatMapIterable, concatWith, count, create, create, defaultIfEmpty, defer, delay, delayMillis, delayMillis, delaySubscription, delaySubscription, delaySubscriptionMillis, delaySubscriptionMillis, dematerialize, distinct, distinct, distinctUntilChanged, distinctUntilChanged, doAfterTerminate, doOnCancel, doOnComplete, doOnError, doOnError, doOnError, doOnNext, doOnRequest, doOnSubscribe, doOnTerminate, elapsed, elapsed, elementAt, elementAt, empty, error, error, filter, firstEmitting, firstEmitting, firstEmittingWith, flatMap, flatMap, flatMap, flatMap, flatMap, flatMapIterable, flatMapIterable, from, fromArray, fromIterable, fromStream, generate, generate, generate, getPrefetch, groupBy, groupBy, groupJoin, handle, hasElement, hasElements, hide, ignoreElements, interval, interval, intervalMillis, intervalMillis, intervalMillis, intervalMillis, join, just, just, last, last, log, log, log, log, map, mapError, mapError, mapError, materialize, merge, merge, merge, merge, merge, merge, mergeWith, never, next, ofType, onAssembly, onAssembly, onBackpressureBuffer, onBackpressureBuffer, onBackpressureBuffer, onBackpressureDrop, onBackpressureDrop, onBackpressureError, onBackpressureLatest, onErrorResumeWith, onErrorResumeWith, onErrorResumeWith, onErrorReturn, onErrorReturn, onErrorReturn, onTerminateDetach, parallel, parallel, parallel, publish, publish, publish, publish, publishNext, publishOn, publishOn, range, reduce, reduce, reduceWith, repeat, repeat, repeat, repeat, repeatWhen, replay, replay, replay, replay, replayMillis, replayMillis, retry, retry, retry, retry, retryWhen, sample, sample, sampleFirst, sampleFirst, sampleFirstMillis, sampleMillis, sampleTimeout, sampleTimeout, scan, scan, scanWith, share, single, single, singleOrEmpty, skip, skip, skipLast, skipMillis, skipMillis, skipUntil, skipUntilOther, skipWhile, sort, sort, startWith, startWith, startWith, subscribe, subscribe, subscribe, subscribe, subscribe, subscribe, subscribe, subscribeOn, subscribeWith, switchIfEmpty, switchMap, switchMap, switchOnError, switchOnError, switchOnError, switchOnNext, switchOnNext, take, take, takeLast, takeMillis, takeMillis, takeUntil, takeUntilOther, takeWhile, then, then, then, thenMany, thenMany, timeout, timeout, timeout, timeout, timeout, timeoutMillis, timeoutMillis, timeoutMillis, timeoutMillis, timestamp, timestamp, toIterable, toIterable, toIterable, toStream, toStream, transform, using, using, window, window, window, window, window, window, window, window, windowMillis, windowMillis, windowMillis, windowMillis, windowMillis, withLatestFrom, zip, zip, zip, zip, zip, zip, zip, zip, zip, zip, zip, zip, zipWith, zipWith, zipWith, zipWith, zipWithIterable, zipWithIterable

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, notify, notifyAll, wait, wait, wait

Methods inherited from interface reactor.core.MultiProducer

hasDownstreams

Methods inherited from interface reactor.core.Trackable

expectedFromUpstream, limit, requestedFromDownstream

Field Detail

TRACEABLE_RING_BUFFER_PROCESSOR

public static final boolean TRACEABLE_RING_BUFFER_PROCESSOR

Whether the RingBuffer*Processor can be graphed by wrapping the individual Sequence with the target downstream

Method Detail

create

public static <E> TopicProcessor<E> create()

Create a new TopicProcessor using QueueSupplier.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy and auto-cancel.

A new Cached ThreadExecutorPool will be implicitely created.

Type Parameters:

E - Type of processed signals

Returns:

a fresh processor

create

public static <E> TopicProcessor<E> create(java.lang.String name)

Create a new TopicProcessor using QueueSupplier.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy and auto-cancel.

A new Cached ThreadExecutorPool will be implicitely created.

Type Parameters:

E - Type of processed signals

Parameters:

name - processor thread logical name

Returns:

a fresh processor

create

public static <E> TopicProcessor<E> create(boolean autoCancel)

Create a new TopicProcessor using QueueSupplier.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy and the passed auto-cancel setting.

A new Cached ThreadExecutorPool will be implicitely created.

Type Parameters:

E - Type of processed signals

Parameters:

autoCancel - Should this propagate cancellation when unregistered by all subscribers ?

Returns:

a fresh processor

create

public static <E> TopicProcessor<E> create(java.util.concurrent.ExecutorService service)

Create a new TopicProcessor using QueueSupplier.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy and auto-cancel.

The passed ExecutorService will execute as many event-loop consuming the ringbuffer as subscribers.

Type Parameters:

E - Type of processed signals

Parameters:

service - A provided ExecutorService to manage threading infrastructure

Returns:

a fresh processor

create

public static <E> TopicProcessor<E> create(java.util.concurrent.ExecutorService service,
                                           boolean autoCancel)

Create a new TopicProcessor using QueueSupplier.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy and the passed auto-cancel setting.

The passed ExecutorService will execute as many event-loop consuming the ringbuffer as subscribers.

Type Parameters:

E - Type of processed signals

Parameters:

service - A provided ExecutorService to manage threading infrastructure

autoCancel - Should this propagate cancellation when unregistered by all subscribers ?

Returns:

a fresh processor

create

public static <E> TopicProcessor<E> create(java.lang.String name,
                                           int bufferSize)

Create a new TopicProcessor using QueueSupplier.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy and the passed auto-cancel setting.

A new Cached ThreadExecutorPool will be implicitely created and will use the passed name to qualify the created threads.

Type Parameters:

E - Type of processed signals

Parameters:

name - Use a new Cached ExecutorService and assign this name to the created threads

bufferSize - A Backlog Size to mitigate slow subscribers

Returns:

the fresh TopicProcessor instance

create

public static <E> TopicProcessor<E> create(java.lang.String name,
                                           int bufferSize,
                                           boolean autoCancel)

Create a new TopicProcessor using the blockingWait Strategy, passed backlog size, and auto-cancel settings.

The passed ExecutorService will execute as many event-loop consuming the ringbuffer as subscribers.

Type Parameters:

E - Type of processed signals

Parameters:

name - Use a new Cached ExecutorService and assign this name to the created threads

bufferSize - A Backlog Size to mitigate slow subscribers

autoCancel - Should this propagate cancellation when unregistered by all subscribers ?

Returns:

a fresh processor

create

public static <E> TopicProcessor<E> create(java.util.concurrent.ExecutorService service,
                                           int bufferSize)

Create a new TopicProcessor using passed backlog size, blockingWait Strategy and will auto-cancel.

The passed ExecutorService will execute as many event-loop consuming the ringbuffer as subscribers.

Type Parameters:

E - Type of processed signals

Parameters:

service - A provided ExecutorService to manage threading infrastructure

bufferSize - A Backlog Size to mitigate slow subscribers

Returns:

a fresh processor

create

public static <E> TopicProcessor<E> create(java.util.concurrent.ExecutorService service,
                                           int bufferSize,
                                           boolean autoCancel)

Create a new TopicProcessor using passed backlog size, blockingWait Strategy and the auto-cancel argument.

The passed ExecutorService will execute as many event-loop consuming the ringbuffer as subscribers.

Type Parameters:

E - Type of processed signals

Parameters:

service - A provided ExecutorService to manage threading infrastructure

bufferSize - A Backlog Size to mitigate slow subscribers

autoCancel - Should this propagate cancellation when unregistered by all subscribers ?

Returns:

a fresh processor

create

public static <E> TopicProcessor<E> create(java.lang.String name,
                                           int bufferSize,
                                           WaitStrategy strategy)

Create a new TopicProcessor using passed backlog size, wait strategy and will auto-cancel.

A new Cached ThreadExecutorPool will be implicitely created and will use the passed name to qualify the created threads.

Type Parameters:

E - Type of processed signals

Parameters:

name - Use a new Cached ExecutorService and assign this name to the created threads

bufferSize - A Backlog Size to mitigate slow subscribers

strategy - A RingBuffer WaitStrategy to use instead of the default blocking wait strategy.

Returns:

a fresh processor

create

public static <E> TopicProcessor<E> create(java.lang.String name,
                                           int bufferSize,
                                           WaitStrategy strategy,
                                           java.util.function.Supplier<E> signalSupplier)

Create a new TopicProcessor using passed backlog size, wait strategy, signal supplier. The created processor is not shared and will auto-cancel.

A new Cached ThreadExecutorPool will be implicitely created and will use the passed name to qualify the created threads.

Type Parameters:

E - Type of processed signals

Parameters:

name - Use a new Cached ExecutorService and assign this name to the created threads

bufferSize - A Backlog Size to mitigate slow subscribers

strategy - A RingBuffer WaitStrategy to use instead of the default blocking wait strategy.

signalSupplier - A supplier of dispatched signals to preallocate in the ring buffer

Returns:

a fresh processor

create

public static <E> TopicProcessor<E> create(java.lang.String name,
                                           int bufferSize,
                                           WaitStrategy strategy,
                                           boolean autoCancel)

Create a new TopicProcessor using passed backlog size, wait strategy and auto-cancel settings.

A new Cached ThreadExecutorPool will be implicitely created and will use the passed name to qualify the created threads.

Type Parameters:

E - Type of processed signals

Parameters:

name - Use a new Cached ExecutorService and assign this name to the created threads

bufferSize - A Backlog Size to mitigate slow subscribers

strategy - A RingBuffer WaitStrategy to use instead of the default blocking wait strategy.

autoCancel - Should this propagate cancellation when unregistered by all subscribers ?

Returns:

a fresh processor

create

public static <E> TopicProcessor<E> create(java.util.concurrent.ExecutorService service,
                                           int bufferSize,
                                           WaitStrategy strategy)

Create a new TopicProcessor using passed backlog size, wait strategy and will auto-cancel.

The passed ExecutorService will execute as many event-loop consuming the ringbuffer as subscribers.

Type Parameters:

E - Type of processed signals

Parameters:

service - A provided ExecutorService to manage threading infrastructure

bufferSize - A Backlog Size to mitigate slow subscribers

strategy - A RingBuffer WaitStrategy to use instead of the default blocking wait strategy.

Returns:

a fresh processor

create

public static <E> TopicProcessor<E> create(java.util.concurrent.ExecutorService service,
                                           int bufferSize,
                                           WaitStrategy strategy,
                                           boolean autoCancel)

Create a new TopicProcessor using passed backlog size, wait strategy and auto-cancel settings.

The passed ExecutorService will execute as many event-loop consuming the ringbuffer as subscribers.

Type Parameters:

E - Type of processed signals

Parameters:

service - A provided ExecutorService to manage threading infrastructure

bufferSize - A Backlog Size to mitigate slow subscribers

strategy - A RingBuffer WaitStrategy to use instead of the default blocking wait strategy.

autoCancel - Should this propagate cancellation when unregistered by all subscribers ?

Returns:

a fresh processor

share

public static <E> TopicProcessor<E> share(boolean autoCancel)

Create a new TopicProcessor using QueueSupplier.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy and the passed auto-cancel setting.

A Shared Processor authorizes concurrent onNext calls and is suited for multi-threaded publisher that will fan-in data.

A new Cached ThreadExecutorPool will be implicitely created.

Type Parameters:

E - Type of processed signals

Parameters:

autoCancel - Should this propagate cancellation when unregistered by all subscribers ?

Returns:

a fresh processor

share

public static <E> TopicProcessor<E> share(java.util.concurrent.ExecutorService service)

Create a new TopicProcessor using QueueSupplier.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy and auto-cancel.

A Shared Processor authorizes concurrent onNext calls and is suited for multi-threaded publisher that will fan-in data.

The passed ExecutorService will execute as many event-loop consuming the ringbuffer as subscribers.

Type Parameters:

E - Type of processed signals

Parameters:

service - A provided ExecutorService to manage threading infrastructure

Returns:

a fresh processor

share

public static <E> TopicProcessor<E> share(java.util.concurrent.ExecutorService service,
                                          boolean autoCancel)

Create a new TopicProcessor using QueueSupplier.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy and the passed auto-cancel setting.

A Shared Processor authorizes concurrent onNext calls and is suited for multi-threaded publisher that will fan-in data.

The passed ExecutorService will execute as many event-loop consuming the ringbuffer as subscribers.

Type Parameters:

E - Type of processed signals

Parameters:

service - A provided ExecutorService to manage threading infrastructure

autoCancel - Should this propagate cancellation when unregistered by all subscribers ?

Returns:

a fresh processor

share

public static <E> TopicProcessor<E> share(java.lang.String name,
                                          int bufferSize)

Create a new TopicProcessor using QueueSupplier.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy and the passed auto-cancel setting.

A Shared Processor authorizes concurrent onNext calls and is suited for multi-threaded publisher that will fan-in data.

A new Cached ThreadExecutorPool will be implicitely created and will use the passed name to qualify the created threads.

Type Parameters:

E - Type of processed signals

Parameters:

name - Use a new Cached ExecutorService and assign this name to the created threads

bufferSize - A Backlog Size to mitigate slow subscribers

Returns:

a fresh processor

share

public static <E> TopicProcessor<E> share(java.lang.String name,
                                          int bufferSize,
                                          boolean autoCancel)

Create a new TopicProcessor using the blockingWait Strategy, passed backlog size, and auto-cancel settings.

A Shared Processor authorizes concurrent onNext calls and is suited for multi-threaded publisher that will fan-in data.

The passed ExecutorService will execute as many event-loop consuming the ringbuffer as subscribers.

Type Parameters:

E - Type of processed signals

Parameters:

name - Use a new Cached ExecutorService and assign this name to the created threads

bufferSize - A Backlog Size to mitigate slow subscribers

autoCancel - Should this propagate cancellation when unregistered by all subscribers ?

Returns:

a fresh processor

share

public static <E> TopicProcessor<E> share(java.util.concurrent.ExecutorService service,
                                          int bufferSize)

Create a new TopicProcessor using passed backlog size, blockingWait Strategy and will auto-cancel.

A Shared Processor authorizes concurrent onNext calls and is suited for multi-threaded publisher that will fan-in data.

The passed ExecutorService will execute as many event-loop consuming the ringbuffer as subscribers.

Type Parameters:

E - Type of processed signals

Parameters:

service - A provided ExecutorService to manage threading infrastructure

bufferSize - A Backlog Size to mitigate slow subscribers

Returns:

a fresh processor

share

public static <E> TopicProcessor<E> share(java.util.concurrent.ExecutorService service,
                                          int bufferSize,
                                          boolean autoCancel)

Create a new TopicProcessor using passed backlog size, blockingWait Strategy and the auto-cancel argument.

A Shared Processor authorizes concurrent onNext calls and is suited for multi-threaded publisher that will fan-in data.

The passed ExecutorService will execute as many event-loop consuming the ringbuffer as subscribers.

Type Parameters:

E - Type of processed signals

Parameters:

service - A provided ExecutorService to manage threading infrastructure

bufferSize - A Backlog Size to mitigate slow subscribers

autoCancel - Should this propagate cancellation when unregistered by all subscribers ?

Returns:

a fresh processor

share

public static <E> TopicProcessor<E> share(java.lang.String name,
                                          int bufferSize,
                                          WaitStrategy strategy)

Create a new TopicProcessor using passed backlog size, wait strategy and will auto-cancel.

A Shared Processor authorizes concurrent onNext calls and is suited for multi-threaded publisher that will fan-in data.

A new Cached ThreadExecutorPool will be implicitely created and will use the passed name to qualify the created threads.

Type Parameters:

E - Type of processed signals

Parameters:

name - Use a new Cached ExecutorService and assign this name to the created threads

bufferSize - A Backlog Size to mitigate slow subscribers

strategy - A RingBuffer WaitStrategy to use instead of the default blocking wait strategy.

Returns:

a fresh processor

share

public static <E> TopicProcessor<E> share(java.lang.String name,
                                          int bufferSize,
                                          java.util.function.Supplier<E> signalSupplier)

Create a new TopicProcessor using passed backlog size, wait strategy and signal supplier. The created processor will auto-cancel and is shared.

A Shared Processor authorizes concurrent onNext calls and is suited for multi-threaded publisher that will fan-in data.

A new Cached ThreadExecutorPool will be implicitely created and will use the passed name to qualify the created threads.

Type Parameters:

E - Type of processed signals

Parameters:

name - Use a new Cached ExecutorService and assign this name to the created threads

bufferSize - A Backlog Size to mitigate slow subscribers

signalSupplier - A supplier of dispatched signals to preallocate in the ring buffer

Returns:

a fresh processor

share

public static <E> TopicProcessor<E> share(java.lang.String name,
                                          int bufferSize,
                                          WaitStrategy waitStrategy,
                                          java.util.function.Supplier<E> signalSupplier)

Create a new TopicProcessor using passed backlog size, wait strategy and signal supplier. The created processor will auto-cancel and is shared.

A Shared Processor authorizes concurrent onNext calls and is suited for multi-threaded publisher that will fan-in data.

A new Cached ThreadExecutorPool will be implicitely created and will use the passed name to qualify the created threads.

Type Parameters:

E - Type of processed signals

Parameters:

name - Use a new Cached ExecutorService and assign this name to the created threads

bufferSize - A Backlog Size to mitigate slow subscribers

waitStrategy - A RingBuffer WaitStrategy to use instead of the default blocking wait strategy. buffer

signalSupplier - the supplier of signals

Returns:

a fresh processor

share

public static <E> TopicProcessor<E> share(java.lang.String name,
                                          int bufferSize,
                                          WaitStrategy strategy,
                                          boolean autoCancel)

Create a new TopicProcessor using passed backlog size, wait strategy and auto-cancel settings.

A Shared Processor authorizes concurrent onNext calls and is suited for multi-threaded publisher that will fan-in data.

A new Cached ThreadExecutorPool will be implicitely created and will use the passed name to qualify the created threads.

Type Parameters:

E - Type of processed signals

Parameters:

name - Use a new Cached ExecutorService and assign this name to the created threads

bufferSize - A Backlog Size to mitigate slow subscribers

strategy - A RingBuffer WaitStrategy to use instead of the default blocking wait strategy.

autoCancel - Should this propagate cancellation when unregistered by all subscribers ?

Returns:

a fresh processor

share

public static <E> TopicProcessor<E> share(java.util.concurrent.ExecutorService service,
                                          int bufferSize,
                                          WaitStrategy strategy)

Create a new TopicProcessor using passed backlog size, wait strategy and will auto-cancel.

A Shared Processor authorizes concurrent onNext calls and is suited for multi-threaded publisher that will fan-in data.

The passed ExecutorService will execute as many event-loop consuming the ringbuffer as subscribers.

Type Parameters:

E - Type of processed signals

Parameters:

service - A provided ExecutorService to manage threading infrastructure

bufferSize - A Backlog Size to mitigate slow subscribers

strategy - A RingBuffer WaitStrategy to use instead of the default blocking wait strategy.

Returns:

a fresh processor

share

public static <E> TopicProcessor<E> share(java.util.concurrent.ExecutorService service,
                                          int bufferSize,
                                          WaitStrategy strategy,
                                          boolean autoCancel)

Create a new TopicProcessor using passed backlog size, wait strategy and auto-cancel settings.

A Shared Processor authorizes concurrent onNext calls and is suited for multi-threaded publisher that will fan-in data.

The passed ExecutorService will execute as many event-loop consuming the ringbuffer as subscribers.

Type Parameters:

E - Type of processed signals

Parameters:

service - A provided ExecutorService to manage threading infrastructure

bufferSize - A Backlog Size to mitigate slow subscribers

strategy - A RingBuffer WaitStrategy to use instead of the default blocking wait strategy.

autoCancel - Should this propagate cancellation when unregistered by all subscribers ?

Returns:

a fresh processor

subscribe

public void subscribe(org.reactivestreams.Subscriber<? super E> subscriber)

Specified by:

subscribe in interface org.reactivestreams.Publisher<E>

Overrides:

subscribe in class FluxProcessor<E,E>

drain

public Flux<E> drain()

Drain is a hot replication of the current buffer delivered if supported. Since it is hot there might be no guarantee to see a end if the buffer keeps replenishing due to concurrent producing.

Returns:

a Flux sequence possibly unbounded of incoming buffered values or empty if not supported.

doError

protected void doError(java.lang.Throwable t)

doComplete

protected void doComplete()

getPending

public long getPending()

Description copied from interface: Trackable

Return current used space in buffer

Returns:

long capacity

requestTask

protected void requestTask(org.reactivestreams.Subscription s)

downstreamCount

public long downstreamCount()

Description copied from interface: MultiProducer

the number of downstream receivers

Returns:

the number of downstream receivers

run

public void run()

alive

public final boolean alive()

Determine whether this Processor can be used.

Returns:

true if this Resource is alive and can be used, false otherwise.

awaitAndShutdown

public final boolean awaitAndShutdown()

Block until all submitted tasks have completed, then do a normal EventLoopProcessor.shutdown().

Returns:

if the underlying executor terminated and false if the timeout elapsed before termination

awaitAndShutdown

public final boolean awaitAndShutdown(long timeout,
                                      java.util.concurrent.TimeUnit timeUnit)

Block until all submitted tasks have completed, then do a normal EventLoopProcessor#shutdown().

Parameters:

timeout - the timeout value

timeUnit - the unit for timeout

Returns:

if the underlying executor terminated and false if the timeout elapsed before termination

forceShutdown

public final Flux<IN> forceShutdown()

Shutdown this Processor, forcibly halting any work currently executing and discarding any tasks that have not yet been executed.

Returns:

a Flux instance with the remaining undelivered values

getAvailableCapacity

public long getAvailableCapacity()

Returns:

a snapshot number of available onNext before starving the resource

downstreams

public java.util.Iterator<?> downstreams()

Description copied from interface: MultiProducer

the connected data receivers

Specified by:

downstreams in interface MultiProducer

Returns:

the connected data receivers

getError

public final java.lang.Throwable getError()

Description copied from interface: Trackable

Current error if any, default to null

Specified by:

getError in interface Trackable

Returns:

Current error if any, default to null

toString

public final java.lang.String toString()

Overrides:

toString in class Flux<IN>

hashCode

public final int hashCode()

Overrides:

hashCode in class java.lang.Object

isCancelled

public final boolean isCancelled()

Specified by:

isCancelled in interface Trackable

Returns:

has the downstream "cancelled" and interrupted its consuming ?

isStarted

public final boolean isStarted()

Description copied from interface: Trackable

Has this upstream started or "onSubscribed" ?

Specified by:

isStarted in interface Trackable

Returns:

has this upstream started or "onSubscribed" ?

isTerminated

public final boolean isTerminated()

Description copied from interface: Trackable

Has this upstream finished or "completed" / "failed" ?

Specified by:

isTerminated in interface Trackable

Returns:

has this upstream finished or "completed" / "failed" ?

onComplete

public final void onComplete()

Specified by:

onComplete in interface org.reactivestreams.Subscriber<IN>

onError

public final void onError(java.lang.Throwable t)

Specified by:

onError in interface org.reactivestreams.Subscriber<IN>

onNext

public final void onNext(IN o)

Specified by:

onNext in interface org.reactivestreams.Subscriber<IN>

onSubscribe

public final void onSubscribe(org.reactivestreams.Subscription s)

Specified by:

onSubscribe in interface org.reactivestreams.Subscriber<IN>

shutdown

public final void shutdown()

Shutdown this active Processor such that it can no longer be used. If the resource carries any work, it will wait (but NOT blocking the caller) for all the remaining tasks to perform before closing the resource.

upstream

public final org.reactivestreams.Subscription upstream()

Description copied from interface: Receiver

Return the direct source of data, Supports reference.

Specified by:

upstream in interface Receiver

Returns:

the direct source of data, Supports reference.

getCapacity

public final long getCapacity()

Description copied from interface: Trackable

Return defined element capacity

Specified by:

getCapacity in interface Trackable

Overrides:

getCapacity in class FluxProcessor<IN,IN>

Returns:

long capacity