Release Notes

/**
 * Extension function - Handles the flow of command messages of type [C] by concurrently distributing the load across finite number of actors/handlers
 *
 * @param commands [Flow] of Command messages of type [C]
 * @param numberOfActors total number of actors/workers available for distributing the load. Minimum one.
 * @param actorsCapacity capacity of the actors channel's buffer
 * @param actorsStart actors coroutine start option
 * @param actorsContext additional to [CoroutineScope.coroutineContext] context of the actor coroutines.
 * @param partitionKey a function that calculates the partition key/routing key of command - commands with the same partition key will be handled with the same 'actor' to keep the ordering
 * @return [Flow] of stored Events of type [E]
 *
 */
fun <C, S, E> EventSourcingAggregate<C, S, E>.handleConcurrently(
    commands: Flow<C>,
    numberOfActors: Int = 100,
    actorsCapacity: Int = Channel.BUFFERED,
    actorsStart: CoroutineStart = CoroutineStart.LAZY,
    actorsContext: CoroutineContext = EmptyCoroutineContext,
    partitionKey: (C) -> Int
): Flow<E>

@Deprecated(
    message = "Use EventSourcingLockingAggregate constructor-like function instead",
    replaceWith = ReplaceWith("EventSourcingLockingAggregate(decider, eventRepository)")
)
fun <C, S, E, V> eventSourcingLockingAggregate(
    decider: IDecider<C, S, E>,
    eventRepository: EventLockingRepository<C, E, V>
): EventSourcingLockingAggregate<C, S, E, V> =
    object : EventSourcingLockingAggregate<C, S, E, V>,
        EventLockingRepository<C, E, V> by eventRepository,
        IDecider<C, S, E> by decider {}

fun evenNumberDecider(): Decider<EvenNumberCommand?, EvenNumberState, EvenNumberEvent?> =
    decider {
        initialState {
            evenNumberState {
                descriptionString { "Initial state" }
                valueInt { 0 }
            }
        }
        decide { c, s ->
            when (c) {
                is AddEvenNumber -> flowOf(
                    evenNumberAdded {
                        description { c.description }
                        value { s.value + c.value }
                    }
                )

                is SubtractEvenNumber -> flowOf(
                    evenNumberSubtracted {
                        description { c.description }
                        value { s.value - c.value }
                    }
                )

                null -> emptyFlow()
            }
        }
        evolve { s, e ->
            when (e) {
                is EvenNumberAdded ->
                    evenNumberState {
                        description { s.description + e.description }
                        value { e.value }
                    }

                is EvenNumberSubtracted ->
                    evenNumberState {
                        description { s.description - e.description }
                        value { e.value }
                    }

                null -> s
            }
        }
    }

  stateStoredLockingAggregate(
      decider = myDecider,
      stateRepository = myLockingRepository
  ).handleOptimistically(myCommand)

  eventSourcingLockingAggregate(
      decider = myDecider,
      stateRepository = myLockingRepository
  ).handleOptimistically(myCommand)

@ExperimentalContracts
@FlowPreview
fun <C, S, E> EventSourcingAggregate<C, S, E>.handleConcurrently(
    commands: Flow<C>,
    numberOfActors: Int = 100,
    actorsCapacity: Int = Channel.BUFFERED,
    actorsStart: CoroutineStart = CoroutineStart.LAZY,
    actorsContext: CoroutineContext = EmptyCoroutineContext,
    partitionKey: (C) -> Int
): Flow<E> = channelFlow {
    val actors: List<SendChannel<C>> = (1..numberOfActors).map {
        commandActor(channel, actorsCapacity, actorsStart, actorsContext) { handle(it) }

    }
    commands
        .onCompletion {
            actors.forEach {
                it.close()
            }
        }
        .collect {
            val partition = partitionKey(it).absoluteValue % numberOfActors.coerceAtLeast(1)
            actors[partition].send(it)
        }
}

class DeciderTest : FunSpec({
    val evenDecider = evenNumberDecider()
    val oddDecider = oddNumberDecider()

    test("Event-sourced Decider - add even number") {
        with(evenDecider) {
            givenEvents(emptyList()) {
                whenCommand(AddEvenNumber(Description("2"), NumberValue(2)))
            } thenEvents listOf(EvenNumberAdded(Description("2"), NumberValue(2)))
        }
    }

    test("Event-sourced Decider - given previous state, add even number") {
        with(evenDecider) {
            givenEvents(listOf(EvenNumberAdded(Description("2"), NumberValue(2)))) {
                whenCommand(AddEvenNumber(Description("4"), NumberValue(4)))
            } thenEvents listOf(EvenNumberAdded(Description("4"), NumberValue(4)))
        }
    }
})