reactor.core.publisher

Class MonoProcessor<O>

java.lang.Object

reactor.core.publisher.Mono<O>

reactor.core.publisher.MonoProcessor<O>

Type Parameters:

O - the type of the value that will be made available

All Implemented Interfaces:

LongSupplier, Processor<O,O>, Publisher<O>, Subscriber<O>, Subscription, CoreSubscriber<O>, Disposable, Scannable

public final class MonoProcessor<O>
extends Mono<O>
implements Processor<O,O>, CoreSubscriber<O>, Disposable, Subscription, Scannable, LongSupplier

A MonoProcessor is a Mono extension that implements stateful semantics. Multi-subscribe is allowed.

Once a MonoProcessor has been resolved, newer subscribers will benefit from the cached result.

Author:

Stephane Maldini

Nested Class Summary

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

Disposable.Composite, Disposable.Swap

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

Scannable.Attr<T>

Method Summary

Modifier and Type	Method and Description
O	block() Block the calling thread indefinitely, waiting for the completion of this MonoProcessor.
O	block(Duration timeout) Block the calling thread for the specified time, waiting for the completion of this MonoProcessor.
void	cancel()
static <T> MonoProcessor<T>	create() Create a MonoProcessor that will eagerly request 1 on onSubscribe(Subscription), cache and emit the eventual result for 1 or N subscribers.
static <T> MonoProcessor<T>	create(WaitStrategy waitStrategy) Create a MonoProcessor that will eagerly request 1 on onSubscribe(Subscription), cache and emit the eventual result for 1 or N subscribers.
Context	currentContext() Request a Context from dependent components which can include downstream operators during subscribing or a terminal Subscriber.
void	dispose() Cancel or dispose the underlying task or resource.
long	downstreamCount() Return the number of active Subscriber or -1 if untracked.
long	getAsLong() Deprecated. Should not use (to be removed in 3.2)
Throwable	getError() Return the produced Throwable error if any or null
boolean	hasDownstreams() Return true if any Subscriber is actively subscribed
Stream<? extends Scannable>	inners() Return a Stream of referenced inners (flatmap, multicast etc)
boolean	isCancelled() Indicates whether this MonoProcessor has been interrupted via cancellation.
boolean	isDisposed() Optionally return true when the resource or task is disposed.
boolean	isError() Indicates whether this MonoProcessor has been completed with an error.
boolean	isSuccess() Indicates whether this MonoProcessor has been successfully completed a value.
boolean	isTerminated() Indicates whether this MonoProcessor has been terminated by the source producer with a success or an error.
void	onComplete()
void	onError(Throwable cause)
void	onNext(O value)
void	onSubscribe(Subscription subscription) Implementors should initialize any state used by Subscriber.onNext(Object) before calling Subscription.request(long).
O	peek() Returns the value that completed this MonoProcessor.
void	request(long n)
Object	scanUnsafe(Scannable.Attr key) This method is used internally by components to define their key-value mappings in a single place.
void	subscribe(CoreSubscriber<? super O> actual) An internal Publisher.subscribe(Subscriber) that will bypass Hooks.onLastOperator(Function) pointcut.

Methods inherited from class reactor.core.publisher.Mono

and, as, blockOptional, blockOptional, cache, cache, cancelOn, cast, checkpoint, checkpoint, checkpoint, compose, concatWith, create, defaultIfEmpty, defer, delay, delay, delayElement, delayElement, delaySubscription, delaySubscription, delaySubscription, delayUntil, dematerialize, doAfterSuccessOrError, doAfterTerminate, doFinally, doOnCancel, doOnEach, doOnError, doOnError, doOnError, doOnNext, doOnRequest, doOnSubscribe, doOnSuccess, doOnSuccessOrError, doOnTerminate, elapsed, elapsed, empty, error, expand, expand, expandDeep, expandDeep, filter, filterWhen, first, first, flatMap, flatMapIterable, flatMapMany, flatMapMany, flux, from, fromCallable, fromCompletionStage, fromDirect, fromFuture, fromRunnable, fromSupplier, handle, hasElement, hide, ignoreElement, ignoreElements, just, justOrEmpty, justOrEmpty, log, log, log, log, log, log, map, materialize, mergeWith, name, never, ofType, onAssembly, onErrorMap, onErrorMap, onErrorMap, onErrorResume, onErrorResume, onErrorResume, onErrorReturn, onErrorReturn, onErrorReturn, onLastAssembly, onTerminateDetach, or, publish, publishOn, repeat, repeat, repeat, repeat, repeatWhen, repeatWhenEmpty, repeatWhenEmpty, retry, retry, retry, retry, retryWhen, sequenceEqual, sequenceEqual, sequenceEqual, single, subscribe, subscribe, subscribe, subscribe, subscribe, subscribe, subscribeOn, subscriberContext, subscriberContext, subscriberContext, subscribeWith, switchIfEmpty, tag, take, take, takeUntilOther, then, then, thenEmpty, thenMany, thenReturn, timeout, timeout, timeout, timeout, timeout, timeout, timestamp, timestamp, toFuture, toProcessor, toString, transform, using, using, when, when, whenDelayError, whenDelayError, zip, zip, zip, zip, zip, zip, zip, zip, zipDelayError, zipDelayError, zipDelayError, zipDelayError, zipDelayError, zipDelayError, zipDelayError, zipWhen, zipWhen, zipWith, zipWith

