A Kotlin Multiplatform CQRS message bus framework. Route Commands, Queries, and Events to their handlers with a composable middleware pipeline, Unit of Work support, and optional KSP code generation for compile-time type-safe handler resolution (zero reflection).

• CQRS message routing — Commands, Queries, and Events each with dedicated handler types
• Kotlin Multiplatform — JVM, JS, WASM, macOS, iOS, Linux, Windows
• Coroutine-first — All handlers are suspend functions
• KSP code generation — Compile-time handler resolution with zero reflection
• Middleware pipeline — Composable middleware chain wrapping handler execution
• Unit of Work — Transaction-aware command execution with domain and integration events
• Domain modeling — Built-in support for Entities, Aggregate Roots, and Value Objects
• Result types — Type-safe BusResult<TValue, TMessageFailure> with Success and Failure variants

Add the dependencies to your build.gradle.kts:

dependencies {
    implementation("com.jimbroze:kbus-core:<version>")

    // For KSP code generation (optional)
    implementation("com.jimbroze:kbus-annotations:<version>")
    ksp("com.jimbroze:kbus-generation:<version>")
}

// A command that returns a String result
class CreateUser(val name: String, val email: String) :
    Command<BusResult<String, MessageFailure>>()

class CreateUserHandler :
    CommandHandler<CreateUser, BusResult<String, MessageFailure>>() {

    override suspend fun handle(message: CreateUser): BusResult<String, MessageFailure> {
        // Create the user...
        return BusResult.success("User ${message.name} created")
    }
}

// A query that returns a String result
class GetUser(val id: Int) :
    Query<BusResult<String, MessageFailure>>()

class GetUserHandler :
    QueryHandler<GetUser, BusResult<String, MessageFailure>>() {

    override suspend fun handle(message: GetUser): BusResult<String, MessageFailure> {
        // Look up the user...
        return BusResult.success("User #${message.id}")
    }
}

You can get the full code here.

suspend fun main() {
    // Register handlers
    val stores = HandlerFactoryStoreCollection()
    stores.commandStore.registerHandlers(
        CreateUser::class,
        listOf(CommandHandlerFactory(CreateUserHandler::class) { _: CommandDependencies -> CreateUserHandler() })
    )
    stores.queryStore.registerHandlers(
        GetUser::class,
        listOf(QueryHandlerFactory(GetUserHandler::class) { GetUserHandler() })
    )

    // Create the bus
    val bus = MessageBus(PersistingHandlerLocator(stores))

    // Execute a command
    val result = bus.execute(CreateUser("Alice", "alice@example.com"))
    if (result.isSuccess) {
        println(result.getOrNull()) // "User Alice created"
    }

    // Fetch a query
    val userResult = bus.fetch(GetUser(1))
}

You can get the full code here.

KBUS has three message types, each with a corresponding handler:

Events support multiple handlers and come in two flavors:

Domain events dispatch relative to the Unit of Work lifecycle. Publish them from a domain object by taking a DomainEventPublisher as a constructor dependency. See the section on // TODO

class OrderShipped(val orderId: String) : DomainEvent()

class Order(private val domainEventPublisher: DomainEventPublisher) {

    suspend fun place(orderId: String) {
        // Place the order...

        domainEventPublisher.publish(OrderShipped(orderId))
    }
}

You can get the full code here.

CommandDependencies (which contains DomainEventPublisher) is injected into command handlers automatically and routes events through the Unit of Work.

The safety of an error strategy depends entirely on when the handler executes relative to the database transaction.

• Pre-Commit: You can safely throw exceptions to roll back the transaction.
• Post-Commit: You lose throwing privileges and must rely on logging or Dead Letter Queues (DLQ).

All events can be dispatched sequentially or concurrently by applying DispatchSequentially or DispatchConcurrently interfaces to the event. This applies regardless of dispatch timing or error strategy. That is, while concurrent events will dispatch to multiple handlers at the same time, FAIL_FAST concurrent events will still throw on the first failure; meaning all running handlers for that event will cancel.

By default, domain event handlers that extend DomainEventHandler directly are dispatched asynchronously after the transaction commits. This default is intentional:

• Asynchronous by default — All events (both domain and integration) default to asynchronous, fire-and-forget dispatch. If work must be synchronous and transactional, it should ideally be modeled as an explicit Command, not an Event. This keeps coupled operations visible in the code rather than hiding them behind event handlers that appear decoupled but are actually tightly bound. Synchronous event dispatch should be avoided where possible outside of infrastructure concerns.
• After transaction by default — Domain events default to dispatching after a unit of work transaction commits because handlers typically trigger side effects (notifications, external calls) that should only happen once the primary work has been persisted. Dispatching before commit risks executing side effects for work that may still be rolled back.

