DEV Community: Nathan Fallet

Part 6: Test and Demo - Ktor Native Worker Tutorial

Nathan Fallet — Sun, 21 Dec 2025 22:45:39 +0000

In this final part, we'll explore how to set up, run, and test the complete application across different platforms.

Prerequisites

Before running the application, ensure you have:

RabbitMQ Server:
- Running locally or accessible via network
- Default connection: localhost:5672 with guest:guest
- You can run RabbitMQ using Docker:
```
  docker run -d --name rabbitmq \
    -p 5672:5672 \
    -p 15672:15672 \
    rabbitmq:3-management
```
Firebase Service Account:
- Download from Firebase Console → Project Settings → Service Accounts
- Save as firebase-admin-sdk.json in the project root
- Or specify custom path via SERVICE_ACCOUNT_PATH environment variable
FCM Device Token:
- Obtain from a mobile app integrated with Firebase Cloud Messaging
- Needed to test actual notification sending

Running the Application

Running on JVM (Development)

The JVM target is ideal for development:

./gradlew jvmRun

This command:

Compiles the Kotlin code for JVM
Starts the Ktor server on port 8080
Enables hot reload (in development mode)
Provides better logging output via Logback

Running on Native Platforms

For production deployment, use native executables:

macOS (ARM64):

./gradlew runDebugExecutableMacosArm64

Linux (x64):

./gradlew runDebugExecutableLinuxX64

Windows (x64):

./gradlew runDebugExecutableMingwX64

The application will:

Connect to RabbitMQ
Declare exchange and queue
Start consuming messages from the queue
Start the HTTP server on port 8080

You should see output similar to:

[main] INFO  Application - Application started
[main] INFO  Application - Responding at http://0.0.0.0:8080

Building Production Executables

To build optimized native executables:

./gradlew build

Executables will be located at:

macOS: build/bin/macosArm64/releaseExecutable/ktor-native-worker-tutorial.kexe
Linux: build/bin/linuxX64/releaseExecutable/ktor-native-worker-tutorial.kexe
Windows: build/bin/mingwX64/releaseExecutable/ktor-native-worker-tutorial.exe

These are standalone executables that can be deployed without a JVM.

Environment Configuration

Configure the application using environment variables:

export RABBITMQ_HOST=localhost
export RABBITMQ_PORT=5672
export RABBITMQ_USER=guest
export RABBITMQ_PASSWORD=guest
export SERVICE_ACCOUNT_PATH=/path/to/firebase-admin-sdk.json

Then run:

./gradlew jvmRun

Or for native executables:

export SERVICE_ACCOUNT_PATH=/path/to/firebase-admin-sdk.json
./build/bin/macosArm64/releaseExecutable/ktor-native-worker-tutorial.kexe

Testing the API

Using curl

Send a test notification request:

curl -X POST http://localhost:8080/api/notifications \
  -H "Content-Type: application/json" \
  -d '{
    "token": "your-fcm-device-token-here",
    "title": "Hello World",
    "body": "This is a test notification from Ktor Native!"
  }'

Expected response:

HTTP/1.1 200 OK

Using HTTPie

If you prefer HTTPie:

http POST localhost:8080/api/notifications \
  token=your-fcm-device-token-here \
  title="Hello World" \
  body="This is a test notification from Ktor Native!"

Using Postman

Create a new POST request
URL: http://localhost:8080/api/notifications
Headers: Content-Type: application/json
Body (raw JSON):

   {
     "token": "your-fcm-device-token-here",
     "title": "Hello World",
     "body": "This is a test notification from Ktor Native!"
   }

Send the request

Understanding the Flow

When you send a request, here's what happens:

HTTP Request Received:
- Ktor receives POST to /api/notifications
- Content negotiation deserializes JSON to NotificationPayload
Message Published:
- Route handler creates SendNotificationEvent
- Event serialized to JSON
- Published to RabbitMQ exchange with routing key
- HTTP 200 OK returned immediately
Message Queued:
- RabbitMQ routes message to the queue
- Message persisted to disk (durable queue)
Message Consumed:
- Application's consumer receives the message
- SendNotificationHandler invoked
Notification Sent:
- Handler deserializes SendNotificationEvent
- Calls NotificationService.sendNotification()
- FCM API sends push notification to device

Monitoring RabbitMQ

If you're running RabbitMQ with the management plugin:

Open http://localhost:15672 in your browser
Login with guest:guest
Navigate to "Queues" tab
You should see notifications_queue
View message rates, consumers, and queue depth

Verifying the Setup

Check RabbitMQ Connection

After starting the application, verify RabbitMQ connection:

In RabbitMQ Management UI, go to "Connections"
You should see an active connection from your application

Check Exchange and Queue

Go to "Exchanges" tab
Verify notifications_exchange exists (type: topic)
Go to "Queues" tab
Verify notifications_queue exists (durable: true)
Check that it's bound to the exchange with routing key notifications.send

Check Consumer

In the queue details for notifications_queue
Scroll to "Consumers" section
You should see 1 active consumer

Testing Error Scenarios

Invalid JSON

curl -X POST http://localhost:8080/api/notifications \
  -H "Content-Type: application/json" \
  -d '{"invalid": "data"}'

Expected: 400 Bad Request (missing required fields)

Missing Content-Type

curl -X POST http://localhost:8080/api/notifications \
  -d '{"token": "xxx", "title": "Hi", "body": "Test"}'

Expected: 415 Unsupported Media Type

Invalid FCM Token

curl -X POST http://localhost:8080/api/notifications \
  -H "Content-Type: application/json" \
  -d '{
    "token": "invalid-token",
    "title": "Test",
    "body": "This will fail"
  }'

Expected: 200 OK (request accepted), but notification sending will fail in the worker. Check application logs for FCM
error.

Performance Testing

Simple Load Test with curl

for i in {1..100}; do
  curl -X POST http://localhost:8080/api/notifications \
    -H "Content-Type: application/json" \
    -d "{\"token\": \"test-$i\", \"title\": \"Test $i\", \"body\": \"Message $i\"}" &
done
wait

This sends 100 concurrent requests. Monitor:

RabbitMQ queue depth
Application memory usage
Response times

Using Apache Bench

ab -n 1000 -c 10 -p payload.json -T application/json \
  http://localhost:8080/api/notifications

Create payload.json:

{
  "token": "test-token",
  "title": "Load Test",
  "body": "Testing"
}

