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:

Runnable, Processor<E,E>, Publisher<E>, Subscriber<E>, CoreSubscriber<E>, Disposable, Scannable

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 Flux.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.

Author:

Stephane Maldini

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static class	reactor.core.publisher.EventLoopProcessor.Slot<T> A simple reusable data container.
static class	WorkQueueProcessor.Builder<T> WorkQueueProcessor builder that can be used to create new processors.

Nested classes/interfaces inherited from interface reactor.core.Scannable

Scannable.Attr<T>

Nested classes/interfaces inherited from interface reactor.core.Disposable

Disposable.Composite, Disposable.Swap

Field Summary

Fields inherited from interface reactor.core.Scannable

OPERATOR_NAME_UNRELATED_WORDS_PATTERN

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.dispose().
boolean	awaitAndShutdown(Duration timeout) Block until all submitted tasks have completed, then do a normal EventLoopProcessor#dispose().
static <T> WorkQueueProcessor.Builder<T>	builder() Create a new WorkQueueProcessor WorkQueueProcessor.Builder with default properties.
static <E> WorkQueueProcessor<E>	create() Create a new WorkQueueProcessor using Queues.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy and auto-cancel.
static <E> WorkQueueProcessor<E>	create(String name, int bufferSize) Create a new WorkQueueProcessor using the passed buffer size, blockingWait Strategy and auto-cancel.
protected static String	defaultName(ThreadFactory threadFactory, Class<? extends reactor.core.publisher.EventLoopProcessor> clazz) A method to extract a name from the ThreadFactory if it turns out to be a Supplier (in which case the supplied value string representation is used).
protected static ExecutorService	defaultRequestTaskExecutor(String name) A method to create a suitable default ExecutorService for use in implementors requestTask(Subscription) (a cached thread pool), reusing a main name and appending [request-task] suffix.
protected void	doComplete()
protected void	doError(Throwable t)
long	downstreamCount() Return the number of active Subscriber or -1 if untracked.
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()
int	getBufferSize() Return the processor buffer capacity if any or Integer.MAX_VALUE
Throwable	getError() Current error if any, default to null
long	getPending() Return the number of parked elements in the emitter backlog.
int	hashCode()
Stream<? extends Scannable>	inners() Return a Stream of referenced inners (flatmap, multicast etc)
boolean	isSerialized() Return true if this FluxProcessor supports multithread producing
boolean	isTerminated() Has this upstream finished or "completed" / "failed" ?
void	onComplete()
void	onError(Throwable t)
void	onNext(IN o)
void	onSubscribe(Subscription s) Implementors should initialize any state used by Subscriber.onNext(Object) before calling Subscription.request(long).
protected void	requestTask(Subscription s)
void	run()
Object	scanUnsafe(Scannable.Attr key) This method is used internally by components to define their key-value mappings in a single place.
protected boolean	serializeAlways() Returns serialization strategy.
static <E> WorkQueueProcessor<E>	share(String name, int bufferSize) Create a new shared WorkQueueProcessor using the passed buffer size, blockingWait Strategy and auto-cancel.
void	shutdown() Shutdown this active Processor such that it can no longer be used.
void	subscribe(CoreSubscriber<? super E> actual) An internal Publisher.subscribe(Subscriber) that will bypass Hooks.onLastOperator(Function) pointcut.
String	toString()

Methods inherited from class reactor.core.publisher.FluxProcessor

dispose, hasCompleted, hasDownstreams, hasError, serialize, sink, sink, switchOnNext, wrap

Methods inherited from class reactor.core.publisher.Flux

