Package 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, Fardan An

  • Field Details

    • 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:

    • DEFAULT_DELAY_WHEN_EMPTY

      public static final Duration DEFAULT_DELAY_WHEN_EMPTY
      A default delay before repeating an empty source Mono as 1 second Duration.

    • isContextPropagationPresent

      public static final boolean isContextPropagationPresent
      The indicator that io.micrometer:context-propagation library is on classpath.
      Since:
      6.2.5

  • Method Details

    • captureReactorContext

      public static reactor.util.context.ContextView captureReactorContext()
      Capture a Reactor ContextView from the current thread local state according to the ContextSnapshotFactory logic. If no io.micrometer:context-propagation library is on classpath, the Context.empty() is returned.
      Returns:
      the Reactor ContextView from the current thread local state or Context.empty().
      Since:
      6.2.5

    • setThreadLocalsFromReactorContext

      public static @Nullable AutoCloseable setThreadLocalsFromReactorContext(reactor.util.context.ContextView context)
      Populate thread local variables from the provided Reactor ContextView according to the ContextSnapshotFactory logic.
      Parameters:
      context - the Reactor ContextView to populate from.
      Returns:
      the ContextSnapshot.Scope as a AutoCloseable to not pollute the target classpath. Can be cast if necessary. Or null if there is no io.micrometer:context-propagation library is on classpath.
      Since:
      6.2.5

    • 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 Flux.doOnNext(java.util.function.Consumer<? super T>) when this flux emits to its subscriber, and nack'ed in the Mono.doOnError(java.util.function.Consumer<? super java.lang.Throwable>).

      Cancellation while MessageSource.receive() is in flight (the sink is already CANCELLED) is handled by Reactor's discard hook on this flux. Only a PollableChannel adapted through messageChannelToFlux(MessageChannel) is best-effort re-queued via non-blocking MessageChannel.send(Message, long) (interceptor chain is re-entered; FIFO is not preserved if the queue is not empty; a full bounded channel is logged and the message is not nack'd). A direct messageSourceToFlux(() -> channel.receive(0)) lambda is not re-queued. Other sources are left unacknowledged so the source's own redelivery (broker requeue, uncommitted offset) can recover them. The discard hook only sees drops at or upstream of this flux; operators a caller adds afterward (ReactiveStreamsConsumer.setReactiveCustomizer, a flatMap on the ReactiveMessageHandler path) use a different context and are not rescued here.

      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 IntegrationReactiveUtils.PollableChannelMessageSource so cancelled in-flight receives can be re-queued, 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.