Comparing JVM vs Native Performance

Startup Time

JVM:

time ./gradlew jvmRun

Native:

time ./build/bin/macosArm64/releaseExecutable/ktor-native-worker-tutorial.kexe

Native executables typically start much faster than JVM.

Memory Usage

Monitor memory usage while running:

On macOS/Linux:

# JVM
ps aux | grep java

# Native
ps aux | grep ktor-native-worker-tutorial

Native executables typically use less memory than JVM.

Request Throughput

Use Apache Bench or a similar tool to compare request handling performance between JVM and native builds.

Deployment Considerations

Docker Deployment

Create a Dockerfile for native builds:

FROM alpine:latest

# Install dependencies (if any needed by your native binary)
RUN apk add --no-cache libstdc++

# Copy the native executable
COPY build/bin/linuxX64/releaseExecutable/ktor-native-worker-tutorial.kexe /app/server

# Copy Firebase service account
COPY firebase-admin-sdk.json /app/

WORKDIR /app

# Set environment variables
ENV RABBITMQ_HOST=rabbitmq
ENV SERVICE_ACCOUNT_PATH=/app/firebase-admin-sdk.json

# Run the executable
CMD ["/app/server"]

Build and run:

docker build -t ktor-worker .
docker run -p 8080:8080 --link rabbitmq ktor-worker

Kubernetes Deployment

The native executable is perfect for Kubernetes:

Fast startup time
Low memory footprint
No JVM overhead
Small container image size

Scaling Workers

You can run multiple instances of the application:

They'll all consume from the same RabbitMQ queue
Messages are distributed among consumers
Enables horizontal scaling
Provides high availability

Troubleshooting

Application Won't Start

Check RabbitMQ Connection:

# Verify RabbitMQ is running
docker ps | grep rabbitmq

# Check connectivity
telnet localhost 5672

Check Firebase Service Account:

# Verify file exists
ls -la firebase-admin-sdk.json

# Verify it's valid JSON
cat firebase-admin-sdk.json | python -m json.tool

Messages Not Being Processed

Check Consumer Status:

Open RabbitMQ Management UI
Verify consumer is connected to the queue
Check application logs for errors

Check Queue Bindings:

Verify routing key matches: notifications.send
Ensure exchange type is topic

Notifications Not Received

Verify FCM Token:

Ensure the token is valid and current
FCM tokens can expire or change

Check Application Logs:

Look for FCM API errors
Verify service account has correct permissions

Summary

This tutorial covered:

Running the application on JVM and Native platforms
Building production-ready native executables
Configuring via environment variables
Testing the API with various tools
Monitoring RabbitMQ
Performance testing and comparison
Deployment strategies
Troubleshooting common issues

Complete Architecture Overview

Putting it all together:

Gradle Setup: Multiplatform build with JVM and Native targets
Notification Service: FCM integration using Flareon library
AMQP Setup: RabbitMQ integration with Kourier client
Routes: Type-safe HTTP endpoints with Ktor
Koin DI: Wiring all components together
Testing: Running and verifying the complete system

The result is a production-ready, native-compiled backend that:

Handles HTTP requests asynchronously
Processes notifications via message queue
Scales horizontally
Deploys with minimal resources
Follows clean code principles

Congratulations on completing the tutorial!

Part 5: Wire Everything with Koin - Ktor Native Worker Tutorial

Nathan Fallet — Sun, 21 Dec 2025 22:45:15 +0000

In this part, we'll explore how Koin dependency injection wires all the components together, creating a cohesive and
testable application architecture.

What is Koin?

Koin is a lightweight dependency injection framework for Kotlin that:

Works with Kotlin Multiplatform (JVM, Native, JS, etc.)
Uses a simple DSL for declaring dependencies
Provides compile-time safety with Kotlin's type system
Integrates seamlessly with Ktor

Koin Dependencies

In gradle/libs.versions.toml:

[versions]
koin = "4.1.0"

[libraries]
koin-core = { module = "io.insert-koin:koin-core", version.ref = "koin" }
koin-ktor = { module = "io.insert-koin:koin-ktor", version.ref = "koin" }

In build.gradle.kts:

sourceSets {
    commonMain.dependencies {
        implementation(libs.koin.core)
        implementation(libs.koin.ktor)
        // ... other dependencies
    }
}

Main Module Definition

File: src/commonMain/kotlin/me/nathanfallet/ktornativeworkertutorial/di/MainModule.kt

val Application.mainModule
get() = module {
    // Services
    single<MessageBroker> {
        RabbitMqMessageBroker(
            coroutineScope = this@mainModule,
            host = getEnv("RABBITMQ_HOST") ?: "localhost",
            port = getEnv("RABBITMQ_PORT")?.toIntOrNull() ?: 5672,
            user = getEnv("RABBITMQ_USER") ?: "guest",
            password = getEnv("RABBITMQ_PASSWORD") ?: "guest",
        )
    }
    single<NotificationService> {
        NotificationServiceImpl(
            serviceAccountPath = getEnv("SERVICE_ACCOUNT_PATH") ?: "firebase-admin-sdk.json",
        )
    }

    // Handlers
    single { SendNotificationHandler(get()) }

    // Routes
    single { RoutesDependencies(get()) }
}

Understanding the Module Structure

Extension Property on Application:
- val Application.mainModule makes the module available as a property
- Provides access to the Application scope within the module
- Enables coroutine scope sharing with the application
Module DSL:
- module { } is Koin's DSL for defining a dependency module
- All service definitions go inside this block
Single vs Factory:
- single<T> { } creates a singleton (one instance for the entire application)
- All dependencies here use single for resource efficiency

Service Definitions

MessageBroker:

single<MessageBroker> {
    RabbitMqMessageBroker(
        coroutineScope = this@mainModule,
        host = getEnv("RABBITMQ_HOST") ?: "localhost",
        port = getEnv("RABBITMQ_PORT")?.toIntOrNull() ?: 5672,
        user = getEnv("RABBITMQ_USER") ?: "guest",
        password = getEnv("RABBITMQ_PASSWORD") ?: "guest",
    )
}

Defines MessageBroker interface binding to RabbitMqMessageBroker implementation
Uses this@mainModule to share the Application's coroutine scope
Reads configuration from environment variables with sensible defaults
getEnv() is a platform-specific function (works on JVM and Native)

NotificationService:

single<NotificationService> {
    NotificationServiceImpl(
        serviceAccountPath = getEnv("SERVICE_ACCOUNT_PATH") ?: "firebase-admin-sdk.json",
    )
}

Binds NotificationService interface to NotificationServiceImpl
Service account path configurable via environment variable
Defaults to firebase-admin-sdk.json in the project root

Handler Definitions

SendNotificationHandler:

single { SendNotificationHandler(get()) }

Creates the message handler for processing notification events
get() automatically resolves and injects NotificationService
Koin's type inference determines which dependency to inject

Route Dependencies

RoutesDependencies:

single { RoutesDependencies(get()) }

Creates the dependencies container for routes
get() resolves and injects MessageBroker
Routes can access dependencies through this container

Koin Configuration

File: src/commonMain/kotlin/me/nathanfallet/ktornativeworkertutorial/config/Koin.kt

fun Application.configureKoin() {
    install(Koin) {
        modules(mainModule)
    }
}

This function:

Installs the Koin plugin into the Ktor application
Registers the mainModule containing all dependency definitions
Must be called before other configuration functions that use dependencies

Using Koin in Configuration

In Message Broker Setup (src/commonMain/kotlin/me/nathanfallet/ktornativeworkertutorial/config/MessageBroker.kt):

fun Application.configureMessageBroker() = runBlocking {
    val messageBroker by inject<MessageBroker>()
    messageBroker.initialize()
    messageBroker.startConsuming(Constants.RABBITMQ_QUEUE, get<SendNotificationHandler>())
}

Two ways to get dependencies:

by inject<T>(): Lazy property delegation
get<T>(): Direct retrieval

In Routing Setup (src/commonMain/kotlin/me/nathanfallet/ktornativeworkertutorial/config/Routing.kt):

fun Application.configureRouting() {
    routing {
        registerRoutes(get())
    }
}

get() without type parameter uses type inference
Resolves RoutesDependencies based on the parameter type of registerRoutes()

Application Bootstrap

File: src/commonMain/kotlin/me/nathanfallet/ktornativeworkertutorial/Application.kt

suspend fun Application.module() {
    configureKoin()
    configureMessageBroker()
    configureSerialization()
    configureRouting()
}

Order is critical:

configureKoin(): Must be first - sets up the DI container
configureMessageBroker(): Uses Koin to get MessageBroker and handlers
configureSerialization(): Sets up JSON handling (no dependencies)
configureRouting(): Uses Koin to get RoutesDependencies

Environment Configuration

The project uses environment variables for configuration, accessed through getEnv():

Variable	Default	Purpose
`RABBITMQ_HOST`	`localhost`	RabbitMQ server hostname
`RABBITMQ_PORT`	`5672`	RabbitMQ server port
`RABBITMQ_USER`	`guest`	RabbitMQ username
`RABBITMQ_PASSWORD`	`guest`	RabbitMQ password
`SERVICE_ACCOUNT_PATH`	`firebase-admin-sdk.json`	Path to Firebase service account JSON

This approach follows the 12-factor app methodology for configuration management.

Dependency Graph

Here's the complete dependency graph:

Application
└── MainModule
    ├── MessageBroker (RabbitMqMessageBroker)
    │   └── coroutineScope (from Application)
    ├── NotificationService (NotificationServiceImpl)
    │   └── serviceAccountPath (from environment)
    ├── SendNotificationHandler
    │   └── NotificationService ← injected
    └── RoutesDependencies
        └── MessageBroker ← injected

Benefits of Dependency Injection

Testability:
- Easy to mock dependencies in tests
- Can create test modules with mock implementations
- No global state or singletons to reset
Modularity:
- Components are loosely coupled
- Can swap implementations without changing consumers
- Clear contracts via interfaces
Configuration Management:
- Centralized dependency configuration
- Easy to see all service instantiations
- Simple to add new dependencies
Type Safety:
- Compile-time dependency resolution
- IDE support for refactoring
- Clear error messages for missing dependencies
Multiplatform Support:
- Koin works across all Kotlin platforms
- Same DI code for JVM and Native
- Platform-specific implementations via expect/actual

Common Patterns

Interface Binding:

single<Interface> { ConcreteImplementation(...) }

Constructor Injection:

single { SomeService(get(), get()) }

Named Dependencies (if needed):

single(named("primary")) { PrimaryService() }
single(named("secondary")) { SecondaryService() }

Lazy Injection:

val service by inject<Service>()

Direct Retrieval:

val service = get<Service>()

Summary

The Koin integration demonstrates:

Clean dependency injection with minimal boilerplate
Environment-based configuration with defaults
Interface-based design for flexibility
Type-safe dependency resolution
Multiplatform compatibility
Centralized service lifecycle management

In the final part, we'll explore how to test and demo the complete application.

Part 4: Routes - Ktor Native Worker Tutorial

Nathan Fallet — Sun, 21 Dec 2025 22:44:46 +0000

In this part, we'll explore how to define HTTP routes in Ktor and how they integrate with the message broker to handle
notification requests asynchronously.

Route Payload

File: src/commonMain/kotlin/me/nathanfallet/ktornativeworkertutorial/routes/NotificationPayload.kt

@Serializable
data class NotificationPayload(
    val title: String,
    val body: String,
    val token: String,
)

This data class represents the expected JSON payload for notification requests. The @Serializable annotation enables
automatic JSON serialization/deserialization with Kotlinx Serialization.

Expected JSON format:

{
  "title": "Notification Title",
  "body": "Notification message body",
  "token": "FCM device token"
}

Routes Dependencies

File: src/commonMain/kotlin/me/nathanfallet/ktornativeworkertutorial/routes/Routes.kt

data class RoutesDependencies(
    val messageBroker: MessageBroker,
)

This pattern follows the Dependency Injection principle by explicitly declaring what dependencies the routes need.
Instead of using global state or service locators within route handlers, all dependencies are passed as a single
parameter.

Benefits:

Clear visibility of route dependencies
Easy to test (can mock dependencies)
Type-safe dependency access
Clean separation of concerns

Route Registration

File: src/commonMain/kotlin/me/nathanfallet/ktornativeworkertutorial/routes/Routes.kt

