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

Package-level declarations

Types

abstract class AbstractNotifierFilter<T : MessageExchange<*, M>, M : Message<*, *>, CommandId, NamedBoundedContext, AggregateIdCapable>(processingStage: CommandStage, commandWaitNotifier: CommandWaitNotifier) : ExchangeFilter<T>

Abstract base class for filters that notify wait plans about command processing completion. These filters intercept message processing pipelines and send notifications to waiting clients when specific processing stages are reached.

ChainWaitTarget

data class ChainWaitTarget(val function: NamedFunctionInfoData, val tail: WaitingChainTail) : WaitTarget

CommandStage

enum CommandStage : Enum<CommandStage>

Enumeration of command processing stages.

CommandStageCapable

interface CommandStageCapable

Interface for objects that have an associated command stage.

CommandWait

object CommandWait

CommandWaitEndpoint

interface CommandWaitEndpoint

Interface for defining command wait endpoints. Command wait endpoints specify where command processing results should be sent when using wait plans in distributed scenarios.

CommandWaitNotifier

interface CommandWaitNotifier

Interface for notifying command wait endpoints about processing results. After command processors complete their work, they send results to waiting clients through implementations of this interface.

CommandWaitNotifierSubscriber

class CommandWaitNotifierSubscriber<E : MessageExchange<*, M>, M : Message<*, *>, CommandId, NamedBoundedContext, AggregateIdCapable>(commandWaitNotifier: CommandWaitNotifier, processingStage: CommandStage, waitPlan: ExtractedWaitPlan, messageExchange: E, actual: CoreSubscriber<in Void>) : BaseSubscriber<Void>

Subscriber that handles command processing completion and sends wait notifications. This subscriber wraps the actual subscriber and intercepts completion/error events to send appropriate wait signals to waiting clients.

DefaultWaitCoordinator

class DefaultWaitCoordinator(streamQueueLinkSize: Int = DEFAULT_WAIT_STREAM_QUEUE_LINK_SIZE) : WaitCoordinator

EventHandledNotifierFilter

@Order(value = -2147483648)

class EventHandledNotifierFilter(commandWaitNotifier: CommandWaitNotifier) : AbstractNotifierFilter<DomainEventExchange<Any>, DomainEvent<*>>

Filter that notifies wait plans when domain events are handled by event processors. Intercepts the domain event dispatcher pipeline to send EVENT_HANDLED stage notifications.

ExtractedWaitPlan

data class ExtractedWaitPlan(val endpoint: String, val waitCommandId: String, val plan: WaitPlan) : CommandWaitEndpoint, WaitCommandIdCapable, MessagePropagator

LocalCommandWaitNotifier

class LocalCommandWaitNotifier(waitCoordinator: WaitCoordinator) : CommandWaitNotifier

Local implementation of CommandWaitNotifier for in-process notifications. This notifier forwards wait signals to registered wait handles within the same JVM instance.

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 plans when command processing completes. This class intercepts the completion of a Mono operation and sends appropriate wait signals based on the processing stage and message exchange.

NullableAggregateVersionCapable

interface NullableAggregateVersionCapable

Interface for objects that may have an aggregate version.

ProcessedNotifierFilter

@Order(value = -2147483648)

class ProcessedNotifierFilter(commandWaitNotifier: CommandWaitNotifier) : CommandFilter, AbstractNotifierFilter<ServerCommandExchange<*>, CommandMessage<*>>

Filter that notifies wait plans when command processing is complete. Intercepts the command dispatcher pipeline to send PROCESSED stage notifications.

ProjectedNotifierFilter

@Order(value = -2147483648)

class ProjectedNotifierFilter(commandWaitNotifier: CommandWaitNotifier) : AbstractNotifierFilter<DomainEventExchange<Any>, DomainEvent<*>>

Filter that notifies wait plans when projections are updated. Intercepts the projection dispatcher pipeline to send PROJECTED stage notifications.

SagaHandledNotifierFilter

@Order(value = -2147483648)

class SagaHandledNotifierFilter(commandWaitNotifier: CommandWaitNotifier) : AbstractNotifierFilter<DomainEventExchange<Any>, DomainEvent<*>>

Filter that notifies wait plans when domain events are handled by sagas. Intercepts the stateless saga dispatcher pipeline to send SAGA_HANDLED stage notifications.

SignalTimeCapable

interface SignalTimeCapable

Interface for objects that have a signal timestamp.

SimpleCommandWaitEndpoint

data class SimpleCommandWaitEndpoint(val endpoint: String) : CommandWaitEndpoint

Simple implementation of CommandWaitEndpoint. Provides a basic data class wrapper for endpoint strings.

SimpleWaitPlan

data class SimpleWaitPlan(val waitCommandId: String, val target: WaitTarget, val supportVoidCommand: Boolean = false) : WaitPlan

SimpleWaitSignal