// Dispatched immediately when the event is raised (synchronous)
class NotifyWarehouse : DomainEventHandler<OrderShipped>() {
    override val dispatchTiming = DispatchTiming.ImmediatelyInTransaction

    override suspend fun handle(message: OrderShipped) {
        /* ... */
    }
}

// Dispatched after the primary handler completes but before transaction commit (synchronous)
class UpdateInventory : DomainEventHandler<OrderShipped>() {
    override val dispatchTiming = DispatchTiming.AtEndOfTransaction

    override suspend fun handle(message: OrderShipped) {
        /* ... */
    }
}

// Dispatched after the transaction has been committed (asynchronous)
class SendShipmentNotification : DomainEventHandler<OrderShipped>() {
    override val dispatchTiming = DispatchTiming.AfterTransaction

    override suspend fun handle(message: OrderShipped) {
        /* ... */
    }
}

You can get the full code here.

Integration events are dispatched after the transaction commits, intended for cross-boundary communication. Integration events are always dispatched asynchronously — handlers run concurrently in a fire-and-forget manner.

class UserRegistered(val userId: String) : IntegrationEvent()

class SyncToExternalCRM :
    IntegrationEventHandler<UserRegistered> {

    override suspend fun handle(message: UserRegistered) {
        // Sync to external system...
    }
}

You can get the full code here.

Integration events can be observed as Kotlin Flows directly from the bus. With the generated bus, only known events can be observed — attempting to observe an unknown event is a compile error:

// With the generated bus — type-safe, one method per known event
val bus = CompileTimeLoadedMessageBus(loader, transactionManager, middleware)

scope.launch {
    bus.observeUserRegistered().collect { event ->
        println("User registered: ${event.userId}")
    }
}

// With the runtime bus — any IntegrationEvent can be observed
val runtimeBus = MessageBus(handlerLocator)

scope.launch {
    runtimeBus.observe(UserRegistered::class).collect { event ->
        println("User registered: ${event.userId}")
    }
}

Events are emitted to observers before handlers are invoked. Observers receive events regardless of handler success or failure.

Command handlers can publish integration events:

class RegisterUserHandler :
    CommandHandler<RegisterUser, BusResult<String, MessageFailure>>() {

    override suspend fun handle(message: RegisterUser): BusResult<String, MessageFailure> {
        val userId = "generated-id"

        publish(UserRegistered(userId))

        return BusResult.success(userId)
    }
}

You can get the full code here.

The AutoPublishIntegrationEvents middleware publishes integration events automatically whenever a registered domain event is dispatched — no explicit publish call needed. Register mappings with autoPublish, either as a lambda or by implementing AutoPublishesFrom on the integration event's companion object to declare the domain event it is derived from. A domain event may be registered multiple times to publish several integration events.

class OrderPlaced(val orderId: String) : DomainEvent()

class OrderPlacedIntegration(val orderId: String) : IntegrationEvent() {
    companion object : AutoPublishesFrom<OrderPlaced> {
        override fun fromDomainEvent(event: OrderPlaced) = OrderPlacedIntegration(event.orderId)
    }
}

class OrderPlacedAnalytics(val orderId: String) : IntegrationEvent()

val busWithAutoPublish = MessageBus(
    handlerLocator = PersistingHandlerLocator(),
    middlewares = listOf(
        AutoPublishIntegrationEvents(
            autoPublish(OrderPlacedIntegration),
            autoPublish<OrderPlaced> { OrderPlacedAnalytics(it.orderId) },
        ),
    ),
)

You can get the full code here.

Registering every mapping by hand doesn't scale in a generated bus. Annotate the integration event with @LoadEvent and code generation collects every AutoPublishesFrom companion it finds into a generated generatedAutoPublishRegistrations list — see Auto-Publish Registrations below.

All commands and queries return BusResult<TValue, TMessageFailure>:

suspend fun main() {
    val result: BusResult<String, MessageFailure> = bus.execute(MyCommand())

    when {
        result.isSuccess -> println("Value: ${result.getOrNull()}")
        result.isFailure -> println("Error: ${result.failureOrNull()?.reason?.message}")
    }
}

You can get the full code here.

Create results with companion functions:

val success = BusResult.success("value")
val failure = BusResult.failure(GenericMessageFailure(GenericFailure("Something went wrong")))

You can get the full code here.