fun Route.registerRoutes(dependencies: RoutesDependencies) = with(dependencies) {
    post<NotificationPayload>("/api/notifications") { payload ->
        val event = SendNotificationEvent(
            title = payload.title,
            body = payload.body,
            token = payload.token,
        )
        messageBroker.publish(
            Constants.RABBITMQ_EXCHANGE, Constants.RABBITMQ_ROUTING_KEY,
            Serialization.json.encodeToString(event),
        )
        call.respond(HttpStatusCode.OK)
    }
}

Understanding the Route

Extension Function:
- Route.registerRoutes() is an extension function on Ktor's Route
- Allows modular route registration
- Can be called from the routing configuration
Dependency Scope:
- with(dependencies) creates a scope where dependency properties are directly accessible
- Provides clean access to messageBroker without prefixing
Type-Safe POST Handler:
- post<NotificationPayload>("/api/notifications") defines a POST endpoint
- Ktor automatically deserializes the request body to NotificationPayload
- Type safety ensures compile-time checking of payload structure
Event Creation:
- Maps the HTTP payload to a SendNotificationEvent
- This separation allows different internal/external representations
- The event is what gets published to RabbitMQ
Message Publishing:
- Serializes the event to JSON using Serialization.json.encodeToString()
- Publishes to the RabbitMQ exchange with the routing key
- Message will be queued and processed asynchronously
Response:
- Returns HTTP 200 OK immediately
- Client doesn't wait for the notification to be sent
- Improves response time and user experience

Integration with Ktor Application

File: src/commonMain/kotlin/me/nathanfallet/ktornativeworkertutorial/config/Routing.kt

fun Application.configureRouting() {
    routing {
        registerRoutes(get())
    }
}

This configuration function:

Is an extension on Application
Calls routing {} to set up routing
Calls registerRoutes() with dependencies from Koin using get()
Ktor's Koin integration automatically resolves the RoutesDependencies

Content Negotiation Configuration

File: src/commonMain/kotlin/me/nathanfallet/ktornativeworkertutorial/config/Serialization.kt

fun Application.configureSerialization() {
    install(ContentNegotiation) {
        json(Serialization.json)
    }
}

This enables automatic JSON handling:

Deserializes incoming JSON to Kotlin objects (request bodies)
Serializes Kotlin objects to JSON (responses)
Uses the same Serialization.json instance for consistency

The ContentNegotiation plugin makes the type-safe post<NotificationPayload> syntax possible.

Application Module Setup

File: src/commonMain/kotlin/me/nathanfallet/ktornativeworkertutorial/Application.kt

suspend fun Application.module() {
    configureKoin()
    configureMessageBroker()
    configureSerialization()
    configureRouting()
}

Configuration order matters:

configureKoin(): Set up dependency injection first
configureMessageBroker(): Initialize message broker
configureSerialization(): Enable JSON handling
configureRouting(): Register routes (depends on all above)

Request Flow

Here's what happens when a notification request is received:

1. HTTP POST → /api/notifications
2. Content Negotiation deserializes JSON → NotificationPayload
3. Route handler creates SendNotificationEvent
4. Event serialized to JSON
5. Message published to RabbitMQ
6. HTTP 200 OK returned to client
7. (Async) RabbitMQ consumer receives message
8. (Async) SendNotificationHandler processes event
9. (Async) NotificationService sends FCM notification

Error Handling

The current implementation has minimal error handling:

Ktor automatically returns 400 Bad Request for invalid JSON
Missing required fields result in serialization errors
Type mismatches are caught during deserialization

In a production environment, you might want to add:

Validation of FCM token format
Rate limiting
Authentication/authorization
Custom error responses
Logging

Example Request

Using curl:

curl -X POST http://localhost:8080/api/notifications \
  -H "Content-Type: application/json" \
  -d '{
    "token": "fcm-device-token-here",
    "title": "Hello World",
    "body": "This is a test notification"
  }'

Response:

HTTP/1.1 200 OK

Benefits of This Architecture

Async Processing:
- HTTP response is immediate
- Notification sending happens in the background
- Better user experience and API performance
Decoupling:
- HTTP layer is separate from notification logic
- Can change notification implementation without touching routes
- Message broker provides a clear boundary
Scalability:
- Can scale HTTP servers independently from workers
- Queue provides buffering during traffic spikes
- Workers can be added/removed dynamically
Reliability:
- Messages are persisted in RabbitMQ
- HTTP failures don't lose notification requests
- Can retry failed notifications
Clean Code:
- Type-safe route handlers
- Explicit dependencies
- Clear separation of concerns
- Easy to test

Summary

The routing layer demonstrates:

Type-safe HTTP endpoint definition with Ktor
Automatic JSON serialization/deserialization
Dependency injection for route handlers
Asynchronous message publishing
Clean separation between HTTP and business logic

In the next part, we'll explore how Koin dependency injection wires everything together.

Part 3: AMQP Setup and Calls - Ktor Native Worker Tutorial

Nathan Fallet — Sun, 21 Dec 2025 22:44:09 +0000

In this part, we'll explore how to set up AMQP (Advanced Message Queuing Protocol) with RabbitMQ for asynchronous
message processing. This enables a worker pattern where HTTP requests queue messages that are processed asynchronously.

Architecture Overview

The messaging system consists of:

MessageBroker: Interface defining message broker operations
RabbitMqMessageBroker: RabbitMQ implementation using the Kourier AMQP client
MessageHandler: Interface for message handlers
SendNotificationHandler: Handler for processing notification events
SendNotificationEvent: Data class representing notification events

Constants Configuration

File: src/commonMain/kotlin/me/nathanfallet/ktornativeworkertutorial/Constants.kt

object Constants {

    const val RABBITMQ_EXCHANGE = "notifications_exchange"
    const val RABBITMQ_QUEUE = "notifications_queue"
    const val RABBITMQ_ROUTING_KEY = "notifications.send"

}

These constants define:

Exchange name for routing messages
Queue name for storing messages
Routing key for binding the queue to the exchange

MessageBroker Interface

File: src/commonMain/kotlin/me/nathanfallet/ktornativeworkertutorial/messaging/MessageBroker.kt

interface MessageBroker {
    suspend fun initialize()
    suspend fun publish(exchange: String, routingKey: String, message: String)
    suspend fun startConsuming(queue: String, handler: MessageHandler)
}

This interface defines three core operations:

initialize(): Set up the connection and declare exchanges/queues
publish(): Send a message to an exchange with a routing key
startConsuming(): Start consuming messages from a queue with a handler

RabbitMQ Implementation

