reactor.core.publisher

Class WorkQueueProcessor<E>

java.lang.Object

reactor.core.publisher.Flux<OUT>

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

reactor.core.publisher.WorkQueueProcessor<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 WorkQueueProcessor<E>
extends FluxProcessor<IN,IN>

An implementation of a RingBuffer backed message-passing Processor implementing work-queue distribution with async event loops.

Created from share(), the WorkQueueProcessor will authorize concurrent publishing (multi-producer) from its receiving side Subscriber.onNext(Object). WorkQueueProcessor is able to replay up to its buffer size number of failed signals (either dropped or fatally throwing on child Subscriber.onNext(T)).

The processor is very similar to TopicProcessor but only partially respects the Reactive Streams contract.

The purpose of this processor is to distribute the signals to only one of the subscribed subscribers and to share the demand amongst all subscribers. The scenario is akin to Executor or Round-Robin distribution. However there is no guarantee the distribution will be respecting a round-robin distribution all the time.

The core use for this component is to scale up easily without suffering the overhead of an Executor and without using dedicated queues by subscriber, which is less used memory, less GC, more win.

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> WorkQueueProcessor<E>	create() Create a new WorkQueueProcessor using QueueSupplier.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy and auto-cancel.
static <E> WorkQueueProcessor<E>	create(boolean autoCancel) Create a new WorkQueueProcessor using QueueSupplier.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy and the passed auto-cancel setting.
static <E> WorkQueueProcessor<E>	create(java.util.concurrent.ExecutorService service) Create a new WorkQueueProcessor using QueueSupplier.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy and auto-cancel.
static <E> WorkQueueProcessor<E>	create(java.util.concurrent.ExecutorService service, boolean autoCancel) Create a new WorkQueueProcessor using QueueSupplier.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy and the passed auto-cancel setting.
static <E> WorkQueueProcessor<E>	create(java.util.concurrent.ExecutorService service, int bufferSize) Create a new TopicProcessor using the passed buffer size, blockingWait Strategy and auto-cancel.
static <E> WorkQueueProcessor<E>	create(java.util.concurrent.ExecutorService service, int bufferSize, boolean autoCancel) Create a new WorkQueueProcessor using the passed buffer size, blockingWait Strategy and auto-cancel.
static <E> WorkQueueProcessor<E>	create(java.util.concurrent.ExecutorService executor, int bufferSize, WaitStrategy strategy) Create a new WorkQueueProcessor using the passed buffer size and blockingWait Strategy settings but will auto-cancel.
static <E> WorkQueueProcessor<E>	create(java.util.concurrent.ExecutorService executor, int bufferSize, WaitStrategy strategy, boolean autoCancel) Create a new WorkQueueProcessor using the passed buffer size, wait strategy and auto-cancel settings.
static <E> WorkQueueProcessor<E>	create(java.lang.String name) Create a new TopicProcessor using the default buffer size 32, blockingWait Strategy and auto-cancel.
static <E> WorkQueueProcessor<E>	create(java.lang.String name, int bufferSize) Create a new TopicProcessor using the passed buffer size, blockingWait Strategy and auto-cancel.
static <E> WorkQueueProcessor<E>	create(java.lang.String name, int bufferSize, boolean autoCancel) Create a new TopicProcessor using the passed buffer size, blockingWait Strategy and the passed auto-cancel setting.
static <E> WorkQueueProcessor<E>	create(java.lang.String name, int bufferSize, WaitStrategy strategy) Create a new WorkQueueProcessor using the passed buffer size, blockingWait Strategy and auto-cancel.
static <E> WorkQueueProcessor<E>	create(java.lang.String name, int bufferSize, WaitStrategy strategy, boolean autoCancel) Create a new WorkQueueProcessor using the passed buffer size, blockingWait Strategy and auto-cancel settings.
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> WorkQueueProcessor<E>	share() Create a new WorkQueueProcessor using QueueSupplier.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy and auto-cancel.
static <E> WorkQueueProcessor<E>	share(boolean autoCancel) Create a new WorkQueueProcessor using QueueSupplier.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy and the passed auto-cancel setting.
static <E> WorkQueueProcessor<E>	share(java.util.concurrent.ExecutorService service) Create a new WorkQueueProcessor using QueueSupplier.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy and auto-cancel.
static <E> WorkQueueProcessor<E>	share(java.util.concurrent.ExecutorService service, boolean autoCancel) Create a new WorkQueueProcessor using QueueSupplier.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy and the passed auto-cancel setting.
static <E> WorkQueueProcessor<E>	share(java.util.concurrent.ExecutorService service, int bufferSize) Create a new TopicProcessor using the passed buffer size, blockingWait Strategy and auto-cancel.
static <E> WorkQueueProcessor<E>	share(java.util.concurrent.ExecutorService service, int bufferSize, boolean autoCancel) Create a new WorkQueueProcessor using the passed buffer size, blockingWait Strategy and auto-cancel.
static <E> WorkQueueProcessor<E>	share(java.util.concurrent.ExecutorService executor, int bufferSize, WaitStrategy strategy) Create a new WorkQueueProcessor using the passed buffer size and blockingWait Strategy settings but will auto-cancel.
static <E> WorkQueueProcessor<E>	share(java.util.concurrent.ExecutorService executor, int bufferSize, WaitStrategy strategy, boolean autoCancel) Create a new WorkQueueProcessor using the passed buffer size, wait strategy and auto-cancel settings.
static <E> WorkQueueProcessor<E>	share(java.lang.String name, int bufferSize) Create a new TopicProcessor using the passed buffer size, blockingWait Strategy and auto-cancel.
static <E> WorkQueueProcessor<E>	share(java.lang.String name, int bufferSize, boolean autoCancel) Create a new TopicProcessor using the passed buffer size, blockingWait Strategy and the passed auto-cancel setting.
static <E> WorkQueueProcessor<E>	share(java.lang.String name, int bufferSize, WaitStrategy strategy) Create a new WorkQueueProcessor using the passed buffer size, blockingWait Strategy and auto-cancel.
static <E> WorkQueueProcessor<E>	share(java.lang.String name, int bufferSize, WaitStrategy strategy, boolean autoCancel) Create a new WorkQueueProcessor using the passed buffer size, blockingWait Strategy and auto-cancel settings.
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, 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, bufferMillis, bufferMillis, bufferMillis, bufferMillis, bufferMillis, bufferMillis, cache, cache, 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, elementAt, elementAtOrDefault, empty, error, error, every, everyFirst, 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, log, log, log, map, mapError, mapError, mapError, materialize, merge, merge, merge, merge, merge, merge, mergeWith, never, next, onAssembly, onAssembly, onBackpressureBuffer, onBackpressureDrop, onBackpressureDrop, onBackpressureError, onBackpressureLatest, onErrorResumeWith, onErrorResumeWith, onErrorResumeWith, onErrorReturn, onErrorReturn, onErrorReturn, onTerminateDetach, parallel, parallel, parallel, process, process, process, process, publish, publish, publish, publish, publishNext, publishOn, publishOn, range, reduce, reduce, reduceWith, repeat, repeat, repeat, repeat, repeatWhen, replay, replay, retry, retry, retry, retry, retryWhen, sample, sample, sampleFirst, sampleFirst, sampleFirstMillis, sampleMillis, sampleTimeout, sampleTimeout, scan, scan, scanWith, single, singleOrDefault, singleOrEmpty, skip, skip, skipLast, skipMillis, skipMillis, skipUntil, skipUntilOther, skipWhile, 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, toIterable, toIterable, toIterable, toStream, toStream, transform, using, using, window, window, window, window, window, window, window, window, windowMillis, 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> WorkQueueProcessor<E> create()