Middleware wraps handler execution in a composable pipeline. Each middleware can run logic before and after the next handler in the chain. Every handle call also receives a MiddlewareInvocationContext, a per-invocation context object passed to all middleware in the chain. It currently exposes integrationEventPublisher, an IntegrationEventPublisher wired to the bus's real dispatch path — middleware can use it to publish integration events directly, independent of any command handler's own publishing.

class TimingMiddleware : Middleware {
    override suspend fun <TMessage : Message, TResult> handle(
        message: TMessage,
        context: MiddlewareInvocationContext,
        nextMiddleware: MiddlewareHandler<TMessage, TResult>,
    ): TResult {
        val mark = TimeSource.Monotonic.markNow()
        try {
            return nextMiddleware(message)
        } finally {
            val duration = mark.elapsedNow()
            println("${message::class.simpleName} took $duration")
        }
    }
}

You can get the full code here.

Pass middleware when creating the bus:

val stores = HandlerFactoryStoreCollection()

val bus = MessageBus(
    handlerLocator = PersistingHandlerLocator(stores),
    middlewares = listOf(
        LoggingMiddleware(logger, DebugLevel, InfoLevel, ErrorLevel),
    )
)

You can get the full code here.

• LoggingMiddleware — Logs message dispatch, completion, and errors at configurable log levels
• LockingMiddleware — Prevents concurrent message handling with a configurable timeout
• AutoPublishIntegrationEvents — Publishes the integration event mapped from a registered domain event via AutoPublishesFrom

Commands execute within a Unit of Work that manages three phases:

1. Primary work — The command handler executes
2. Secondary work — Domain event handlers run (within the same transaction)
3. Post-commit work — Integration event handlers run (after transaction commit)

To opt into transactional execution, pass a TransactionManager to the bus to apply it globally:

val stores = HandlerFactoryStoreCollection()

val bus = MessageBus(
    handlerLocator = PersistingHandlerLocator(stores),
    transactionManager = myTransactionManager,
)

You can get the full code here.

Command handlers execute within a transaction by default. No additional configuration is needed:

class TransferFundsHandler : CommandHandler<TransferFunds, BusResult<Unit, MessageFailure>>() {

    override suspend fun handle(message: TransferFunds): BusResult<Unit, MessageFailure> {
        // This runs inside a transaction (default behavior)
        return BusResult.success(Unit)
    }
}

You can get the full code here.

You can provide a TransactionManager override to individual command handlers via TransactionConfig:

class TransferFundsHandler(
    transactionManager: TransactionManager
) : CommandHandler<TransferFunds, BusResult<Unit, MessageFailure>>() {
    override val executeInTransaction: TransactionConfig? =
        TransactionConfig(transactionManagerOverride = transactionManager)

    override suspend fun handle(message: TransferFunds): BusResult<Unit, MessageFailure> {
        // This runs inside a transaction with a custom TransactionManager
        return BusResult.success(Unit)
    }
}

You can get the full code here.

To opt out of transaction execution, set executeInTransaction to null:

class MyCommandHandler : CommandHandler<MyCommand, BusResult<Unit, MessageFailure>>() {
    override val executeInTransaction: TransactionConfig? = null

    override suspend fun handle(message: MyCommand): BusResult<Unit, MessageFailure> {
        // This runs without a transaction
        return BusResult.success(Unit)
    }
}

A bus with background work — an outbox, and/or any LifecycleAwareMiddleware (e.g. LockingMiddleware) — must be explicitly started before it dispatches messages:

val bus = MessageBus(/* ... */)
bus.start()

start() runs each LifecycleAwareMiddleware's onStart and, if an outbox is configured, launches its poller. It's idempotent — calling it again is a no-op — and a bus with neither an outbox nor lifecycle-aware middleware needs no start() call at all; execute/fetch work immediately, at zero ceremony.

Calling execute/fetch on a bus that does have background work, before start(), throws IllegalStateException rather than silently running with that work never having started (e.g. an outbox that never polls).

stop() is suspend: it calls each middleware's onStop(), then cancels the bus's root job and waits for that cancellation to complete, so shutdown is deterministic in tests and at application exit.

bus.stop()

Restart is unsupported — cancelling the root job is terminal, so a stop()ped bus cannot be start()ed again. stop() before start() is a no-op.

Integration events published during a command (both the imperative publish() and the AutoPublishIntegrationEvents middleware) normally publish inside the command's transaction and dispatch to handlers immediately. This leaves two gaps: a phantom event (published, then the transaction rolls back) and a lost event (committed, then the process crashes before handlers run). A transactional outbox closes both gaps by persisting events to a durable store inside the transaction, then delivering them separately.