File: src/commonMain/kotlin/me/nathanfallet/ktornativeworkertutorial/messaging/RabbitMqMessageBroker.kt

class RabbitMqMessageBroker(
    private val coroutineScope: CoroutineScope,
    private val host: String,
    private val port: Int,
    private val user: String,
    private val password: String,
) : MessageBroker {
    private lateinit var amqpConnection: AMQPConnection
    private lateinit var amqpChannel: AMQPChannel

    override suspend fun initialize() {
        amqpConnection = createRobustAMQPConnection(coroutineScope) {
            server {
                host = this@RabbitMqMessageBroker.host
                port = this@RabbitMqMessageBroker.port
                user = this@RabbitMqMessageBroker.user
                password = this@RabbitMqMessageBroker.password
            }
        }

        amqpChannel = amqpConnection.openChannel().also {
            it.exchangeDeclare {
                name = Constants.RABBITMQ_EXCHANGE
                type = BuiltinExchangeType.TOPIC
            }
            it.queueDeclare {
                name = Constants.RABBITMQ_QUEUE
                durable = true
            }
            it.queueBind {
                queue = Constants.RABBITMQ_QUEUE
                exchange = Constants.RABBITMQ_EXCHANGE
                routingKey = Constants.RABBITMQ_ROUTING_KEY
            }
        }
    }

    override suspend fun publish(exchange: String, routingKey: String, message: String) {
        amqpChannel.basicPublish(
            body = message.toByteArray(),
            exchange = exchange,
            routingKey = routingKey,
            properties = properties {
                deliveryMode = 2u // Make message persistent
            }
        )
    }

    override suspend fun startConsuming(queue: String, handler: MessageHandler) {
        amqpChannel.basicConsume(
            queue = queue,
            noAck = false,
            onDelivery = { delivery ->
                handler(delivery.message.routingKey, delivery.message.body.decodeToString())
                amqpChannel.basicAck(delivery.message.deliveryTag, false)
            }
        )
    }
}

Initialization Flow

Connection Creation:
- Uses createRobustAMQPConnection() for automatic reconnection
- Configured with RabbitMQ server credentials
- Scoped to a coroutine scope for lifecycle management
Channel Setup:
- Opens an AMQP channel on the connection
- Declares a TOPIC exchange for flexible routing
- Declares a durable queue (survives server restarts)
- Binds the queue to the exchange with a routing key
Publishing Messages:
- Converts message string to bytes
- Sets delivery mode to 2 (persistent messages)
- Routes message via exchange and routing key
Consuming Messages:
- Sets up a consumer on the queue
- Uses manual acknowledgment (noAck = false)
- Delegates message handling to the provided handler
- Acknowledges message after successful processing (basicAck)

Using the Kourier AMQP Client

The project uses the Kourier AMQP client (dev.kourier:amqp-client-robust), which
provides:

Kotlin Multiplatform support
Native compatibility (JVM and Native targets)
Robust connection with automatic reconnection
Coroutine-based API

Dependency Configuration

In build.gradle.kts:

sourceSets {
    commonMain.dependencies {
        implementation(libs.amqp)
        // ... other dependencies
    }
}

And in gradle/libs.versions.toml:

[versions]
amqp = "0.4.0"

[libraries]
amqp = { module = "dev.kourier:amqp-client-robust", version.ref = "amqp" }

Message Handler Interface

File: src/commonMain/kotlin/me/nathanfallet/ktornativeworkertutorial/messaging/MessageHandler.kt

interface MessageHandler {
    suspend operator fun invoke(routingKey: String, body: String)
}

The handler uses the invoke operator, allowing instances to be called like functions. It receives:

routingKey: The routing key of the message
body: The message body as a string

Notification Event and Handler

Event Data Class (
src/commonMain/kotlin/me/nathanfallet/ktornativeworkertutorial/messaging/handlers/SendNotificationEvent.kt):

@Serializable
data class SendNotificationEvent(
    val title: String,
    val body: String,
    val token: String,
)

Handler Implementation (
src/commonMain/kotlin/me/nathanfallet/ktornativeworkertutorial/messaging/handlers/SendNotificationHandler.kt):

class SendNotificationHandler(
    private val notificationService: NotificationService,
) : MessageHandler {
    override suspend fun invoke(routingKey: String, body: String) {
        val event = Serialization.json.decodeFromString<SendNotificationEvent>(body)
        notificationService.sendNotification(
            token = event.token,
            title = event.title,
            body = event.body,
        )
    }
}

Handler Flow

Receives the message body as JSON string
Deserializes to SendNotificationEvent using Kotlinx Serialization
Calls the NotificationService to send the notification
The service is injected via constructor (dependency injection)

Serialization Setup

File: src/commonMain/kotlin/me/nathanfallet/ktornativeworkertutorial/Serialization.kt

object Serialization {
    val json = Json {
        ignoreUnknownKeys = true
        explicitNulls = false
    }
}

This configured JSON instance is used for:

Serializing events before publishing to RabbitMQ
Deserializing events when consuming messages
Consistent JSON handling across the application

Configuration:

ignoreUnknownKeys = true: Allows adding fields without breaking old consumers
explicitNulls = false: Omits null values from JSON output

Integration with Application

File: src/commonMain/kotlin/me/nathanfallet/ktornativeworkertutorial/config/MessageBroker.kt

fun Application.configureMessageBroker() = runBlocking {
    val messageBroker by inject<MessageBroker>()
    messageBroker.initialize()
    messageBroker.startConsuming(Constants.RABBITMQ_QUEUE, get<SendNotificationHandler>())
}

This configuration function:

Injects the MessageBroker instance from Koin
Initializes the connection and channel
Starts consuming messages with the SendNotificationHandler

Environment Configuration

The RabbitMQ connection is configured via environment variables in the Koin module:

single<MessageBroker> {
    RabbitMqMessageBroker(
        coroutineScope = this@mainModule,
        host = getEnv("RABBITMQ_HOST") ?: "localhost",
        port = getEnv("RABBITMQ_PORT")?.toIntOrNull() ?: 5672,
        user = getEnv("RABBITMQ_USER") ?: "guest",
        password = getEnv("RABBITMQ_PASSWORD") ?: "guest",
    )
}

Default values:

Host: localhost
Port: 5672
User: guest
Password: guest

Message Flow

