org.springframework.integration.util

Class IntegrationReactiveUtils

java.lang.Object

org.springframework.integration.util.IntegrationReactiveUtils

public final class IntegrationReactiveUtils
extends Object

Utilities for adapting integration components to/from reactive types.

Since:

5.3

Author:

Artem Bilan

Field Summary

Fields

Modifier and Type	Field and Description
static java.time.Duration	DEFAULT_DELAY_WHEN_EMPTY A default delay before repeating an empty source Mono as 1 second Duration.
static String	DELAY_WHEN_EMPTY_KEY The subscriber context entry for Flux.delayElements(java.time.Duration) from the Mono.repeatWhenEmpty(java.util.function.Function).

Method Summary

Modifier and Type	Method and Description
static <T> reactor.core.publisher.Flux<Message<T>>	messageChannelToFlux(MessageChannel messageChannel) Adapt a provided MessageChannel into a Flux source: - a FluxMessageChannel is returned as is because it is already a Publisher; - a SubscribableChannel is subscribed with a MessageHandler for the Sinks.Many#tryEmitNext(Object) which is returned from this method; - a PollableChannel is wrapped into a MessageSource lambda and reuses messageSourceToFlux(MessageSource).
static <T> reactor.core.publisher.Flux<Message<T>>	messageSourceToFlux(MessageSource<T> messageSource) Wrap a provided MessageSource into a Flux for pulling the on demand.

Methods inherited from class java.lang.Object

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

Field Detail

DELAY_WHEN_EMPTY_KEY

public static final String DELAY_WHEN_EMPTY_KEY

The subscriber context entry for Flux.delayElements(java.time.Duration) from the Mono.repeatWhenEmpty(java.util.function.Function).

See Also:

Constant Field Values

DEFAULT_DELAY_WHEN_EMPTY

public static final java.time.Duration DEFAULT_DELAY_WHEN_EMPTY

A default delay before repeating an empty source Mono as 1 second Duration.

Method Detail

messageSourceToFlux

public static <T> reactor.core.publisher.Flux<Message<T>> messageSourceToFlux(MessageSource<T> messageSource)

Wrap a provided MessageSource into a Flux for pulling the on demand. When MessageSource.receive() returns null, the source Mono goes to the Mono.repeatWhenEmpty(java.util.function.Function<reactor.core.publisher.Flux<java.lang.Long>, ? extends org.reactivestreams.Publisher<?>>) state and performs a delay based on the DELAY_WHEN_EMPTY_KEY Duration entry in the subscriber context or falls back to 1 second duration. If a produced message has an IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK header it is ack'ed in the Mono.doOnSuccess(java.util.function.Consumer<? super T>) and nack'ed in the Mono.doOnError(java.util.function.Consumer<? super java.lang.Throwable>).

Type Parameters:

T - the expected payload type.

Parameters:

messageSource - the MessageSource to adapt.

Returns:

a Flux which pulls messages from the MessageSource on demand.

messageChannelToFlux

public static <T> reactor.core.publisher.Flux<Message<T>> messageChannelToFlux(MessageChannel messageChannel)

Adapt a provided MessageChannel into a Flux source: - a FluxMessageChannel is returned as is because it is already a Publisher; - a SubscribableChannel is subscribed with a MessageHandler for the Sinks.Many#tryEmitNext(Object) which is returned from this method; - a PollableChannel is wrapped into a MessageSource lambda and reuses messageSourceToFlux(MessageSource).

Type Parameters:

T - the expected payload type.

Parameters:

messageChannel - the MessageChannel to adapt.

Returns:

a Flux which uses a provided MessageChannel as a source for events to publish.