KState-Saga/ca.acendas.kstate.saga/FlowBasedSagaExecutor

FlowBasedSagaExecutor

class FlowBasedSagaExecutor<C, R>(delegate: SagaExecutor<C, R>) : SagaExecutor<C, R>

Flow-based saga executor that emits events as a reactive stream.

This executor wraps a standard SagaExecutor and provides Flow-based event streaming for reactive consumption of saga lifecycle events.

Usage Example

val executor = sagaExecutor<OrderContext, OrderResult> {
    step("reserve") { context -> /* ... */}
    step("charge") { context -> /* ... */}
}.asFlow()

// Collect events reactively
launch {
    executor.events.collect { event ->
        println("Event: $event")
    }
}

// Execute saga
val result = executor.execute(context)

Parameters

The type of context passed to saga steps

The type of result produced by saga steps

Constructors

FlowBasedSagaExecutor

constructor(delegate: SagaExecutor<C, R>)

Properties

events

val events: SharedFlow<SagaEvent>

Reactive stream of saga events.

Functions

addMonitor

open override fun addMonitor(monitor: SagaMonitor)

Add a monitor to observe saga events.

asFlow

fun <C, R> SagaExecutor<C, R>.asFlow(): FlowBasedSagaExecutor<C, R>

Extension function to convert a SagaExecutor to a FlowBasedSagaExecutor.

execute

open suspend override fun execute(context: C): SagaResult<R>

Execute the saga with the given context.

open suspend override fun execute(context: C, coroutineContext: CoroutineContext, timeout: Duration?): SagaResult<R>

Execute the saga with the given context, coroutine context, and optional timeout.

executeToFlow

fun <C, R> SagaExecutor<C, R>.executeToFlow(context: C, coroutineContext: CoroutineContext = EmptyCoroutineContext, timeout: Duration? = null): Flow<SagaEvent>

Execute saga and return a Flow that emits all events followed by a terminal result event.

removeMonitor

open override fun removeMonitor(monitor: SagaMonitor)

Remove a monitor.