HTTP request arrives at the API endpoint
Route publishes a SendNotificationEvent to RabbitMQ
Message is stored in the queue
Consumer receives the message
SendNotificationHandler deserializes and processes it
NotificationService sends the FCM notification

This architecture provides:

Decoupling of HTTP request handling from notification sending
Reliability through message persistence
Scalability through async processing
Error isolation (notification failures don't fail HTTP requests)

Summary

The AMQP setup demonstrates:

Clean interface-based design for message brokers
RabbitMQ integration with Kourier AMQP client
Message handler pattern for processing events
Serialization with Kotlinx Serialization
Environment-based configuration
Worker pattern for async processing

In the next part, we'll explore how to define HTTP routes that publish messages to this message broker.

Part 2: Sending Notifications - Ktor Native Worker Tutorial

Nathan Fallet — Sun, 21 Dec 2025 22:43:24 +0000

In this part, we'll explore how to send Firebase Cloud Messaging (FCM) notifications using the Flareon library in a
Kotlin Multiplatform environment.

Overview

The notification system in this project follows the Service pattern with a clean interface-implementation structure:

NotificationService: Interface defining the contract
NotificationServiceImpl: Implementation using Flareon library

This design follows clean code principles by separating the interface from the implementation, making it easy to test
and swap implementations if needed.

Notification Service Interface

File: src/commonMain/kotlin/me/nathanfallet/ktornativeworkertutorial/services/NotificationService.kt

interface NotificationService {
    suspend fun sendNotification(token: String, title: String, body: String)
}

This simple interface defines a single suspend function for sending notifications. It accepts:

token: The FCM device token
title: The notification title
body: The notification body/message

Notification Service Implementation

File: src/commonMain/kotlin/me/nathanfallet/ktornativeworkertutorial/services/NotificationServiceImpl.kt

class NotificationServiceImpl(
    serviceAccountPath: String,
) : NotificationService {

    private val serviceAccountJson = readFile(serviceAccountPath)
    private val credentials = GoogleCredentials.fromJson(serviceAccountJson)
    private val messaging = FcmService(credentials)

    override suspend fun sendNotification(token: String, title: String, body: String) {
        messaging.sendNotification(token, title, body)
    }

}

How It Works

Service Account Loading:
- The constructor accepts a serviceAccountPath parameter
- Uses the platform-specific readFile() function to load the JSON file
- Creates GoogleCredentials from the JSON content
FCM Service Initialization:
- Creates an FcmService instance using the credentials
- This service is reused for all notification sending operations
Sending Notifications:
- The sendNotification() method delegates to the Flareon library's FcmService
- The method is marked as suspend, enabling coroutine-based async operations

Using the Flareon Library

The project uses the Flareon library (
me.nathanfallet.flareon:messaging), which provides:

Kotlin Multiplatform support for FCM
Native compatibility (works on JVM and Native targets)
Simple API for sending notifications
Google service account authentication

Dependency Configuration

In build.gradle.kts, Flareon is included in the common dependencies:

sourceSets {
    commonMain.dependencies {
        implementation(libs.flareon.messaging)
        // ... other dependencies
    }
}

And in gradle/libs.versions.toml:

[versions]
flareon = "0.1.1"

[libraries]
flareon-messaging = { module = "me.nathanfallet.flareon:messaging", version.ref = "flareon" }

Firebase Service Account Setup

To use FCM, you need a Firebase service account JSON file:

Go to the Firebase Console
Select your project
Navigate to Project Settings > Service Accounts
Click "Generate New Private Key"
Save the JSON file (commonly named firebase-admin-sdk.json)
Place it in your project root or specify its path via environment variable

The service account JSON contains:

Project ID
Private key for authentication
Client email
Other Firebase project metadata

Platform-Specific File Reading

Notice how the implementation uses readFile(), which is an expect function with platform-specific implementations:

JVM: Uses java.io.File.readText()
Native: Uses POSIX file operations (fopen, fread, fclose)

This demonstrates how Kotlin Multiplatform allows you to write common business logic while using platform-specific APIs
when necessary.

Error Handling Considerations

The current implementation is straightforward and doesn't include explicit error handling. In a production environment,
you might want to add:

Error handling for missing or invalid service account files
Retry logic for failed notification sends
Logging for debugging
Validation of notification parameters

However, this tutorial focuses on the clean, minimal implementation to demonstrate the core concepts.

Integration with Dependency Injection

The NotificationService is registered in the Koin dependency injection container (covered in Part 5), making it
available throughout the application:

single<NotificationService> {
    NotificationServiceImpl(
        serviceAccountPath = getEnv("SERVICE_ACCOUNT_PATH") ?: "firebase-admin-sdk.json",
    )
}

This allows the service account path to be configured via environment variable, with a sensible default fallback.

Summary

The notification service demonstrates:

Clean separation of interface and implementation
Use of Kotlin Multiplatform libraries (Flareon)
Platform-specific code integration (file reading)
Coroutine-based async operations
Dependency injection compatibility

In the next part, we'll explore how to set up AMQP message brokering with RabbitMQ to enable asynchronous notification
processing.

Part 1: Gradle Setup - Ktor native worker tutorial

Nathan Fallet — Sun, 21 Dec 2025 22:42:50 +0000

In this tutorial, we'll explore how to set up a Kotlin Multiplatform project targeting both JVM and Native platforms (
Linux, macOS, Windows) using Ktor for building a backend server.

Project Structure

The project follows a standard Kotlin Multiplatform structure:

settings.gradle.kts - Project settings
build.gradle.kts - Build configuration with multiplatform setup
gradle/libs.versions.toml - Centralized dependency version management

Version Catalog (libs.versions.toml)

The project uses Gradle's version catalog feature to manage dependencies centrally. Here's how it's configured:

File: gradle/libs.versions.toml

[versions]
kotlin = "2.2.21"
ktor = "3.3.3"
koin = "4.1.0"
amqp = "0.4.0"
flareon = "0.1.1"
logback = "1.5.18"

[plugins]
multiplatform = { id = "org.jetbrains.kotlin.multiplatform", version.ref = "kotlin" }
serialization = { id = "org.jetbrains.kotlin.plugin.serialization", version.ref = "kotlin" }

