TopicProcessor (Reactor Core 3.1.14.RELEASE)

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:

Runnable, Processor<E,E>, Publisher<E>, Subscriber<E>, CoreSubscriber<E>, Disposable, Scannable
```
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(java.lang.String, int), 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(java.lang.String, int) 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.

Author:

Stephane Maldini, Anatoly Kadyshev

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`	`TopicProcessor.Builder<T>` `TopicProcessor` 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

Method Summary

All Methods Static Methods Instance Methods Concrete Methods Deprecated Methods
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()`.
`boolean`	`awaitAndShutdown(long timeout, TimeUnit timeUnit)` Deprecated. use `awaitAndShutdown(Duration)` instead
`static <E> TopicProcessor.Builder<E>`	`builder()` Create a new `TopicProcessor` `TopicProcessor.Builder` with default properties.
`static <E> TopicProcessor<E>`	`create()` Create a new TopicProcessor using `Queues.SMALL_BUFFER_SIZE` backlog size, blockingWait Strategy and auto-cancel.
`static <E> TopicProcessor<E>`	`create(String name, int bufferSize)` Create a new TopicProcessor using the provided backlog size, with a blockingWait Strategy and auto-cancellation.
`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> TopicProcessor<E>`	`share(String name, int bufferSize)` Create a new shared TopicProcessor using the passed backlog size, with a blockingWait Strategy and auto-cancellation.
`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, 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, doOnCancel, doOnComplete, doOnEach, doOnError, doOnError, doOnError, doOnNext, doOnRequest, doOnSubscribe, doOnTerminate, elapsed, elapsed, elementAt, elementAt, empty, 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, name, never, next, ofType, onAssembly, onAssembly, onBackpressureBuffer, onBackpressureBuffer, onBackpressureBuffer, onBackpressureBuffer, onBackpressureBuffer, onBackpressureBuffer, onBackpressureBuffer, onBackpressureDrop, onBackpressureDrop, onBackpressureError, onBackpressureLatest, onErrorMap, onErrorMap, onErrorMap, onErrorResume, onErrorResume, onErrorResume, onErrorReturn, onErrorReturn, onErrorReturn, 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, 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, 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, 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, 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, operatorName, parents, scan, scanOrDefault, tags

Methods inherited from interface reactor.core.Disposable
isDisposed

- Method Detail
  - builder
```
public static <E> TopicProcessor.Builder<E> builder()
```
    Create a new TopicProcessor TopicProcessor.Builder with default properties.
    
    Returns:
    
    new TopicProcessor builder
  - create
```
public static <E> TopicProcessor<E> create()
```
    Create a new TopicProcessor 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> TopicProcessor<E> create(String name,
                                           int bufferSize)
```
    Create a new TopicProcessor using the provided backlog size, with a blockingWait Strategy and auto-cancellation.
    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:
    
    the fresh TopicProcessor instance
  - share
```
public static <E> TopicProcessor<E> share(String name,
                                          int bufferSize)
```
    Create a new shared TopicProcessor using the passed backlog size, with a blockingWait Strategy and auto-cancellation.
    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)
  - 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(Throwable t)
```
  - doComplete
```
protected void doComplete()
```
  - 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()
```
  - 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.
  - 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.
  - 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
```
@Deprecated
public final boolean awaitAndShutdown(long timeout,
                                                  TimeUnit timeUnit)
```
    Deprecated. use awaitAndShutdown(Duration) instead
    
    Block until all submitted tasks have completed, then do a normal EventLoopProcessor#dispose().
    
    Parameters:
    
    timeout - the timeout value
    
    timeUnit - the unit for timeout
    
    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
  - 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)
  - 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
  - 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
  - toString
```
public final String toString()
```
    Overrides:
    
    toString in class Flux<IN>
  - hashCode
```
public final int hashCode()
```
    Overrides:
    
    hashCode in class Object
  - 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>
  - 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.
  - 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
  - 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

Class TopicProcessor<E>

Nested Class Summary

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

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

Method Summary

Methods inherited from class reactor.core.publisher.FluxProcessor

Methods inherited from class reactor.core.publisher.Flux

Methods inherited from class java.lang.Object

Methods inherited from interface org.reactivestreams.Publisher

Methods inherited from interface reactor.core.CoreSubscriber

Methods inherited from interface reactor.core.Scannable

Methods inherited from interface reactor.core.Disposable

Method Detail

builder

create

create

share

subscribe

drain

doError

doComplete

getPending

requestTask

run

scanUnsafe

defaultName

defaultRequestTaskExecutor

alive

awaitAndShutdown

awaitAndShutdown

awaitAndShutdown

inners

forceShutdown

getAvailableCapacity

getError

toString

hashCode

isSerialized

isTerminated

onComplete

onError

onNext

onSubscribe

serializeAlways

shutdown

getBufferSize

downstreamCount