The outbox hangs off CommandInvocation, the bus's per-command scope — a peer of TransactionManager on the bus constructor, not middleware. The Unit of Work itself has no knowledge of the outbox; the outbox instead receives the command's Unit of Work at construction time and self-registers into its generic secondary-work/post-commit-work phases. Opt in by passing an OutboxConfig:

val stores = HandlerFactoryStoreCollection()

val bus = MessageBus(
    handlerLocator = PersistingHandlerLocator(stores),
    outbox = OutboxConfig(store = InMemoryOutboxStore()),
).apply { start() }

You can get the full code here.

An outbox is background work, so the bus must be started before it dispatches messages — start() is what launches the poller.

InMemoryOutboxStore is a reference implementation for tests and examples. For production, implement OutboxStore against a durable table. The outbox defers the actual store write until an internal flush, self-registered as the first secondary-work item on the command's Unit of Work when the outbox is constructed — so save runs inside the command's TransactionManager.execute block, and your implementation must join the ambient transaction (the same by-convention contract as TransactionManager itself) — otherwise a rolled-back command would leave a phantom event saved to the outbox. This deferral is also what makes command-chain middleware publishes rollback-safe even though middleware runs before the transaction opens: a middleware-published event sits in memory only until the flush runs, so if the handler subsequently fails, the flush never happens and nothing is ever saved.

Publish and dispatch are decoupled. "Publish" is the durable save inside the transaction; "dispatch" is the async delivery to handlers afterward. A command's return value never awaits integration handler execution, whether or not an outbox is configured.

When an outbox is configured, every integration publish routes through it — not just command-scoped ones. Query middleware, AutoPublishIntegrationEvents outside a command, and integration event handlers that publish further events all get a durable, at-least-once-delivered publish too. Command-scoped publishes get the transactional save-then-flush behaviour described above; everything else gets "save immediately, then drain opportunistically" — there's no transaction to defer the save to, so the save happens up front and delivery follows the same fire-and-forget/poller-backstop pattern. The trade-off: non-command saves aren't atomic with whatever surrounding work triggered them, but events are never silently lost, and the same retry/DLQ policy applies uniformly across every publish path. Integration event handlers can themselves publish further events by extending CanPublishIntegrationEvent, the same mixin command handlers and domain event handlers use — the dispatcher wires each handler's publisher before dispatch, so these publishes are outbox-durable too.

A publish call's failure semantics differ slightly by path: on the transactional (command-scoped) path, publish only fails if the buffering itself fails (essentially never); on every other path, publish fails if the store save fails. Either way, delivery failures never propagate to the caller — they're the poller's problem.

Delivery is at-least-once. A bus-owned background poller is the delivery guarantee, repeatedly fetching unpublished entries and delivering them — this is what survives a crash between commit and delivery. The poller is started by bus.start(), not by construction. After a command's transaction commits (and any post-commit-phase handlers finish), a fire-and-forget drain of the events just captured also runs — self-registered as post-commit work by the outbox itself — purely as a latency optimisation; it is never awaited, and if it fails or is skipped the poller retries the entry on its next tick. Because the drain and the poller can overlap, and because multiple bus instances may run against the same store, a handler may occasionally run more than once — design handlers to be idempotent. OutboxStore.fetchUnpublished is the hook for narrowing that window (e.g. hiding very recently-saved or already-claimed entries).

An event's errorStrategy doubles as the outbox's acknowledgement semantics: FireAndForget marks an entry published as soon as delivery is handed off, while FailFast only marks it published once every handler has actually completed, letting the poller retry entries whose handlers threw. Either way, a handler failure no longer fails the command itself — the command has already returned by the time delivery happens.

A few edge cases worth knowing:

• Commands with executeInTransaction = null still route their events through the outbox, just without the atomicity of the save being part of a real transaction.
• A detached (FireAndForget, post-commit) domain event handler that calls publish() itself still publishes through the outbox, via its already-flushed path (saved immediately, no atomicity with the original command) — it does not bypass the outbox, but delivery for that event now depends on the poller rather than the command's own drain.
• Events published by POST_COMMIT-phase domain event handlers are saved to the outbox non-transactionally (the transaction has already committed), but are still captured and drained like any other event.

For compile-time type-safe handler resolution with zero reflection, use the KSP code generation module. This requires adding no annotations or coupling to anything outside your message handlers (which are already coupled to Kbus).

plugins {
    kotlin("multiplatform")
    alias(libs.plugins.devtools.ksp)
}

dependencies {
    implementation("com.jimbroze:kbus-core:<version>")
    implementation("com.jimbroze:kbus-annotations:<version>")
    add("kspCommonMainMetadata", "com.jimbroze:kbus-generation:<version>")
}