data class SimpleWaitSignal(val id: String, val waitCommandId: String, val commandId: String, val aggregateId: AggregateId, val stage: CommandStage, val function: FunctionInfoData, val aggregateVersion: Int? = null, val isLastProjection: Boolean = false, val errorCode: String = ErrorCodes.SUCCEEDED, val errorMsg: String = ErrorCodes.SUCCEEDED_MESSAGE, val bindingErrors: List<BindingError> = emptyList(), val result: Map<String, Any> = emptyMap(), val commands: List<String> = listOf(), val signalTime: Long = System.currentTimeMillis()) : WaitSignal

Simple implementation of WaitSignal.

SnapshotNotifierFilter

@Order(value = -2147483648)

class SnapshotNotifierFilter(commandWaitNotifier: CommandWaitNotifier) : AbstractNotifierFilter<StateEventExchange<*>, StateEvent<*>>

Filter that notifies wait plans when aggregate snapshots are generated. Intercepts the snapshot dispatcher pipeline to send SNAPSHOT stage notifications.

StageWaitTarget

data class StageWaitTarget(val stage: CommandStage, val function: NamedFunctionInfoData? = null) : WaitTarget

WaitCommandIdCapable

interface WaitCommandIdCapable

Interface for objects that have an associated wait command ID. The wait command ID is used to correlate wait plans with their corresponding command executions in distributed scenarios.

WaitCoordinator

interface WaitCoordinator

WaitHandle

interface WaitHandle : WaitCommandIdCapable

WaitLastHandle

interface WaitLastHandle : WaitHandle

WaitPlan

interface WaitPlan : WaitCommandIdCapable

WaitSignal

interface WaitSignal : Identifier, WaitCommandIdCapable, CommandId, AggregateIdCapable, NullableAggregateVersionCapable, ErrorInfo, SignalTimeCapable, CommandResultCapable, FunctionInfoCapable<FunctionInfoData>

Signal representing a command processing stage completion.

WaitStreamHandle

interface WaitStreamHandle : WaitHandle

WaitTarget

sealed interface WaitTarget

Properties

COMMAND_WAIT_CONTEXT

const val COMMAND_WAIT_CONTEXT: String

COMMAND_WAIT_ENDPOINT

const val COMMAND_WAIT_ENDPOINT: String

COMMAND_WAIT_FUNCTION

const val COMMAND_WAIT_FUNCTION: String

COMMAND_WAIT_PREFIX

const val COMMAND_WAIT_PREFIX: String

COMMAND_WAIT_PROCESSOR

const val COMMAND_WAIT_PROCESSOR: String

COMMAND_WAIT_STAGE

const val COMMAND_WAIT_STAGE: String

DEFAULT_WAIT_STREAM_QUEUE_LINK_SIZE

const val DEFAULT_WAIT_STREAM_QUEUE_LINK_SIZE: Int = 16

WAIT_COMMAND_ID

const val WAIT_COMMAND_ID: String

Functions

extractCommandWaitEndpoint

fun Header.extractCommandWaitEndpoint(): String?

extractCommandWaitId

fun Header.extractCommandWaitId(): String?

extractWaitFunction

fun Header.extractWaitFunction(): NamedFunctionInfoData

extractWaitingStage

fun Header.extractWaitingStage(): CommandStage?

extractWaitPlan

fun Header.extractWaitPlan(): ExtractedWaitPlan?

extractWaitTarget

fun Header.extractWaitTarget(): WaitTarget?

isLocalWaitCommandId

fun isLocalWaitCommandId(commandWaitId: String): Boolean

Determines if a command wait ID belongs to the current JVM instance. Uses the global ID generator to check if the machine ID in the wait ID matches the current machine's ID.

matchesWaitFunction

fun NamedFunctionInfo?.matchesWaitFunction(function: FunctionInfo): Boolean

notifyAndForget

fun CommandWaitNotifier.notifyAndForget(waitPlan: ExtractedWaitPlan, waitSignal: WaitSignal)

Extension function to notify and forget using an extracted wait plan. Only sends notification if the wait plan should be notified for this signal.

propagateCommandWaitEndpoint

fun Header.propagateCommandWaitEndpoint(endpoint: String): Header

propagateWaitCommandId

fun Header.propagateWaitCommandId(commandId: String): Header

propagateWaitFunction

fun Header.propagateWaitFunction(function: NamedFunctionInfoData?): Header

propagateWaitingStage

fun Header.propagateWaitingStage(stage: CommandStage): Header

propagateWaitTarget

fun Header.propagateWaitTarget(target: WaitTarget): Header

requireExtractCommandWaitEndpoint

fun Header.requireExtractCommandWaitEndpoint(): String

requireExtractWaitCommandId

fun Header.requireExtractWaitCommandId(): String

shouldPropagate

fun WaitTarget.shouldPropagate(upstream: Message<*, *>): Boolean

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 plans on completion. This provides a convenient way to add wait notification behavior to any Mono operation in the command processing pipeline.