Package io.keldysh.messaging

Class MessageFlow<T>

java.lang.Object

io.keldysh.messaging.MessageFlow<T>

All Implemented Interfaces:

MessageSource<T>, Supplier<MessageFlow<T>>, org.reactivestreams.Publisher<Message<T>>

public abstract class MessageFlow<T>
extends Object
implements MessageSource<T>

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
static interface	MessageFlow.Handler<T,U>

Field Summary

Fields

Modifier and Type	Field	Description
static final reactor.util.retry.Retry	DEFAULT_RETRY_STRATEGY

Constructor Summary

Constructors

Constructor	Description
MessageFlow()

Method Summary

Modifier and Type	Method	Description
reactor.core.publisher.Flux<T>	acknowledge()
static reactor.util.retry.Retry	backoffRetry(long min, long max)
<U> MessageFlow<U>	batchFixed(int size, Function<? super MessageFlow<T>,? extends MessageFlow<U>> batchHandler)
<U> MessageFlow<U>	batchInterval(int duration, Function<? super MessageFlow<T>,? extends MessageFlow<U>> batchHandler)
<U> MessageFlow<U>	batchSamplingInterval(int duration, Function<? super MessageFlow<T>,? extends MessageFlow<U>> batchHandler)
T	blockLast()
T	blockLastRetry()
T	blockLastRetry(reactor.util.retry.Retry retry)
<A, R> MessageFlow<R>	collect(Collector<T,A,R> collector)
<U> MessageFlow<U>	concatMap(Function<? super T,? extends org.reactivestreams.Publisher<? extends U>> handler)
MessageFlow<T>	concatWith(org.reactivestreams.Publisher<? extends Message<T>> other)
<U> MessageFlow<U>	connect(Function<? super MessageFlow<T>,? extends org.reactivestreams.Publisher<? extends Message<U>>> handler)
MessageFlow<Void>	consume(Function<? super T,? extends org.reactivestreams.Publisher<Void>> handler)
MessageFlow<Void>	consume(Function<? super T,? extends org.reactivestreams.Publisher<Void>> handler, int concurrency)
MessageFlow<T>	doFinally(Consumer<reactor.core.publisher.SignalType> signalConsumer)
MessageFlow<T>	doOnEach(Consumer<? super reactor.core.publisher.Signal<T>> signalConsumer)
<E extends Throwable> MessageFlow<T>	doOnError(Class<E> exceptionType, Consumer<? super E> onError)
MessageFlow<T>	doOnError(Consumer<? super Throwable> onError)
MessageFlow<T>	doOnError(Predicate<? super Throwable> predicate, Consumer<? super Throwable> onError)
MessageFlow<T>	doOnNext(Consumer<? super T> onNext)
static <T> MessageFlow<T>	empty()
static <T> MessageFlow<T>	error(Throwable error)
final <U> MessageFlow<U>	fan(Function<? super MessageFlow<T>,? extends org.reactivestreams.Publisher<? extends Message<U>>>... consumerFlows)
MessageFlow<T>	filter(Predicate<? super T> predicate)
<U> MessageFlow<U>	flatMap(Function<? super T,? extends org.reactivestreams.Publisher<? extends U>> handler)
<U> MessageFlow<U>	flatMap(Function<? super T,? extends org.reactivestreams.Publisher<? extends U>> handler, int concurrency)
<K, U> MessageFlow<U>	group(Function<T,K> keyMapper, Function<K,Function<? super MessageFlow<T>,? extends org.reactivestreams.Publisher<Message<U>>>> groupHandlerFlow)
<K, U> MessageFlow<U>	group(Function<T,K> keyMapper, Function<K,Function<? super MessageFlow<T>,? extends org.reactivestreams.Publisher<Message<U>>>> groupHandlerFlow, int maxGroups)
<U> MessageFlow<U>	handle(BiConsumer<? super T,reactor.core.publisher.SynchronousSink<U>> biConsumer)
static <T> MessageFlow<T>	just(Message<T>... data)
<U> MessageFlow<U>	map(Function<? super T,? extends U> mapper)
static <T> MessageFlow<T>	of(org.reactivestreams.Publisher<? extends Message<T>> messageFlux)
MessageFlow<T>	reduce(BinaryOperator<T> reducer)
MessageFlow<T>	retry()
MessageFlow<T>	retry(reactor.util.retry.Retry retry)
reactor.core.Disposable	subscribe()
void	subscribe(org.reactivestreams.Subscriber<? super Message<T>> s)
reactor.core.Disposable	subscribeRetry()
reactor.core.Disposable	subscribeRetry(reactor.util.retry.Retry retry)
MessageFlow<T>	take(long n)
MessageFlow<T>	take(Duration duration)
MessageFlow<T>	takeUntilOther(org.reactivestreams.Publisher<?> other)
MessageFlow<T>	to(Function<T,String> addressMapper)
<U> MessageFlow<U>	withMessages(Function<? super reactor.core.publisher.Flux<Message<T>>,? extends org.reactivestreams.Publisher<Message<U>>> fluxMapper)

Methods inherited from class java.lang.Object

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

Methods inherited from interface io.keldysh.messaging.MessageSource

flow, get

Field Details

DEFAULT_RETRY_STRATEGY

