Package reactor.core.publisher

Class Hooks

java.lang.Object

reactor.core.publisher.Hooks

public abstract class Hooks
extends Object

A set of overridable lifecycle hooks that can be used for cross-cutting added behavior on Flux/Mono operators.

Method Summary

Modifier and Type	Method	Description
static <T, P extends Publisher<T>> @Nullable Publisher<T>	addCallSiteInfo(@Nullable P publisher, String callSite)	Deprecated. Should only be used by the instrumentation, DOES NOT guarantee any compatibility
static void	addQueueWrapper(String key, Function<Queue<?>,Queue<?>> decorator)	Adds a wrapper for every Queue used in Reactor.
static <T, P extends Publisher<T>> @Nullable Publisher<T>	addReturnInfo(@Nullable P publisher, String method)	Deprecated. Should only be used by the instrumentation, DOES NOT guarantee any compatibility
static <T> Flux<T>	convertToFluxBypassingHooks(Publisher<T> publisher)	Utility method to convert a Publisher to a Flux without applying Hooks.
static <T> Mono<T>	convertToMonoBypassingHooks(Publisher<T> publisher, boolean enforceMonoContract)	Utility method to convert a Publisher to a Mono without applying Hooks.
static void	disableAutomaticContextPropagation()	Globally disables automatic context propagation to ThreadLocals.
static void	disableContextLossTracking()	Globally disables the Context loss detection that was previously enabled by enableContextLossTracking().
static void	enableAutomaticContextPropagation()	Globally enables automatic context propagation to ThreadLocals.
static void	enableContextLossTracking()	Globally enables the Context loss detection in operators like Flux.transform(java.util.function.Function<? super reactor.core.publisher.Flux<T>, ? extends org.reactivestreams.Publisher<V>>) or Mono.transformDeferred(java.util.function.Function<? super reactor.core.publisher.Mono<T>, ? extends org.reactivestreams.Publisher<V>>) when non-Reactor types are used.
static boolean	isAutomaticContextPropagationEnabled()
static void	onEachOperator(String key, Function<? super Publisher<Object>,? extends Publisher<Object>> onEachOperator)	Add or replace a named Publisher operator interceptor for each operator created (Flux or Mono).
static void	onEachOperator(Function<? super Publisher<Object>,? extends Publisher<Object>> onEachOperator)	Add a Publisher operator interceptor for each operator created (Flux or Mono).
static void	onErrorDropped(Consumer<? super Throwable> c)	Override global error dropped strategy which by default bubble back the error.
static void	onLastOperator(String key, Function<? super Publisher<Object>,? extends Publisher<Object>> onLastOperator)	Add or replace a named Publisher operator interceptor for the last operator created in every flow (Flux or Mono).
static void	onLastOperator(Function<? super Publisher<Object>,? extends Publisher<Object>> onLastOperator)	Add a Publisher operator interceptor for the last operator created in every flow (Flux or Mono).
static void	onNextDropped(Consumer<Object> c)	Override global data dropped strategy which by default logs at DEBUG level.
static void	onNextDroppedFail()	Resets onNextDropped hook(s) and apply a strategy of throwing Exceptions.failWithCancel() instead.
static void	onNextError(BiFunction<? super Throwable,Object,? extends Throwable> onNextError)	Set the custom global error mode hook for operators that support resuming during an error in their Subscriber.onNext(Object).
static void	onOperatorDebug()	Enable operator stack recorder that captures a declaration stack whenever an operator is instantiated.
static void	onOperatorError(String key, BiFunction<? super Throwable,@Nullable Object,? extends Throwable> onOperatorError)	Add or replace a named custom error mapping, overriding the default one.
static void	onOperatorError(BiFunction<? super Throwable,@Nullable Object,? extends Throwable> onOperatorError)	Add a custom error mapping, overriding the default one.
static void	removeQueueWrapper(String key)	Removes existing Queue wrapper by key.
static void	removeQueueWrappers()	Remove all queue wrappers.
static void	resetOnEachOperator()	Reset global "assembly" hook tracking
static void	resetOnEachOperator(String key)	Remove the sub-hook with key key from the onEachOperator hook.
static void	resetOnErrorDropped()	Reset global error dropped strategy to bubbling back the error.
static void	resetOnLastOperator()	Reset global "subscriber" hook tracking
static void	resetOnLastOperator(String key)	Remove the sub-hook with key key from the onLastOperator hook.
static void	resetOnNextDropped()	Reset global data dropped strategy to throwing via Exceptions.failWithCancel()
static void	resetOnNextError()	Reset global onNext error handling strategy to terminating the sequence with an onError and cancelling upstream (OnNextFailureStrategy.STOP).
static void	resetOnOperatorDebug()	Reset global operator debug.
static void	resetOnOperatorError()	Reset global operator error mapping to the default behavior.
static void	resetOnOperatorError(String key)	Remove the sub-hook with key key from the onOperatorError hook.
static <T> Queue<T>	wrapQueue(Queue<T> queue)	Applies the Queue wrappers that were previously registered.

