|
A broker-agnostic integration event relay for Kotlin.
events.kt provides a simple, type-safe API for publishing and consuming
integration events without coupling your application to a specific message broker.
Define an integration event, specifying the topic it's published to and, optionally, it's name & key property path.
@Serializable
@IntegrationEvent("users", name = "created", key = "user.id")
data class UserCreated(val user: User, createdAt: Instant)Configure a publisher and create an event producer:
val publisher = KafkaPublisher(brokers = listOf("localhost:9092"))
val producer = Producer(publisher) {
serialization { json() }
}
val user = User(id = 123, name = "Jane Doe")
producer.publish { UserCreated(user, createdAt = Clock.System,now()) }Configure a consumer relay and create an event consumer:
val relay = KafkaConsumer(brokers = listOf("localhost:9092"), groupId = "my-group")
val consumer = Consumer(relay) {
serialization { json() }
}
consumer.subscribe<UserCreated> { event, metadata ->
println("User ${event.user.name} created at ${event.createdAt}")
}See the documentation for a complete guide on using the library.
events.kt provides a Gradle plugin that configures code generation for event definitions
and registration helper functions.
You can also use the library without the Gradle plugin by registering event definitions manually.
See the manual registration guide for details.
plugins {
id("io.github.briangits.events.integration") version "<version>"
}
intagrationEvents {
packageName = "com.exmaple.events"
loaderFunctionName = "registerGenratedDefintions"
}Generate the event definitions:
./gradlew :generateEventDefinitionsThe generated registration function can then be used by your producers and consumers:
// Producer
val producer = Producer(publisher) { /* Producer config */ }
producer.registerGeneratedDefinitions()
// Consumer
val consumer = Consumer(relay) { /* Consumer config */ }
consumer.registerGeneratedDefinitions()dependencies {
// Prodcer API
implementation("io.github.briangits.events:producer:<version>")
// Kafka Publihser
implementation("io.github.briangits.events:kafka-publisher:<version>")
}A publisher can be shared by one or more producers.
val publisher = KafkaPublisher(brokers = listOf("localhost:9092")) {
/* Kafka producer configurartin */
}See creating a publisher for a complete guide.
Create and configure a producer:
val producer = Producer(publisher) {
serialization { json() }
}You can now publish events:
val event = UserCreatd(user, createdAt = Clock.System.now())
producer.publish(event) {
// Metadata
"traceId" to "trace-123"
}dependencies {
// Consumer API
implementation("io.github.briangits.events:consumer:<version>")
// Kafka Consumer
implementation("io.github.briangits.events:kafka-consumer:<version>")
}A relay can be shared by one or more consumers
val relay = KafkaConsumer(brokers = listOf("localhost:9092"), groupId = "my-group") {
/* Kafka consumer configuration */
}See creating a relay for a complete guide.
Create and configure consumer:
val consumer = Consumer(relay) {
serialization { json() }
}Subscribe to an event:
consumer.subscribe<UserCreated> { event, metadata ->
println("User ${event.user.name} created at ${event.createdAt}")
}The relay can be started explicitly after all subscriptions have been configured.
This ensures that consumers are fully configured before message processing begins.
relay.start()inetgartion-events is designed to be broker-agnostic.
The library provides a unified API for publishing and consuming events, while allowing the underlying broker implementation to be swapped independently.
Currently supported brokers:
- Kafka - Kafka publisher and consumer implementations, currently only supported on JVM & Native targets