public static final reactor.util.retry.Retry DEFAULT_RETRY_STRATEGY

Constructor Details

MessageFlow

public MessageFlow()

Method Details

backoffRetry

public static reactor.util.retry.Retry backoffRetry(long min,
 long max)

of

public static <T> MessageFlow<T> of(org.reactivestreams.Publisher<? extends Message<T>> messageFlux)

just

@SafeVarargs
public static <T> MessageFlow<T> just(Message<T>... data)

empty

public static <T> MessageFlow<T> empty()

error

public static <T> MessageFlow<T> error(Throwable error)

subscribe

public void subscribe(org.reactivestreams.Subscriber<? super Message<T>> s)

Specified by:

subscribe in interface org.reactivestreams.Publisher<T>

connect

public <U> MessageFlow<U> connect(Function<? super MessageFlow<T>,? extends org.reactivestreams.Publisher<? extends Message<U>>> handler)

withMessages

public <U> MessageFlow<U> withMessages(Function<? super reactor.core.publisher.Flux<Message<T>>,? extends org.reactivestreams.Publisher<Message<U>>> fluxMapper)

map

public <U> MessageFlow<U> map(Function<? super T,? extends U> mapper)

filter

public MessageFlow<T> filter(Predicate<? super T> predicate)

handle

public <U> MessageFlow<U> handle(BiConsumer<? super T,reactor.core.publisher.SynchronousSink<U>> biConsumer)

concatMap

public <U> MessageFlow<U> concatMap(Function<? super T,? extends org.reactivestreams.Publisher<? extends U>> handler)

flatMap

public <U> MessageFlow<U> flatMap(Function<? super T,? extends org.reactivestreams.Publisher<? extends U>> handler)

flatMap

public <U> MessageFlow<U> flatMap(Function<? super T,? extends org.reactivestreams.Publisher<? extends U>> handler,
 int concurrency)

consume

public MessageFlow<Void> consume(Function<? super T,? extends org.reactivestreams.Publisher<Void>> handler)

consume

public MessageFlow<Void> consume(Function<? super T,? extends org.reactivestreams.Publisher<Void>> handler,
 int concurrency)

batchFixed

public <U> MessageFlow<U> batchFixed(int size,
 Function<? super MessageFlow<T>,? extends MessageFlow<U>> batchHandler)

batchInterval

public <U> MessageFlow<U> batchInterval(int duration,
 Function<? super MessageFlow<T>,? extends MessageFlow<U>> batchHandler)

batchSamplingInterval

public <U> MessageFlow<U> batchSamplingInterval(int duration,
 Function<? super MessageFlow<T>,? extends MessageFlow<U>> batchHandler)

reduce

public MessageFlow<T> reduce(BinaryOperator<T> reducer)

collect

public <A,
R> MessageFlow<R> collect(Collector<T,A,R> collector)

group

public <K,
U> MessageFlow<U> group(Function<T,K> keyMapper,
 Function<K,Function<? super MessageFlow<T>,? extends org.reactivestreams.Publisher<Message<U>>>> groupHandlerFlow)

group

public <K,
U> MessageFlow<U> group(Function<T,K> keyMapper,
 Function<K,Function<? super MessageFlow<T>,? extends org.reactivestreams.Publisher<Message<U>>>> groupHandlerFlow,
 int maxGroups)

fan

@SafeVarargs
public final <U> MessageFlow<U> fan(Function<? super MessageFlow<T>,? extends org.reactivestreams.Publisher<? extends Message<U>>>... consumerFlows)

retry

public MessageFlow<T> retry()

retry

public MessageFlow<T> retry(reactor.util.retry.Retry retry)

to

public MessageFlow<T> to(Function<T,String> addressMapper)

acknowledge

public reactor.core.publisher.Flux<T> acknowledge()

subscribe

public reactor.core.Disposable subscribe()

subscribeRetry

public reactor.core.Disposable subscribeRetry()

subscribeRetry

public reactor.core.Disposable subscribeRetry(reactor.util.retry.Retry retry)

take

public MessageFlow<T> take(long n)

take

public MessageFlow<T> take(Duration duration)

takeUntilOther

public MessageFlow<T> takeUntilOther(org.reactivestreams.Publisher<?> other)

blockLast

public T blockLast()

blockLastRetry

public T blockLastRetry()

blockLastRetry

public T blockLastRetry(reactor.util.retry.Retry retry)

doOnEach

public MessageFlow<T> doOnEach(Consumer<? super reactor.core.publisher.Signal<T>> signalConsumer)

doOnError

public MessageFlow<T> doOnError(Consumer<? super Throwable> onError)

doOnError

public <E extends Throwable> MessageFlow<T> doOnError(Class<E> exceptionType,
 Consumer<? super E> onError)

doOnError

public MessageFlow<T> doOnError(Predicate<? super Throwable> predicate,
 Consumer<? super Throwable> onError)

doOnNext

public MessageFlow<T> doOnNext(Consumer<? super T> onNext)

doFinally

public MessageFlow<T> doFinally(Consumer<reactor.core.publisher.SignalType> signalConsumer)

concatWith

public MessageFlow<T> concatWith(org.reactivestreams.Publisher<? extends Message<T>> other)