Methods inherited from class java.lang.Object

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

Method Details

convertToFluxBypassingHooks

public static <T> Flux<T> convertToFluxBypassingHooks(Publisher<T> publisher)

Utility method to convert a Publisher to a Flux without applying Hooks.

Type Parameters:

T - the type of data emitted by the Publisher

Parameters:

publisher - the Publisher to convert to a Flux

Returns:

the Publisher wrapped as a Flux, or the original if it was a Flux

convertToMonoBypassingHooks

public static <T> Mono<T> convertToMonoBypassingHooks(Publisher<T> publisher,
 boolean enforceMonoContract)

Utility method to convert a Publisher to a Mono without applying Hooks. Can optionally perform a "direct" (or unsafe) conversion when the caller is certain the Publisher has Mono semantics.

Type Parameters:

T - the type of data emitted by the Publisher

Parameters:

publisher - the Publisher to convert to a Mono

enforceMonoContract - true to ensure Mono semantics (by cancelling on first onNext if source isn't already a Mono), false to perform a direct conversion (see Mono.fromDirect(Publisher)).

Returns:

the Publisher wrapped as a Mono, or the original if it was a Mono

onEachOperator

public static void onEachOperator(Function<? super Publisher<Object>,? extends Publisher<Object>> onEachOperator)

Add a Publisher operator interceptor for each operator created (Flux or Mono). The passed function is applied to the original operator Publisher and can return a different Publisher, on the condition that it generically maintains the same data type as the original. Use of the Flux/Mono APIs is discouraged as it will recursively call this hook, leading to StackOverflowError.

Note that sub-hooks are cumulative, but invoking this method twice with the same instance (or any instance that has the same `toString`) will result in only a single instance being applied. See onEachOperator(String, Function) for a variant that allows you to name the sub-hooks (and thus replace them or remove them individually later on). Can be fully reset via resetOnEachOperator().

This pointcut function cannot make use of Flux, Mono or ParallelFlux APIs as it would lead to a recursive call to the hook: the operator calls would effectively invoke onEachOperator from onEachOperator. See convertToFluxBypassingHooks(Publisher) and convertToMonoBypassingHooks(Publisher, boolean).

Parameters:

onEachOperator - the sub-hook: a function to intercept each operation call (e.g. map (fn) and map(fn2) in flux.map(fn).map(fn2).subscribe())

See Also:

• onEachOperator(String, Function)
• resetOnEachOperator(String)
• resetOnEachOperator()
• onLastOperator(Function)

onEachOperator

public static void onEachOperator(String key,
 Function<? super Publisher<Object>,? extends Publisher<Object>> onEachOperator)

Add or replace a named Publisher operator interceptor for each operator created (Flux or Mono). The passed function is applied to the original operator Publisher and can return a different Publisher, on the condition that it generically maintains the same data type as the original. Use of the Flux/Mono APIs is discouraged as it will recursively call this hook, leading to StackOverflowError.

Note that sub-hooks are cumulative. Invoking this method twice with the same key will replace the old sub-hook with that name, but keep the execution order (eg. A-h1, B-h2, A-h3 will keep A-B execution order, leading to hooks h3 then h2 being executed). Removing a particular key using resetOnEachOperator(String) then adding it back will result in the execution order changing (the later sub-hook being executed last). Can be fully reset via resetOnEachOperator().

This pointcut function cannot make use of Flux, Mono or ParallelFlux APIs as it would lead to a recursive call to the hook: the operator calls would effectively invoke onEachOperator from onEachOperator. See convertToFluxBypassingHooks(Publisher) and convertToMonoBypassingHooks(Publisher, boolean).

Parameters:

key - the key for the sub-hook to add/replace

onEachOperator - the sub-hook: a function to intercept each operation call (e.g. map (fn) and map(fn2) in flux.map(fn).map(fn2).subscribe())

See Also:

• onEachOperator(Function)
• resetOnEachOperator(String)
• resetOnEachOperator()
• onLastOperator(String, Function)

resetOnEachOperator

public static void resetOnEachOperator(String key)

Remove the sub-hook with key key from the onEachOperator hook. No-op if no such key has been registered, and equivalent to calling resetOnEachOperator() if it was the last sub-hook.

Parameters:

key - the key of the sub-hook to remove

resetOnEachOperator

public static void resetOnEachOperator()

Reset global "assembly" hook tracking

onErrorDropped

public static void onErrorDropped(Consumer<? super Throwable> c)

Override global error dropped strategy which by default bubble back the error.

The hook is cumulative, so calling this method several times will set up the hook for as many consumer invocations (even if called with the same consumer instance).

Parameters:

c - the Consumer to apply to dropped errors

onLastOperator

public static void onLastOperator(Function<? super Publisher<Object>,? extends Publisher<Object>> onLastOperator)

Add a Publisher operator interceptor for the last operator created in every flow (Flux or Mono). The passed function is applied to the original operator Publisher and can return a different Publisher, on the condition that it generically maintains the same data type as the original.

Note that sub-hooks are cumulative, but invoking this method twice with the same instance (or any instance that has the same `toString`) will result in only a single instance being applied. See onLastOperator(String, Function) for a variant that allows you to name the sub-hooks (and thus replace them or remove them individually later on). Can be fully reset via resetOnLastOperator().

This pointcut function cannot make use of Flux, Mono or ParallelFlux APIs as it would lead to a recursive call to the hook: the operator calls would effectively invoke onEachOperator from onEachOperator. See convertToFluxBypassingHooks(Publisher) and convertToMonoBypassingHooks(Publisher, boolean).

Parameters:

onLastOperator - the sub-hook: a function to intercept last operation call (e.g. map(fn2) in flux.map(fn).map(fn2).subscribe())

See Also:

• onLastOperator(String, Function)
• resetOnLastOperator(String)
• resetOnLastOperator()
• onEachOperator(Function)

onLastOperator

public static void onLastOperator(String key,
 Function<? super Publisher<Object>,? extends Publisher<Object>> onLastOperator)

Add or replace a named Publisher operator interceptor for the last operator created in every flow (Flux or Mono). The passed function is applied to the original operator Publisher and can return a different Publisher, on the condition that it generically maintains the same data type as the original. Use of the Flux/Mono APIs is discouraged as it will recursively call this hook, leading to StackOverflowError.

Note that sub-hooks are cumulative. Invoking this method twice with the same key will replace the old sub-hook with that name, but keep the execution order (eg. A-h1, B-h2, A-h3 will keep A-B execution order, leading to hooks h3 then h2 being executed). Removing a particular key using resetOnLastOperator(String) then adding it back will result in the execution order changing (the later sub-hook being executed last). Can be fully reset via resetOnLastOperator().

This pointcut function cannot make use of Flux, Mono or ParallelFlux APIs as it would lead to a recursive call to the hook: the operator calls would effectively invoke onEachOperator from onEachOperator. See convertToFluxBypassingHooks(Publisher) and convertToMonoBypassingHooks(Publisher, boolean).

Parameters:

key - the key for the sub-hook to add/replace

onLastOperator - the sub-hook: a function to intercept last operation call (e.g. map(fn2) in flux.map(fn).map(fn2).subscribe())

See Also:

• onLastOperator(Function)
• resetOnLastOperator(String)
• resetOnLastOperator()
• onEachOperator(String, Function)

resetOnLastOperator

public static void resetOnLastOperator(String key)

Remove the sub-hook with key key from the onLastOperator hook. No-op if no such key has been registered, and equivalent to calling resetOnLastOperator() if it was the last sub-hook.

Parameters:

key - the key of the sub-hook to remove

resetOnLastOperator

public static void resetOnLastOperator()

Reset global "subscriber" hook tracking

onNextDropped

public static void onNextDropped(Consumer<Object> c)

Override global data dropped strategy which by default logs at DEBUG level.

The hook is cumulative, so calling this method several times will set up the hook for as many consumer invocations (even if called with the same consumer instance).

Parameters:

c - the Consumer to apply to data (onNext) that is dropped

See Also:

• onNextDroppedFail()

onNextDroppedFail

public static void onNextDroppedFail()

Resets onNextDropped hook(s) and apply a strategy of throwing Exceptions.failWithCancel() instead.

Use resetOnNextDropped() to reset to the default strategy of logging.

onOperatorDebug

public static void onOperatorDebug()

Enable operator stack recorder that captures a declaration stack whenever an operator is instantiated. When errors are observed later on, they will be enriched with a Suppressed Exception detailing the original assembly line stack. Must be called before producers (e.g. Flux.map, Mono.fromCallable) are actually called to intercept the right stack information.

This is added as a specifically-keyed sub-hook in onEachOperator(String, Function).

resetOnOperatorDebug

public static void resetOnOperatorDebug()

Reset global operator debug.

onNextError

public static void onNextError(BiFunction<? super Throwable,Object,? extends Throwable> onNextError)

Set the custom global error mode hook for operators that support resuming during an error in their Subscriber.onNext(Object).

The hook is a BiFunction of Throwable and potentially null Object. If it is also a BiPredicate, its test method should be used to determine if an error should be processed (matching predicate) or completely skipped (non-matching predicate). Typical usage, as in Operators, is to check if the predicate matches and fallback to Operators.onOperatorError(Throwable, Context) if it doesn't.

Parameters:

onNextError - the new BiFunction to use.

onOperatorError

public static void onOperatorError(BiFunction<? super Throwable,@Nullable Object,? extends Throwable> onOperatorError)

Add a custom error mapping, overriding the default one. Custom mapping can be an accumulation of several sub-hooks each subsequently added via this method.

Note that sub-hooks are cumulative, but invoking this method twice with the same instance (or any instance that has the same `toString`) will result in only a single instance being applied. See onOperatorError(String, BiFunction) for a variant that allows you to name the sub-hooks (and thus replace them or remove them individually later on). Can be fully reset via resetOnOperatorError().

For reference, the default mapping is to unwrap the exception and, if the second parameter is another exception, to add it to the first as suppressed.

Parameters:

onOperatorError - an operator error BiFunction mapper, returning an arbitrary exception given the failure and optionally some original context (data or error).

See Also:

• onOperatorError(String, BiFunction)
• resetOnOperatorError(String)
• resetOnOperatorError()

onOperatorError

public static void onOperatorError(String key,
 BiFunction<? super Throwable,@Nullable Object,? extends Throwable> onOperatorError)

Add or replace a named custom error mapping, overriding the default one. Custom mapping can be an accumulation of several sub-hooks each subsequently added via this method.

Note that invoking this method twice with the same key will replace the old sub-hook with that name, but keep the execution order (eg. A-h1, B-h2, A-h3 will keep A-B execution order, leading to hooks h3 then h2 being executed). Removing a particular key using resetOnOperatorError(String) then adding it back will result in the execution order changing (the later sub-hook being executed last). Can be fully reset via resetOnOperatorError().

For reference, the default mapping is to unwrap the exception and, if the second parameter is another exception, to add it to the first as a suppressed.

Parameters:

key - the key for the sub-hook to add/replace

onOperatorError - an operator error BiFunction mapper, returning an arbitrary exception given the failure and optionally some original context (data or error).

See Also:

• onOperatorError(String, BiFunction)
• resetOnOperatorError(String)
• resetOnOperatorError()

resetOnOperatorError

public static void resetOnOperatorError(String key)

Remove the sub-hook with key key from the onOperatorError hook. No-op if no such key has been registered, and equivalent to calling resetOnOperatorError() if it was the last sub-hook.

Parameters:

key - the key of the sub-hook to remove

resetOnOperatorError

public static void resetOnOperatorError()

Reset global operator error mapping to the default behavior.

For reference, the default mapping is to unwrap the exception and, if the second parameter is another exception, to add it to the first as a suppressed.

resetOnErrorDropped

public static void resetOnErrorDropped()

Reset global error dropped strategy to bubbling back the error.

resetOnNextDropped

public static void resetOnNextDropped()

Reset global data dropped strategy to throwing via Exceptions.failWithCancel()

resetOnNextError

public static void resetOnNextError()

Reset global onNext error handling strategy to terminating the sequence with an onError and cancelling upstream (OnNextFailureStrategy.STOP).

enableContextLossTracking

public static void enableContextLossTracking()

Globally enables the Context loss detection in operators like Flux.transform(java.util.function.Function<? super reactor.core.publisher.Flux<T>, ? extends org.reactivestreams.Publisher<V>>) or Mono.transformDeferred(java.util.function.Function<? super reactor.core.publisher.Mono<T>, ? extends org.reactivestreams.Publisher<V>>) when non-Reactor types are used. An exception will be thrown upon applying the transformation if the original Context isn't reachable (ie. it has been replaced by a totally different Context, or no Context at all)

disableContextLossTracking

public static void disableContextLossTracking()

Globally disables the Context loss detection that was previously enabled by enableContextLossTracking().

enableAutomaticContextPropagation

public static void enableAutomaticContextPropagation()

Globally enables automatic context propagation to ThreadLocals.

It requires the context-propagation library to be on the classpath to have an effect. Using the implicit global ContextRegistry it reads entries present in the modified Context using Flux.contextWrite(ContextView) (or Mono.contextWrite(ContextView)) and Flux.contextWrite(Function) (or Mono.contextWrite(Function)) and restores all ThreadLocals associated via same keys for which ThreadLocalAccessors are registered.

The ThreadLocals are present in the upstream operators from the contextWrite(...) call and the unmodified (downstream) Context is used when signals are delivered downstream, making the contextWrite(...) a logical boundary for the context propagation mechanism.

This mechanism automatically performs Flux.contextCapture() and Mono.contextCapture() in Flux.blockFirst(), Flux.blockLast(), Flux.toIterable(), and Mono.block() (and their overloads).

Since:

3.5.3

disableAutomaticContextPropagation

public static void disableAutomaticContextPropagation()

Globally disables automatic context propagation to ThreadLocals.

See Also:

• enableAutomaticContextPropagation()

isAutomaticContextPropagationEnabled

public static boolean isAutomaticContextPropagationEnabled()

Returns:

true if automatic context propagation to ThreadLocals is enabled, false otherwise

Since:

3.7.12

See Also:

• enableAutomaticContextPropagation()
• disableAutomaticContextPropagation()

addReturnInfo

@Deprecated
public static <T,
P extends Publisher<T>>
@Nullable Publisher<T> addReturnInfo(@Nullable P publisher,
 String method)

Deprecated.

Should only be used by the instrumentation, DOES NOT guarantee any compatibility

addCallSiteInfo

@Deprecated
public static <T,
P extends Publisher<T>>
@Nullable Publisher<T> addCallSiteInfo(@Nullable P publisher,
 String callSite)

Deprecated.

Should only be used by the instrumentation, DOES NOT guarantee any compatibility

addQueueWrapper

public static void addQueueWrapper(String key,
 Function<Queue<?>,Queue<?>> decorator)

Adds a wrapper for every Queue used in Reactor. Note that it won't affect existing instances of Queue. Hint: one may use AbstractQueue to reduce the number of methods to implement.

See Also:

• removeQueueWrapper(String)

Implementation Note:

the resulting Queue MUST NOT change Queue's behavior. Only side effects are allowed.

removeQueueWrapper

public static void removeQueueWrapper(String key)

Removes existing Queue wrapper by key.

See Also:

• addQueueWrapper(String, Function)

removeQueueWrappers

public static void removeQueueWrappers()

Remove all queue wrappers.

See Also:

• addQueueWrapper(String, Function)
• removeQueueWrapper(String)

wrapQueue

public static <T> Queue<T> wrapQueue(Queue<T> queue)

Applies the Queue wrappers that were previously registered. SHOULD NOT change the behavior of the provided Queue.

Parameters:

queue - the Queue to wrap.

Returns:

the result of applying Queue wrappers registered with addQueueWrapper(String, Function).

See Also:

• addQueueWrapper(String, Function)
• removeQueueWrapper(String)