[libraries]
ktor-serialization-kotlinx-json = { module = "io.ktor:ktor-serialization-kotlinx-json", version.ref = "ktor" }
ktor-server-core = { module = "io.ktor:ktor-server-core", version.ref = "ktor" }
ktor-server-cio = { module = "io.ktor:ktor-server-cio", version.ref = "ktor" }
ktor-server-content-negotiation = { module = "io.ktor:ktor-server-content-negotiation", version.ref = "ktor" }
koin-core = { module = "io.insert-koin:koin-core", version.ref = "koin" }
koin-ktor = { module = "io.insert-koin:koin-ktor", version.ref = "koin" }
amqp = { module = "dev.kourier:amqp-client-robust", version.ref = "amqp" }
flareon-messaging = { module = "me.nathanfallet.flareon:messaging", version.ref = "flareon" }
logback-core = { module = "ch.qos.logback:logback-core", version.ref = "logback" }
logback-classic = { module = "ch.qos.logback:logback-classic", version.ref = "logback" }

This centralized approach makes it easy to update versions across the entire project.

Build Configuration (build.gradle.kts)

File: build.gradle.kts

plugins {
    alias(libs.plugins.multiplatform)
    alias(libs.plugins.serialization)
}

group = "me.nathanfallet.ktornativeworkertutorial"
version = "1.0"

repositories {
    mavenCentral()
}

kotlin {
    // JVM target for development and testing
    jvm {
        mainRun {
            mainClass.set("me.nathanfallet.ktornativeworkertutorial.ApplicationKt")
        }
    }

    // Native targets for production deployment
    listOf(
        linuxX64(),
        macosArm64(),
        mingwX64()
    ).forEach {
        it.binaries {
            executable {
                entryPoint = "me.nathanfallet.ktornativeworkertutorial.main"
            }
        }
    }

    sourceSets {
        commonMain.dependencies {
            implementation(libs.ktor.server.core)
            implementation(libs.ktor.server.cio)
            implementation(libs.ktor.server.content.negotiation)
            implementation(libs.ktor.serialization.kotlinx.json)
            implementation(libs.koin.core)
            implementation(libs.koin.ktor)
            implementation(libs.amqp)
            implementation(libs.flareon.messaging)
        }
        jvmMain.dependencies {
            // Required for logging to appear on JVM
            implementation(libs.logback.core)
            implementation(libs.logback.classic)
        }
    }
}

Key Configuration Points

Plugins:
- multiplatform: Enables Kotlin Multiplatform support
- serialization: Enables Kotlinx Serialization for JSON handling
JVM Target:
- Configured with mainRun to specify the main class
- Useful for development and testing
- Entry point: me.nathanfallet.ktornativeworkertutorial.ApplicationKt
Native Targets:
- Linux x64 (linuxX64())
- macOS ARM64 (macosArm64())
- Windows x64 (mingwX64())
- Each configured with an executable binary
- Entry point: me.nathanfallet.ktornativeworkertutorial.main
Source Sets:
- commonMain: Dependencies shared across all platforms
- jvmMain: JVM-specific dependencies (Logback for logging)

Project Settings (settings.gradle.kts)

File: settings.gradle.kts

plugins {
    id("org.gradle.toolchains.foojay-resolver-convention") version "0.8.0"
}
rootProject.name = "ktor-native-worker-tutorial"

This configuration enables automatic JDK provisioning through the Foojay toolchain resolver.

Platform-Specific Code with expect/actual

To handle platform-specific implementations (like reading environment variables or files), the project uses Kotlin's
expect/actual mechanism.

Common Interface (src/commonMain/kotlin/me/nathanfallet/ktornativeworkertutorial/Env.kt):

expect fun getEnv(name: String): String?
expect fun readFile(path: String): String

JVM Implementation (src/jvmMain/kotlin/me/nathanfallet/ktornativeworkertutorial/Env.jvm.kt):

actual fun getEnv(name: String): String? {
    return System.getenv(name)
}

actual fun readFile(path: String): String {
    return java.io.File(path).readText()
}

Native Implementation (src/nativeMain/kotlin/me/nathanfallet/ktornativeworkertutorial/Env.native.kt):

@OptIn(ExperimentalForeignApi::class)
actual fun getEnv(name: String): String? {
    return getenv(name)?.toKString()
}

@OptIn(ExperimentalForeignApi::class, UnsafeNumber::class)
actual fun readFile(path: String): String {
    val file = fopen(path, "r") ?: throw IllegalArgumentException("Cannot open file: $path")
    try {
        fseek(file, 0, SEEK_END)
        val size = ftell(file)
        fseek(file, 0, SEEK_SET)

        return buildString {
            val buffer = ByteArray(1024)
            while (true) {
                val read = fread(buffer.refTo(0), 1u, buffer.size.toULong(), file).toInt()
                if (read <= 0) break
                append(buffer.decodeToString(0, read))
            }
        }
    } finally {
        fclose(file)
    }
}

This approach allows the same common code to work across all platforms while using platform-specific APIs under the
hood.

Running the Project

With this Gradle setup, you can run the project on different platforms:

# JVM (for development/testing)
./gradlew jvmRun

# Native platforms
./gradlew runDebugExecutableMacosArm64   # macOS
./gradlew runDebugExecutableLinuxX64     # Linux
./gradlew runDebugExecutableMingwX64     # Windows

Building Native Executables

To build optimized native executables for production:

./gradlew build

The executables will be located in build/bin/{platform}/releaseExecutable/.

Summary

This Gradle setup provides:

Multiplatform support (JVM + Native)
Centralized dependency management via version catalogs
Platform-specific implementations using expect/actual
Easy development with JVM and production deployment with native binaries
Clean code organization following Kotlin conventions

In the next part, we'll explore how to implement notification sending using Firebase Cloud Messaging.

Stop wrapping your RabbitMQ code in runBlocking

Nathan Fallet — Mon, 01 Dec 2025 07:00:00 +0000

You're using Kotlin. You love coroutines. Your entire codebase is suspend functions, Flow, and structured concurrency.

Then you need to consume messages from RabbitMQ.

And suddenly, you're writing this:

runBlocking {
    processMessage(message)
}

Congratulations. You just mass-deployed an anti-pattern.

The problem nobody warns you about

The official RabbitMQ Java client was designed in 2007. Kotlin coroutines were released in 2018. These two worlds don't mix well.

Here's what consuming messages looks like with the Java client:

val factory = ConnectionFactory().apply {
    host = "localhost"
    username = "guest"
    password = "guest"
}

val connection = factory.newConnection()
val channel = connection.createChannel()