Create a new WorkQueueProcessor 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> WorkQueueProcessor<E> create(boolean autoCancel)

Create a new WorkQueueProcessor 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> WorkQueueProcessor<E> create(java.util.concurrent.ExecutorService service)

Create a new WorkQueueProcessor 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> WorkQueueProcessor<E> create(java.util.concurrent.ExecutorService service,
                                               boolean autoCancel)

Create a new WorkQueueProcessor 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> WorkQueueProcessor<E> create(java.lang.String name)

Create a new TopicProcessor using the default buffer size 32, blockingWait Strategy and 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

Returns:

a fresh processor

create

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

Create a new TopicProcessor using the passed buffer size, blockingWait Strategy and 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

Returns:

a fresh processor

create

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

Create a new TopicProcessor using the passed buffer 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

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

Returns:

a fresh processor

create

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

Create a new TopicProcessor using the passed buffer 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

bufferSize - A Backlog Size to mitigate slow subscribers

Returns:

a fresh processor

create

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

Create a new WorkQueueProcessor using the passed buffer 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

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> WorkQueueProcessor<E> create(java.lang.String name,
                                               int bufferSize,
                                               WaitStrategy strategy)

Create a new WorkQueueProcessor using the passed buffer size, blockingWait Strategy and 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 smart blocking wait strategy.

Returns:

a fresh processor

create

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

Create a new WorkQueueProcessor using the passed buffer size, blockingWait 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 smart blocking wait strategy.

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

Returns:

a fresh processor

create

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

Create a new WorkQueueProcessor using the passed buffer size and blockingWait Strategy settings but 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:

executor - 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 smart blocking wait strategy.

Returns:

a fresh processor

create

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

Create a new WorkQueueProcessor using the passed buffer 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:

executor - 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 smart blocking wait strategy.

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

Returns:

a fresh processor

share

public static <E> WorkQueueProcessor<E> share()

Create a new WorkQueueProcessor 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.

A new Cached ThreadExecutorPool will be implicitely created.

Type Parameters:

E - Type of processed signals

Returns:

a fresh processor

share

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

Create a new WorkQueueProcessor 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> WorkQueueProcessor<E> share(java.util.concurrent.ExecutorService service)

Create a new WorkQueueProcessor 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

share

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

Create a new WorkQueueProcessor 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> WorkQueueProcessor<E> share(java.lang.String name,
                                              int bufferSize)

Create a new TopicProcessor using the passed buffer 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.

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> WorkQueueProcessor<E> share(java.lang.String name,
                                              int bufferSize,
                                              boolean autoCancel)

Create a new TopicProcessor using the passed buffer 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

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

Returns:

a fresh processor

share

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

Create a new TopicProcessor using the passed buffer 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

bufferSize - A Backlog Size to mitigate slow subscribers

Returns:

a fresh processor

share

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

Create a new WorkQueueProcessor using the passed buffer 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

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> WorkQueueProcessor<E> share(java.lang.String name,
                                              int bufferSize,
                                              WaitStrategy strategy)

Create a new WorkQueueProcessor using the passed buffer 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.

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 smart blocking wait strategy.

Returns:

a fresh processor

share

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

Create a new WorkQueueProcessor using the passed buffer size, blockingWait 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 smart blocking wait strategy.

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

Returns:

a fresh processor

share

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

Create a new WorkQueueProcessor using the passed buffer size and blockingWait Strategy settings but 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:

executor - 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 smart blocking wait strategy.

Returns:

a fresh processor

share

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

Create a new WorkQueueProcessor using the passed buffer 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:

executor - 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 smart 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()

requestTask

protected void requestTask(org.reactivestreams.Subscription s)

getPending

public long getPending()

Description copied from interface: Trackable

Return current used space in buffer

Returns:

long capacity

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