wow-core/me.ahoo.wow.command.wait/MonoCommandWaitNotifier

MonoCommandWaitNotifier

class MonoCommandWaitNotifier<E : MessageExchange<*, M>, M : Message<*, *>, CommandId, NamedBoundedContext, AggregateIdCapable>(commandWaitNotifier: CommandWaitNotifier, processingStage: CommandStage, messageExchange: E, source: Mono<Void>) : Mono<Void>

A Mono wrapper that automatically notifies wait strategies when command processing completes. This class intercepts the completion of a Mono operation and sends appropriate wait signals to registered wait strategies based on the processing stage and message exchange.

Parameters

The type of message exchange.

The type of message in the exchange.

commandWaitNotifier

The notifier used to send wait signals.

processingStage

The command processing stage being notified.

messageExchange

The message exchange containing processing context.

source

The original Mono operation to wrap.

Constructors

MonoCommandWaitNotifier

constructor(commandWaitNotifier: CommandWaitNotifier, processingStage: CommandStage, messageExchange: E, source: Mono<Void>)

Functions

and

fun and(other: Publisher<*>?): Mono<Void?>?

fun as(transformer: Function<in Mono<Void?>?, P?>?): P?

block

@Nullable

open fun block(): Void?

@Nullable

open fun block(timeout: Duration?): Void?

blockOptional

open fun blockOptional(): Optional<Void?>?

open fun blockOptional(timeout: Duration?): Optional<Void?>?

cache

fun cache(): Mono<Void?>?

fun cache(ttl: Duration?): Mono<Void?>?

fun cache(ttl: Duration?, timer: Scheduler?): Mono<Void?>?

fun cache(ttlForValue: Function<in Void?, Duration?>?, ttlForError: Function<Throwable?, Duration?>?, ttlForEmpty: Supplier<Duration?>?): Mono<Void?>?

fun cache(ttlForValue: Function<in Void?, Duration?>?, ttlForError: Function<Throwable?, Duration?>?, ttlForEmpty: Supplier<Duration?>?, timer: Scheduler?): Mono<Void?>?

cacheInvalidateIf

fun cacheInvalidateIf(invalidationPredicate: Predicate<in Void?>?): Mono<Void?>?

cacheInvalidateWhen

fun cacheInvalidateWhen(invalidationTriggerGenerator: Function<in Void?, Mono<Void?>?>?): Mono<Void?>?

fun cacheInvalidateWhen(invalidationTriggerGenerator: Function<in Void?, Mono<Void?>?>?, onInvalidate: Consumer<in Void?>?): Mono<Void?>?

cancelOn

fun cancelOn(scheduler: Scheduler?): Mono<Void?>?

cast

fun <E : Any?> cast(clazz: Class<E?>?): Mono<E?>?

checkpoint

fun checkpoint(): Mono<Void?>?

fun checkpoint(description: String?): Mono<Void?>?

fun checkpoint(@Nullable description: String?, forceStackTrace: Boolean): Mono<Void?>?

concatWith

fun concatWith(other: Publisher<out Void?>?): Flux<Void?>?

contextCapture

fun contextCapture(): Mono<Void?>?

contextWrite

fun contextWrite(contextModifier: Function<Context?, Context?>?): Mono<Void?>?

fun contextWrite(contextToAppend: ContextView?): Mono<Void?>?

defaultIfEmpty

fun defaultIfEmpty(defaultV: Void?): Mono<Void?>?

delayElement

fun delayElement(delay: Duration?): Mono<Void?>?

fun delayElement(delay: Duration?, timer: Scheduler?): Mono<Void?>?

delaySubscription

fun delaySubscription(delay: Duration?): Mono<Void?>?

fun delaySubscription(subscriptionDelay: Publisher<U?>?): Mono<Void?>?

fun delaySubscription(delay: Duration?, timer: Scheduler?): Mono<Void?>?

delayUntil

fun delayUntil(triggerProvider: Function<in Void?, out Publisher<*>?>?): Mono<Void?>?

dematerialize

fun <X : Any?> dematerialize(): Mono<X?>?

doAfterTerminate

fun doAfterTerminate(afterTerminate: Runnable?): Mono<Void?>?

doFinally

fun doFinally(onFinally: Consumer<SignalType?>?): Mono<Void?>?

doFirst

fun doFirst(onFirst: Runnable?): Mono<Void?>?

doOnCancel

fun doOnCancel(onCancel: Runnable?): Mono<Void?>?

doOnDiscard

fun <R : Any?> doOnDiscard(type: Class<R?>?, discardHook: Consumer<in R?>?): Mono<Void?>?

doOnEach

fun doOnEach(signalConsumer: Consumer<in Signal<Void?>?>?): Mono<Void?>?