all, any, as, blockFirst, blockFirst, blockLast, blockLast, buffer, buffer, buffer, buffer, buffer, buffer, buffer, buffer, buffer, buffer, buffer, bufferTimeout, bufferTimeout, bufferTimeout, bufferTimeout, bufferUntil, bufferUntil, bufferWhen, bufferWhen, bufferWhile, cache, cache, cache, cache, cache, cache, cancelOn, cast, checkpoint, checkpoint, checkpoint, 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, concatWithValues, count, create, create, defaultIfEmpty, defer, delayElements, delayElements, delaySequence, delaySequence, delaySubscription, delaySubscription, delaySubscription, delayUntil, dematerialize, distinct, distinct, distinct, distinct, distinctUntilChanged, distinctUntilChanged, distinctUntilChanged, doAfterTerminate, doFinally, doFirst, doOnCancel, doOnComplete, doOnDiscard, doOnEach, doOnError, doOnError, doOnError, doOnNext, doOnRequest, doOnSubscribe, doOnTerminate, elapsed, elapsed, elementAt, elementAt, empty, error, error, error, expand, expand, expandDeep, expandDeep, filter, filterWhen, filterWhen, first, first, flatMap, flatMap, flatMap, flatMap, flatMapDelayError, flatMapIterable, flatMapIterable, flatMapSequential, flatMapSequential, flatMapSequential, flatMapSequentialDelayError, from, fromArray, fromIterable, fromStream, fromStream, generate, generate, generate, getPrefetch, groupBy, groupBy, groupBy, groupBy, groupJoin, handle, hasElement, hasElements, hide, ignoreElements, index, index, interval, interval, interval, interval, join, just, just, last, last, limitRate, limitRate, limitRequest, log, log, log, log, log, log, map, materialize, merge, merge, merge, merge, merge, merge, mergeDelayError, mergeOrdered, mergeOrdered, mergeOrdered, mergeOrderedWith, mergeSequential, mergeSequential, mergeSequential, mergeSequential, mergeSequential, mergeSequential, mergeSequentialDelayError, mergeSequentialDelayError, mergeSequentialDelayError, mergeWith, metrics, name, never, next, ofType, onAssembly, onAssembly, onBackpressureBuffer, onBackpressureBuffer, onBackpressureBuffer, onBackpressureBuffer, onBackpressureBuffer, onBackpressureBuffer, onBackpressureBuffer, onBackpressureDrop, onBackpressureDrop, onBackpressureError, onBackpressureLatest, onErrorContinue, onErrorContinue, onErrorContinue, onErrorMap, onErrorMap, onErrorMap, onErrorResume, onErrorResume, onErrorResume, onErrorReturn, onErrorReturn, onErrorReturn, onErrorStop, onLastAssembly, onTerminateDetach, or, parallel, parallel, parallel, publish, publish, publish, publish, publishNext, publishOn, publishOn, publishOn, push, push, range, reduce, reduce, reduceWith, repeat, repeat, repeat, repeat, repeatWhen, replay, replay, replay, replay, replay, replay, retry, retry, retry, retry, retryBackoff, retryBackoff, retryBackoff, retryBackoff, retryBackoff, retryWhen, sample, sample, sampleFirst, sampleFirst, sampleTimeout, sampleTimeout, scan, scan, scanWith, share, single, single, singleOrEmpty, skip, skip, skip, skipLast, skipUntil, skipUntilOther, skipWhile, sort, sort, startWith, startWith, startWith, subscribe, subscribe, subscribe, subscribe, subscribe, subscribe, subscribeOn, subscribeOn, subscriberContext, subscriberContext, subscribeWith, switchIfEmpty, switchMap, switchMap, switchOnFirst, switchOnNext, switchOnNext, tag, take, take, take, takeLast, takeUntil, takeUntilOther, takeWhile, then, then, thenEmpty, thenMany, timeout, timeout, timeout, timeout, timeout, timeout, timeout, timestamp, timestamp, toIterable, toIterable, toIterable, toStream, toStream, transform, using, using, usingWhen, usingWhen, usingWhen, window, window, window, window, window, window, window, windowTimeout, windowTimeout, windowUntil, windowUntil, windowUntil, windowWhen, windowWhile, windowWhile, withLatestFrom, zip, 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 org.reactivestreams.Publisher

subscribe

Methods inherited from interface reactor.core.CoreSubscriber

currentContext

Methods inherited from interface reactor.core.Scannable

actuals, from, isScanAvailable, name, parents, scan, scanOrDefault, stepName, steps, tags

Methods inherited from interface reactor.core.Disposable

isDisposed

Method Detail

builder

public static final <T> WorkQueueProcessor.Builder<T> builder()

Create a new WorkQueueProcessor WorkQueueProcessor.Builder with default properties.

Returns:

new WorkQueueProcessor builder

create

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

Create a new WorkQueueProcessor using Queues.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy and auto-cancel.

A new Cached ThreadExecutorPool will be implicitly created.

Type Parameters:

E - Type of processed signals

Returns:

a fresh processor

create

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

Create a new WorkQueueProcessor using the passed buffer size, blockingWait Strategy and auto-cancel.

A new Cached ThreadExecutorPool will be implicitly 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

doComplete

protected void doComplete()

doError

protected void doError(Throwable t)

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.

getPending

public long getPending()

Return the number of parked elements in the emitter backlog.

Returns:

the number of parked elements in the emitter backlog.

requestTask

protected void requestTask(Subscription s)

run

public void run()

share

public static <E> WorkQueueProcessor<E> share(String name,
                                              int bufferSize)

Create a new shared 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 implicitly 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

subscribe

public void subscribe(CoreSubscriber<? super E> actual)

Description copied from class: Flux

An internal Publisher.subscribe(Subscriber) that will bypass Hooks.onLastOperator(Function) pointcut.

In addition to behave as expected by Publisher.subscribe(Subscriber) in a controlled manner, it supports direct subscribe-time Context passing.

Specified by:

subscribe in class Flux<E>

Parameters:

actual - the Subscriber interested into the published sequence