channel.basicConsume("orders", false, object : DefaultConsumer(channel) {
    override fun handleDelivery(
        consumerTag: String?,
        envelope: Envelope?,
        properties: AMQP.BasicProperties?,
        body: ByteArray?
    ) {
        val message = String(body!!, Charsets.UTF_8)

        // You want to call a suspend function here
        // But you can't. This is a callback on a Java thread.

        processOrder(message) // Compile error if suspend

        channel.basicAck(envelope!!.deliveryTag, false)
    }
})

See the problem? handleDelivery is a regular function called by the Java client's internal thread pool. You can't just sprinkle suspend on it.

The "solution" everyone uses

So you reach for runBlocking:

override fun handleDelivery(
    consumerTag: String?,
    envelope: Envelope?,
    properties: AMQP.BasicProperties?,
    body: ByteArray?
) {
    val message = String(body!!, Charsets.UTF_8)

    runBlocking {
        processOrder(message)  // Compiles!
        updateInventory(message)
        notifyShipping(message)
    }

    channel.basicAck(envelope!!.deliveryTag, false)
}

It compiles. It runs. You ship it.

But here's what you actually did:

Blocked a thread — runBlocking parks the current thread until all coroutines complete
Defeated the purpose of coroutines — You're not saving resources; you're using MORE (coroutine overhead + blocked thread)
Created backpressure issues — The Java client has a limited thread pool. Block those threads, and message consumption stalls.

You wanted non-blocking message processing. You got blocking message processing with extra steps.

"But I use `launch` instead!"

Some developers try to be clever:

val scope = CoroutineScope(Dispatchers.Default)

override fun handleDelivery(
    consumerTag: String?,
    envelope: Envelope?,
    properties: AMQP.BasicProperties?,
    body: ByteArray?
) {
    val message = String(body!!, Charsets.UTF_8)

    scope.launch {
        processOrder(message)
        updateInventory(message)
        notifyShipping(message)
    }

    channel.basicAck(envelope!!.deliveryTag, false)  // ACK before processing!
}

Now you have a different problem: you're acknowledging the message before processing completes. If your app crashes mid-processing, that message is gone forever.

Move the ACK inside the coroutine? Now you're calling channel.basicAck() from a different thread than the one that received the delivery — and the Java client's channels are not thread-safe.

There's no clean way out.

What coroutine-first actually looks like

Here's the same consumer with Kourier:

fun main() = runBlocking {
    val connection = createAMQPConnection(this, "amqp://guest:guest@localhost:5672/")
    val channel = connection.openChannel()

    channel.queueDeclare("orders", durable = true)
    val consumer = channel.basicConsume("orders")

    for (delivery in consumer) {
        val message = delivery.message.body.decodeToString()

        processOrder(message)      // suspend function - works!
        updateInventory(message)   // suspend function - works!
        notifyShipping(message)    // suspend function - works!

        channel.basicAck(delivery.message)
    }
}

Wait — there's a runBlocking at the top! Am I a hypocrite?

No. That's just the entry point — your main function needs somewhere to start the coroutine world. The difference is inside the loop: every suspend function you call actually suspends. The thread isn't blocked while processOrder waits for a database response. Other coroutines can run. That's the whole point.

With the Java client's callback, you're stuck in a non-suspend function. Here, you're in a proper coroutine context where suspension works as intended.

No callbacks. No thread-safety gymnastics. Just a for loop over messages.

Prefer callbacks? Kourier also supports a callback-style API with full suspend support:

channel.basicConsume(
    queue = "orders",
    onDelivery = { delivery ->
        // This is a suspend lambda — call any suspend function directly
        val message = delivery.message.body.decodeToString()
        processOrder(message)
        updateInventory(message)
        notifyShipping(message)
        channel.basicAck(delivery.message)
    }
)

Unlike the Java client's callback, onDelivery is a proper suspend lambda. No runBlocking wrapper needed. No thread-safety tricks. Just suspend functions doing what suspend functions do.

Want automatic reconnection when the network hiccups? One word change:

val connection = createRobustAMQPConnection(this, "amqp://guest:guest@localhost:5672/")
// Everything else stays the same

The robust client handles reconnection and channel recovery automatically. No custom retry logic, no connection state machines.

The null safety bonus

Did you notice the !! operators in the Java client code?

val message = String(body!!, Charsets.UTF_8)
channel.basicAck(envelope!!.deliveryTag, false)

The Java client returns platform types. Every parameter in handleDelivery is technically nullable from Kotlin's perspective, even though they're never actually null in practice.

You either:

Scatter !! everywhere (and pray)
Add defensive ?.let blocks (verbose)
Suppress warnings and hope for the best

Kourier gives you actual Kotlin types:

for (delivery in consumer) {
    val body: ByteArray = delivery.message.body        // Non-nullable
    val message: String = body.decodeToString()        // Clean conversion
    val tag: Long = delivery.message.deliveryTag       // Non-nullable
}

No force unwrapping. No null checks. Just types that mean what they say.

"But is it production ready?"

Fair question. Here's what Kourier brings to production:

Automatic reconnection — Connection drops happen. Kourier reconnects and resumes consuming.
Publisher confirms — Ensure messages reach the broker before proceeding.
OpenTelemetry integration — Distributed tracing out of the box.
Listed on RabbitMQ's official client page — Not some random GitHub repo.

It's not a wrapper around the Java client. It's a pure Kotlin implementation of the AMQP 0-9-1 protocol, built for coroutines from day one.

The migration path

You don't have to rewrite everything at once.

Start with one consumer. The one that's been causing backpressure issues. The one where you've been fighting thread pools. Replace the Java client callback with Kourier's suspend-based consumer.

See how it feels. Check the metrics. Then decide.

// build.gradle.kts
implementation("dev.kourier:amqp-client:0.3.2")

// If you want automatic reconnection
implementation("dev.kourier:amqp-client-robust:0.3.2")

Stop fighting your language

Kotlin gave you coroutines for a reason. Structured concurrency, cancellation, backpressure — these aren't features you should have to work around.

Every runBlocking in a message handler is a sign that your tools don't match your language. Every !! on a delivery callback is friction that shouldn't exist.

You wouldn't use a Java HTTP client in a Ktor app. Why use a Java AMQP client in a coroutine-based system?

Kourier is a pure Kotlin AMQP/RabbitMQ client with native coroutine support. Star it, try it, break it — and open an issue when you do.