doOnError

fun doOnError(onError: Consumer<in Throwable?>?): Mono<Void?>?

fun <E : Throwable?> doOnError(exceptionType: Class<E?>?, onError: Consumer<in E?>?): Mono<Void?>?

fun doOnError(predicate: Predicate<in Throwable?>?, onError: Consumer<in Throwable?>?): Mono<Void?>?

doOnNext

fun doOnNext(onNext: Consumer<in Void?>?): Mono<Void?>?

doOnRequest

fun doOnRequest(consumer: LongConsumer?): Mono<Void?>?

doOnSubscribe

fun doOnSubscribe(onSubscribe: Consumer<in Subscription?>?): Mono<Void?>?

doOnSuccess

fun doOnSuccess(onSuccess: Consumer<in Void?>?): Mono<Void?>?

doOnTerminate

fun doOnTerminate(onTerminate: Runnable?): Mono<Void?>?

elapsed

fun elapsed(): Mono<Tuple2<Long?, Void?>?>?

fun elapsed(scheduler: Scheduler?): Mono<Tuple2<Long?, Void?>?>?

expand

fun expand(expander: Function<in Void?, out Publisher<out Void?>?>?): Flux<Void?>?

fun expand(expander: Function<in Void?, out Publisher<out Void?>?>?, capacityHint: Int): Flux<Void?>?

expandDeep

fun expandDeep(expander: Function<in Void?, out Publisher<out Void?>?>?): Flux<Void?>?

fun expandDeep(expander: Function<in Void?, out Publisher<out Void?>?>?, capacityHint: Int): Flux<Void?>?

filter

fun filter(tester: Predicate<in Void?>?): Mono<Void?>?

filterWhen

fun filterWhen(asyncPredicate: Function<in Void?, out Publisher<Boolean?>?>?): Mono<Void?>?

finallyAck

fun Mono<*>.finallyAck(exchange: MessageExchange<*, *>): Mono<Void>

Ensures the exchange is acknowledged after Mono completion, even on error.

flatMap

fun <R : Any?> flatMap(transformer: Function<in Void?, out Mono<out R?>?>?): Mono<R?>?

flatMapIterable

fun <R : Any?> flatMapIterable(mapper: Function<in Void?, out Iterable<out R?>?>?): Flux<R?>?

flatMapMany

fun <R : Any?> flatMapMany(mapper: Function<in Void?, out Publisher<out R?>?>?): Flux<R?>?

fun <R : Any?> flatMapMany(mapperOnNext: Function<in Void?, out Publisher<out R?>?>?, mapperOnError: Function<in Throwable?, out Publisher<out R?>?>?, mapperOnComplete: Supplier<out Publisher<out R?>?>?): Flux<R?>?

flux

fun flux(): Flux<Void?>?

handle

fun <R : Any?> handle(handler: BiConsumer<in Void?, SynchronousSink<R?>?>?): Mono<R?>?

hasElement

fun hasElement(): Mono<Boolean?>?

hide

fun hide(): Mono<Void?>?

ignoreElement

fun ignoreElement(): Mono<Void?>?

log

fun log(): Mono<Void?>?

fun log(@Nullable category: String?): Mono<Void?>?

fun log(logger: Logger?): Mono<Void?>?

fun log(@Nullable category: String?, level: Level?, vararg options: SignalType?): Mono<Void?>?

fun log(@Nullable category: String?, level: Level?, showOperatorLine: Boolean, vararg options: SignalType?): Mono<Void?>?

fun log(logger: Logger?, level: Level?, showOperatorLine: Boolean, vararg options: SignalType?): Mono<Void?>?

logErrorResume

fun <T : Any> Mono<T>.logErrorResume(): Mono<T>

Extension function to log errors and resume with empty Mono.

map

fun <R : Any?> map(mapper: Function<in Void?, out R?>?): Mono<R?>?

mapNotNull

fun <R : Any?> mapNotNull(mapper: Function<in Void?, out R?>?): Mono<R?>?

materialize

fun materialize(): Mono<Signal<Void?>?>?

mergeWith

fun mergeWith(other: Publisher<out Void?>?): Flux<Void?>?

metrics

fun ~~metrics~~(): Mono<Void?>?

name

fun name(name: String?): Mono<Void?>?

ofType

fun ofType(clazz: Class<U?>?): Mono<U?>?

onErrorComplete

fun onErrorComplete(): Mono<Void?>?

fun onErrorComplete(type: Class<out Throwable?>?): Mono<Void?>?

fun onErrorComplete(predicate: Predicate<in Throwable?>?): Mono<Void?>?

onErrorContinue

fun onErrorContinue(errorConsumer: BiConsumer<Throwable?, Any?>?): Mono<Void?>?

