Facilitates cross-platform application development with shared code and platform-specific implementations, supporting seamless integration of iOS and web components alongside SwiftUI and Compose Multiplatform.
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).
suspend functionsBusResult<TValue, TMessageFailure> with Success and Failure variantsAdd 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>")
}
You can get the full code here.
suspend {
stores = HandlerFactoryStoreCollection()
stores.commandStore.registerHandlers(
CreateUser::,
listOf(CommandHandlerFactory(CreateUserHandler::) { _: CommandDependencies -> CreateUserHandler() })
)
stores.queryStore.registerHandlers(
GetUser::,
listOf(QueryHandlerFactory(GetUserHandler::) { GetUserHandler() })
)
bus = MessageBus(PersistingHandlerLocator(stores))
result = bus.execute(CreateUser(, ))
(result.isSuccess) {
println(result.getOrNull())
}
userResult = bus.fetch(GetUser())
}
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.
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:
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 dispatch integration events:
class RegisterUserHandler :
CommandHandler<RegisterUser, BusResult<String, MessageFailure>>() {
override suspend fun handle(message: RegisterUser): BusResult<String, MessageFailure> {
val userId = "generated-id"
dispatch(UserRegistered(userId))
return BusResult.success(userId)
}
}
You can get the full code here.
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.
class TimingMiddleware : Middleware {
override suspend fun <TMessage : Message, TResult> : TResult {
mark = TimeSource.Monotonic.markNow()
{
nextMiddleware(message)
} {
duration = mark.elapsedNow()
println()
}
}
}
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 levelsLockingMiddleware — Prevents concurrent message handling with a configurable timeoutCommands execute within a Unit of Work that manages three phases:
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<, MessageFailure> {
BusResult.success()
}
}
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<, MessageFailure> {
BusResult.success()
}
}
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> {
orderId = orderRepository.save(message.items)
paymentService.charge(orderId)
BusResult.success(orderId)
}
}
You can get the full code here.
The KSP processor generates:
// 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))
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.
KBUS includes base types for domain-driven design:
You can get the full code here.
See LICENSE for details.
// 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}")
}
}
| Message | Handler | Cardinality | Returns | Purpose |
|---|
Command<TResult> | CommandHandler | One handler per command | Yes (minimal data suggested) | State-modifying operations, executes within Unit of Work |
Query<TResult> | QueryHandler | One handler per query | Yes | Read-only operations |
Event | EventHandler | Multiple handlers per event | No | Notifications, side effects, eventual consistency |
| Dispatch Timing | FIRE_AND_FORGET | FAIL_FAST | CONTINUE_AND_AGGREGATE |
|---|
DispatchImmediatelyInTransaction(Before main work finishes) | ✅ Safe Errors logged; transaction continues. | ✅ Standard Throws immediately; rolls back DB. | ✅ Safe Collects all, throws at end; rolls back DB. |
DispatchAfterPrimaryWork(Before DB commit) | ✅ Safe Secondary work fails quietly; DB commits. | ✅ Safe Throws before commit; rolls back DB. | ✅ Safe Collects all, throws before commit; rolls back DB. |
DispatchAfterTransaction(After DB commit) | ✅ Standard Failures caught and sent to DLQ. | ❌ Dangerous Throws after transaction commits | ❌ Dangerous Throws after transaction commits |
// 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) {
/* ... */
}
}
AllDependencies — Interface listing all required dependencies (implement this to provide them)AllHandlers — Interface with factory methods for every handlerHandlerFactory — Factory that creates handlers with their dependencies resolvedCompileTimeLoadedMessageBus — A type-safe bus with strongly-typed execute, fetch, and observe methods for
each message typeAutoLoader — Auto-loading support for runtime handler registration// 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>()
| Platform | Targets |
|---|
| JVM | Java 17+ |
| JS | Node, Browser |
| WASM | Node, Browser |
| macOS | x64, ARM64 |
| iOS | x64, ARM64, Simulator ARM64 |
| Linux | x64, ARM64 |
| Windows | x64 |
Surfaced from shared tags and platforms — no rankings paid for.