Methods inherited from class java.lang.Object

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

Methods inherited from interface org.reactivestreams.Publisher

subscribe

Methods inherited from interface reactor.core.Scannable

actuals, from, isScanAvailable, name, operatorName, parents, scan, scanOrDefault, tags

Method Detail

create

public static <T> MonoProcessor<T> create()

Create a MonoProcessor that will eagerly request 1 on onSubscribe(Subscription), cache and emit the eventual result for 1 or N subscribers.

Type Parameters:

T - type of the expected value

Returns:

A MonoProcessor.

create

public static <T> MonoProcessor<T> create(WaitStrategy waitStrategy)

Create a MonoProcessor that will eagerly request 1 on onSubscribe(Subscription), cache and emit the eventual result for 1 or N subscribers.

Type Parameters:

T - type of the expected value

Parameters:

waitStrategy - a WaitStrategy for blocking block() strategy

Returns:

A MonoProcessor.

cancel

public final void cancel()

Specified by:

cancel in interface Subscription

dispose

public void dispose()

Description copied from interface: Disposable

Cancel or dispose the underlying task or resource.

Implementations are required to make this method idempotent.

Specified by:

dispose in interface Disposable

block

@Nullable
public O block()

Block the calling thread indefinitely, waiting for the completion of this MonoProcessor. If the MonoProcessor is completed with an error a RuntimeException that wraps the error is thrown.

Overrides:

block in class Mono<O>

Returns:

the value of this MonoProcessor

block

@Nullable
public O block(Duration timeout)

Block the calling thread for the specified time, waiting for the completion of this MonoProcessor. If the MonoProcessor is completed with an error a RuntimeException that wraps the error is thrown.

Overrides:

block in class Mono<O>

Parameters:

timeout - the timeout value as a Duration

Returns:

the value of this MonoProcessor or null if the timeout is reached and the MonoProcessor has not completed

getAsLong

public long getAsLong()

Deprecated. Should not use (to be removed in 3.2)

Returns the internal state as a long fulfilled.

• -1 : cancelled
• 0 : ready
• 2 : subscribed
• 3 : terminated with value
• 4 : terminated without value
• 5 : terminated with error

Specified by:

getAsLong in interface LongSupplier

Returns:

the internal state from -1 Cancelled to 5 errored, beyond 3 included is fulfilled.

getError

@Nullable
public final Throwable getError()

Return the produced Throwable error if any or null

Returns:

the produced Throwable error if any or null

isCancelled

public boolean isCancelled()

Indicates whether this MonoProcessor has been interrupted via cancellation.

Returns:

true if this MonoProcessor is cancelled, false otherwise.

isError

public final boolean isError()

Indicates whether this MonoProcessor has been completed with an error.

Returns:

true if this MonoProcessor was completed with an error, false otherwise.

isSuccess

public final boolean isSuccess()

Indicates whether this MonoProcessor has been successfully completed a value.

Returns:

true if this MonoProcessor is successful, false otherwise.

isTerminated

public final boolean isTerminated()

Indicates whether this MonoProcessor has been terminated by the source producer with a success or an error.

Returns:

true if this MonoProcessor is successful, false otherwise.

isDisposed

public boolean isDisposed()

Description copied from interface: Disposable

Optionally return true when the resource or task is disposed.

Implementations are not required to track disposition and as such may never return true even when disposed. However, they MUST only return true when there's a guarantee the resource or task is disposed.

Specified by:

isDisposed in interface Disposable

Returns:

true when there's a guarantee the resource or task is disposed.

onComplete

public final void onComplete()

Specified by:

onComplete in interface Subscriber<O>

onError

public final void onError(Throwable cause)

Specified by:

onError in interface Subscriber<O>

onNext

public final void onNext(@Nullable
                         O value)

Specified by:

onNext in interface Subscriber<O>

onSubscribe

public final void onSubscribe(Subscription subscription)

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<O>

Specified by:

onSubscribe in interface CoreSubscriber<O>

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

Returns:

a Stream of referenced inners (flatmap, multicast etc)

peek

@Nullable
public O peek()

Returns the value that completed this MonoProcessor. Returns null if the MonoProcessor has not been completed. If the MonoProcessor is completed with an error a RuntimeException that wraps the error is thrown.

Returns:

the value that completed the MonoProcessor, or null if it has not been completed

Throws:

RuntimeException - if the MonoProcessor was completed with an error

request

public final void request(long n)

Specified by:

request in interface Subscription

subscribe

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

Description copied from class: Mono

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 Mono<O>

Parameters:

actual - the Subscriber interested into the published sequence

See Also:

Publisher.subscribe(Subscriber)

currentContext

public Context currentContext()

Description copied from interface: CoreSubscriber

Request a Context from dependent components which can include downstream operators during subscribing or a terminal Subscriber.

Specified by:

currentContext in interface CoreSubscriber<O>

Returns:

a resolved context or Context.empty()

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

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.

downstreamCount

public final long downstreamCount()

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

Returns:

the number of active Subscriber or -1 if untracked

hasDownstreams

public final boolean hasDownstreams()

Return true if any Subscriber is actively subscribed

Returns:

true if any Subscriber is actively subscribed