fun <E : Throwable?> onErrorContinue(type: Class<E?>?, errorConsumer: BiConsumer<Throwable?, Any?>?): Mono<Void?>?

fun <E : Throwable?> onErrorContinue(errorPredicate: Predicate<E?>?, errorConsumer: BiConsumer<Throwable?, Any?>?): Mono<Void?>?

onErrorMap

fun onErrorMap(mapper: Function<in Throwable?, out Throwable?>?): Mono<Void?>?

fun <E : Throwable?> onErrorMap(type: Class<E?>?, mapper: Function<in E?, out Throwable?>?): Mono<Void?>?

fun onErrorMap(predicate: Predicate<in Throwable?>?, mapper: Function<in Throwable?, out Throwable?>?): Mono<Void?>?

onErrorResume

fun onErrorResume(fallback: Function<in Throwable?, out Mono<out Void?>?>?): Mono<Void?>?

fun <E : Throwable?> onErrorResume(type: Class<E?>?, fallback: Function<in E?, out Mono<out Void?>?>?): Mono<Void?>?

fun onErrorResume(predicate: Predicate<in Throwable?>?, fallback: Function<in Throwable?, out Mono<out Void?>?>?): Mono<Void?>?

onErrorReturn

fun onErrorReturn(fallbackValue: Void?): Mono<Void?>?

fun <E : Throwable?> onErrorReturn(type: Class<E?>?, fallbackValue: Void?): Mono<Void?>?

fun onErrorReturn(predicate: Predicate<in Throwable?>?, fallbackValue: Void?): Mono<Void?>?

onErrorStop

fun onErrorStop(): Mono<Void?>?

onTerminateDetach

fun onTerminateDetach(): Mono<Void?>?

fun or(other: Mono<out Void?>?): Mono<Void?>?

publish

fun <R : Any?> publish(transform: Function<in Mono<Void?>?, out Mono<out R?>?>?): Mono<R?>?

publishOn

fun publishOn(scheduler: Scheduler?): Mono<Void?>?

repeat

fun repeat(): Flux<Void?>?

fun repeat(predicate: BooleanSupplier?): Flux<Void?>?

fun repeat(numRepeat: Long): Flux<Void?>?

fun repeat(numRepeat: Long, predicate: BooleanSupplier?): Flux<Void?>?

repeatWhen

fun repeatWhen(repeatFactory: Function<Flux<Long?>?, out Publisher<*>?>?): Flux<Void?>?

repeatWhenEmpty

fun repeatWhenEmpty(repeatFactory: Function<Flux<Long?>?, out Publisher<*>?>?): Mono<Void?>?

fun repeatWhenEmpty(maxRepeat: Int, repeatFactory: Function<Flux<Long?>?, out Publisher<*>?>?): Mono<Void?>?

retry

fun retry(): Mono<Void?>?

fun retry(numRetries: Long): Mono<Void?>?

retryWhen

fun retryWhen(retrySpec: Retry?): Mono<Void?>?

fun share(): Mono<Void?>?

single

fun single(): Mono<Void?>?

singleOptional

fun singleOptional(): Mono<Optional<Void?>?>?

open override fun subscribe(actual: CoreSubscriber<in Void>)

fun subscribe(actual: Subscriber<in Void?>?)

fun subscribe(): Disposable?

fun subscribe(consumer: Consumer<in Void?>?): Disposable?

fun subscribe(@Nullable consumer: Consumer<in Void?>?, errorConsumer: Consumer<in Throwable?>?): Disposable?

fun subscribe(@Nullable consumer: Consumer<in Void?>?, @Nullable errorConsumer: Consumer<in Throwable?>?, @Nullable completeConsumer: Runnable?): Disposable?

fun subscribe(@Nullable consumer: Consumer<in Void?>?, @Nullable errorConsumer: Consumer<in Throwable?>?, @Nullable completeConsumer: Runnable?, @Nullable subscriptionConsumer: Consumer<in Subscription?>?): Disposable?

fun subscribe(@Nullable consumer: Consumer<in Void?>?, @Nullable errorConsumer: Consumer<in Throwable?>?, @Nullable completeConsumer: Runnable?, @Nullable initialContext: Context?): Disposable?

subscribeOn

fun subscribeOn(scheduler: Scheduler?): Mono<Void?>?

subscribeWith

fun <E : Subscriber<in Void?>?> subscribeWith(subscriber: E?): E?

switchIfEmpty

fun switchIfEmpty(alternate: Mono<out Void?>?): Mono<Void?>?

tag

fun tag(key: String?, value: String?): Mono<Void?>?

tagSource

fun <M> Mono<M>.tagSource(source: String): Mono<M>