See Also:

Flux.subscribe(Subscriber)

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.dispose().

Returns:

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

awaitAndShutdown

public final boolean awaitAndShutdown(Duration timeout)

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

Parameters:

timeout - the timeout value as a Duration. Note this is converted to a Long of nanoseconds (which amounts to roughly 292 years maximum timeout).

Returns:

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

defaultName

protected static String defaultName(@Nullable
                                    ThreadFactory threadFactory,
                                    Class<? extends reactor.core.publisher.EventLoopProcessor> clazz)

A method to extract a name from the ThreadFactory if it turns out to be a Supplier (in which case the supplied value string representation is used). Otherwise return the current class's simpleName.

Parameters:

threadFactory - the factory to test for a supplied name

clazz -

Returns:

the name to use in thread pools

defaultRequestTaskExecutor

protected static ExecutorService defaultRequestTaskExecutor(String name)

A method to create a suitable default ExecutorService for use in implementors requestTask(Subscription) (a cached thread pool), reusing a main name and appending [request-task] suffix.

Parameters:

name - the main thread name used by the processor.

Returns:

a default ExecutorService for requestTask.

downstreamCount

public long downstreamCount()

Description copied from class: FluxProcessor

Return the number of active Subscriber or -1 if untracked.

Overrides:

downstreamCount in class FluxProcessor<IN,IN>

Returns:

the number of active Subscriber or -1 if untracked

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 final long getAvailableCapacity()

Returns:

a snapshot number of available onNext before starving the resource

getBufferSize

public final int getBufferSize()

Description copied from class: FluxProcessor

Return the processor buffer capacity if any or Integer.MAX_VALUE

Overrides:

getBufferSize in class FluxProcessor<IN,IN>

Returns:

processor buffer capacity if any or Integer.MAX_VALUE

getError

@Nullable
public final Throwable getError()

Description copied from class: FluxProcessor

Current error if any, default to null

Overrides:

getError in class FluxProcessor<IN,IN>

Returns:

Current error if any, default to null

hashCode

public final int hashCode()

Overrides:

hashCode in class Object

inners

public Stream<? extends Scannable> inners()

Description copied from interface: Scannable

Return a Stream of referenced inners (flatmap, multicast etc)

Specified by:

inners in interface Scannable

Overrides:

inners in class FluxProcessor<IN,IN>

Returns:

a Stream of referenced inners (flatmap, multicast etc)

isSerialized

public boolean isSerialized()

Description copied from class: FluxProcessor

Return true if this FluxProcessor supports multithread producing

Overrides:

isSerialized in class FluxProcessor<IN,IN>

Returns:

true if this FluxProcessor supports multithread producing

isTerminated

public final boolean isTerminated()

Description copied from class: FluxProcessor

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

Overrides:

isTerminated in class FluxProcessor<IN,IN>

Returns:

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

onComplete

public final void onComplete()

Specified by:

onComplete in interface Subscriber<IN>

onError

public final void onError(Throwable t)

Specified by:

onError in interface Subscriber<IN>

onNext

public final void onNext(IN o)

Specified by:

onNext in interface Subscriber<IN>

onSubscribe

public final void onSubscribe(Subscription s)

Description copied from interface: CoreSubscriber

Implementors should initialize any state used by Subscriber.onNext(Object) before calling Subscription.request(long). Should further onNext related state modification occur, thread-safety will be required.

Note that an invalid request <= 0 will not produce an onError and will simply be ignored or reported through a debug-enabled Logger.

Specified by:

onSubscribe in interface Subscriber<IN>

Specified by:

onSubscribe in interface CoreSubscriber<IN>

scanUnsafe

@Nullable
public Object scanUnsafe(Scannable.Attr key)

Description copied from interface: Scannable

This method is used internally by components to define their key-value mappings in a single place. Although it is ignoring the generic type of the Scannable.Attr key, implementors should take care to return values of the correct type, and return null if no specific value is available.

For public consumption of attributes, prefer using Scannable.scan(Attr), which will return a typed value and fall back to the key's default if the component didn't define any mapping.

Specified by:

scanUnsafe in interface Scannable

Overrides:

scanUnsafe in class FluxProcessor<IN,IN>

Parameters:

key - a Scannable.Attr to resolve for the component.

Returns:

the value associated to the key for that specific component, or null if none.

serializeAlways

protected boolean serializeAlways()

Description copied from class: FluxProcessor

Returns serialization strategy. If true, FluxProcessor.sink() will always be serialized. Otherwise sink is serialized only if FluxSink.onRequest(java.util.function.LongConsumer) is invoked.

Overrides:

serializeAlways in class FluxProcessor<IN,IN>

Returns:

true to serialize any sink, false to delay serialization till onRequest

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.

toString

public final String toString()

Overrides:

toString in class Flux<IN>