// Include generated sources
kotlin.sourceSets.commonMain {
    kotlin.srcDir("build/generated/ksp/metadata/commonMain/kotlin")
}

Mark handler classes with @LoadMessageHandler. Constructor parameters become automatically resolved dependencies:

@LoadMessageHandler
class PlaceOrderHandler(
    private val orderRepository: OrderRepository,
    private val paymentService: PaymentService,
) : CommandHandler<PlaceOrder, BusResult<String, MessageFailure>>() {

    override suspend fun handle(message: PlaceOrder): BusResult<String, MessageFailure> {
        val orderId = orderRepository.save(message.items)
        paymentService.charge(orderId)
        return BusResult.success(orderId)
    }
}

You can get the full code here.

The KSP processor generates:

• AllDependencies — Interface listing all required dependencies (implement this to provide them)
• AllHandlers — Interface with factory methods for every handler
• HandlerFactory — Factory that creates handlers with their dependencies resolved
• CompileTimeLoadedMessageBus — A type-safe bus with strongly-typed execute, fetch, and observe methods for each message type
• AutoLoader — Auto-loading support for runtime handler registration
• generatedAutoPublishRegistrations — List<AutoPublishRegistration<*>> collected from every @LoadEvent integration event whose companion implements AutoPublishesFrom (only generated when at least one exists)

// Implement the generated AutoLoader abstract class (or AllDependencies interface)
class MyDependencies : AutoLoader() {
    override val orderRepository: OrderRepository = OrderRepositoryImpl()
    override val paymentService: PaymentService = PaymentServiceImpl()
}

// Create the type-safe bus
val bus = CompileTimeLoadedMessageBus(
    loader = MyDependencies(),
    transactionManager = myTransactionManager,
    middleware = listOf(LoggingMiddleware(logger)),
)

// Strongly-typed dispatch — compile error if message type is wrong
val result = bus.execute(PlaceOrder(items))

@LoadEvent makes an event known to code generation. On its own it generates nothing — but if the annotated integration event's companion implements AutoPublishesFrom (see Auto-Publishing Integration Events), the processor also collects it into the generated generatedAutoPublishRegistrations list, so the mapping doesn't need to be registered by hand:

@LoadEvent
class OrderPlacedIntegration(val orderId: String) : IntegrationEvent() {
    companion object : AutoPublishesFrom<OrderPlaced> {
        override fun fromDomainEvent(event: OrderPlaced) = OrderPlacedIntegration(event.orderId)
    }
}

val bus = CompileTimeLoadedMessageBus(
    loader = MyDependencies(),
    transactionManager = myTransactionManager,
    middleware = listOf(AutoPublishIntegrationEvents(generatedAutoPublishRegistrations)),
)

An event annotated with @LoadEvent whose companion does not implement AutoPublishesFrom (or that has no companion) is still known to the processor — it just contributes no registration; this is not an error, and leaves room for other @LoadEvent-driven code generation in future.

For multi-module projects, submodules can export their handler metadata for the main module to consume. You must provide a package name for the indexes. This prevents trying to load indexes from a dependent library that uses Kbus.

// In the submodule's build.gradle.kts
ksp {
    arg("kbus.subModuleName", project.name)
    arg("kbus.indexPackage", "com.example.myApp.indexes")
}

// In the top-level module's build.gradle.kts
ksp {
    arg("kbus.indexPackage", "com.example.myApp.indexes")
}

Submodules generate a DependencyIndex with @KbusIndex metadata instead of full bus code. The main module picks up these indexes automatically, including any @LoadEvent/AutoPublishesFrom opt-ins, which are folded into the main module's generatedAutoPublishRegistrations.

KBUS includes base types for domain-driven design:

// Value Object — equals() and hashCode() required
class Money(val amount: Double, val currency: String) : ValueObject<Money>() {
    override fun equals(other: Any?) =
        other is Money && amount == other.amount && currency == other.currency

    override fun hashCode() = 31 * amount.hashCode() + currency.hashCode()
}

// Entity
class OrderId(private val value: String) : Identifier {
    override fun equals(other: Any?) = other is OrderId && value == other.value
    override fun hashCode() = value.hashCode()
}

class Order(override val id: OrderId, val items: List<String>) : Entity<Order>()

// Aggregate Root
class CartId(private val value: String) : Identifier {
    override fun equals(other: Any?) = other is CartId && value == other.value
    override fun hashCode() = value.hashCode()
}

class ShoppingCart(override val id: CartId) : AggregateRoot<ShoppingCart>()

You can get the full code here.

See LICENSE for details.