Static extension function to tag a Mono publisher with a specified source identifier. This is used for metrics tagging to identify the component that generated the metrics.

take

fun take(duration: Duration?): Mono<Void?>?

fun take(duration: Duration?, timer: Scheduler?): Mono<Void?>?

takeUntilOther

fun takeUntilOther(other: Publisher<*>?): Mono<Void?>?

tap

fun tap(listenerGenerator: Function<ContextView?, SignalListener<Void?>?>?): Mono<Void?>?

fun tap(simpleListenerGenerator: Supplier<SignalListener<Void?>?>?): Mono<Void?>?

fun tap(listenerFactory: SignalListenerFactory<Void?, *>?): Mono<Void?>?

then

fun then(): Mono<Void?>?

fun <V : Any?> then(other: Mono<V?>?): Mono<V?>?

thenDefer

fun <R> Mono<*>.thenDefer(defer: () -> Mono<R>): Mono<R>

Chains a deferred Mono creation after this Mono completes.

thenEmpty

fun thenEmpty(other: Publisher<Void?>?): Mono<Void?>?

thenMany

fun <V : Any?> thenMany(other: Publisher<V?>?): Flux<V?>?

thenNotifyAndForget

fun <E : MessageExchange<*, M>, M : Message<*, *>, CommandId, NamedBoundedContext, AggregateIdCapable> Mono<Void>.thenNotifyAndForget(commandWaitNotifier: CommandWaitNotifier, processingStage: CommandStage, messageExchange: E): Mono<Void>

Extension function that wraps a Mono to automatically notify wait strategies on completion. This provides a convenient way to add wait notification behavior to any Mono operation in the command processing pipeline.

thenReturn

fun <V : Any?> thenReturn(value: V?): Mono<V?>?

thenRunnable

fun Mono<*>.thenRunnable(runnable: () -> Unit): Mono<Void>

Executes a runnable after this Mono completes successfully.

throwNotFoundIfEmpty

fun <T> Mono<T>.throwNotFoundIfEmpty(errorMsg: String = ErrorCodes.NOT_FOUND_MESSAGE, cause: Throwable? = null): Mono<T>

Throws NotFoundResourceException if the Mono is empty.

timed

fun timed(): Mono<Timed<Void?>?>?

fun timed(clock: Scheduler?): Mono<Timed<Void?>?>?

timeout

fun timeout(timeout: Duration?): Mono<Void?>?

fun timeout(firstTimeout: Publisher<U?>?): Mono<Void?>?

fun timeout(timeout: Duration?, fallback: Mono<out Void?>?): Mono<Void?>?

fun timeout(timeout: Duration?, timer: Scheduler?): Mono<Void?>?

fun timeout(firstTimeout: Publisher<U?>?, fallback: Mono<out Void?>?): Mono<Void?>?

fun timeout(timeout: Duration?, @Nullable fallback: Mono<out Void?>?, timer: Scheduler?): Mono<Void?>?

timestamp

fun timestamp(): Mono<Tuple2<Long?, Void?>?>?

fun timestamp(scheduler: Scheduler?): Mono<Tuple2<Long?, Void?>?>?

toBlockable

fun <T> Mono<T>.toBlockable(scheduler: Scheduler = Schedulers.boundedElastic()): Mono<T>

Extension function that makes a Mono blockable by scheduling it on a separate thread if needed. If the current thread is non-blocking (reactive), it subscribes the Mono on the provided scheduler. If already on a blocking thread, returns the Mono unchanged.

toFuture

fun toFuture(): CompletableFuture<Void?>?

transform

fun <V : Any?> transform(transformer: Function<in Mono<Void?>?, out Publisher<V?>?>?): Mono<V?>?

transformDeferred

fun <V : Any?> transformDeferred(transformer: Function<in Mono<Void?>?, out Publisher<V?>?>?): Mono<V?>?

transformDeferredContextual

fun <V : Any?> transformDeferredContextual(transformer: BiFunction<in Mono<Void?>?, in ContextView?, out Publisher<V?>?>?): Mono<V?>?

zipWhen

fun <T2 : Any?> zipWhen(rightGenerator: Function<Void?, Mono<out T2?>?>?): Mono<Tuple2<Void?, T2?>?>?

fun <T2 : Any?, O : Any?> zipWhen(rightGenerator: Function<Void?, Mono<out T2?>?>?, combinator: BiFunction<Void?, T2?, O?>?): Mono<O?>?

zipWith

fun <T2 : Any?> zipWith(other: Mono<out T2?>?): Mono<Tuple2<Void?, T2?>?>?

fun <T2 : Any?, O : Any?> zipWith(other: Mono<out T2?>?, combinator: BiFunction<in Void?, in T2?, out O?>?): Mono<O?>?