How to Use Message Properties and Headers in Kourier: A Complete Guide

Kourier models AMQP 0-9-1 message properties through the Properties data class and provides a properties { } DSL builder to set standard fields like contentType and replyTo, while custom headers are managed via the Table type using the tableOf() helper.

Kourier is a Kotlin multiplatform AMQP client that strictly follows the AMQP 0-9-1 specification. Every message can carry a properties block containing 14 standard fields plus an optional headers table for arbitrary metadata. This guide demonstrates how to construct these structures using the DSL defined in amqp-core/src/commonMain/kotlin/dev/kourier/amqp/Extensions.kt and how to access them during consumption via AMQPMessage.

Understanding AMQP Message Properties in Kourier

The AMQP specification defines a fixed set of message attributes. In Kourier, these are represented by an immutable data class.

The Properties Data Class

Located in amqp-core/src/commonMain/kotlin/dev/kourier/amqp/Properties.kt, the Properties class declares all 14 optional fields as nullable types:

data class Properties(
    val contentType: String? = null,
    val contentEncoding: String? = null,
    val headers: Table? = null,
    val deliveryMode: UByte? = null,
    val priority: UByte? = null,
    val correlationId: String? = null,
    val replyTo: String? = null,
    val expiration: String? = null,
    val messageId: String? = null,
    val timestamp: Long? = null,
    val type: String? = null,
    val userId: String? = null,
    val appId: String? = null,
    val reserved1: String? = null,
)

Because every field defaults to null, you only need to specify the properties relevant to your use case.

The Properties Builder DSL

Manually constructing the data class is verbose. Kourier provides a type-safe builder DSL defined in amqp-core/src/commonMain/kotlin/dev/kourier/amqp/Extensions.kt:

fun properties(block: PropertiesBuilder.() -> Unit): Properties =
    PropertiesBuilder().apply(block).build()

The PropertiesBuilder class (in PropertiesBuilder.kt) mirrors the data class with mutable fields, allowing idiomatic Kotlin configuration:

val props = properties {
    contentType = "application/json"
    deliveryMode = 2u               // Persistent delivery
    replyTo = "response-queue"
    correlationId = "uuid-1234"
}

Working with Message Headers

While the 14 standard properties cover common metadata, the headers property accepts a Table—a map of user-defined key-value pairs.

The Table and Field Types

In amqp-core/src/commonMain/kotlin/dev/kourier/amqp/Field.kt, the sealed class Field represents AMQP data types:

sealed class Field {
    data class Boolean(val value: kotlin.Boolean) : Field()
    data class Short(val value: kotlin.Short) : Field()
    data class Long(val value: kotlin.Int) : Field()
    data class LongLong(val value: kotlin.Long) : Field()
    data class LongString(val value: String) : Field()
    data class Table(val value: dev.kourier.amqp.Table) : Field()
    // ... additional primitive types
}

A Table is defined as Map<String, Field>, allowing nested structures.

Creating Headers with tableOf()

The Extensions.kt file provides the tableOf() helper to convert Kotlin types into the AMQP Table format:

fun tableOf(vararg pairs: Pair<String, Any?>): Table = 
    pairs.toMap().mapValues { it.value.toField() }

This automatically boxes values into the correct Field subtype. Example:

val headers = tableOf(
    "x-trace-id" to "abc-123",
    "x-retry-count" to 3,
    "x-priority" to "high"
)

Publishing Messages with Properties and Headers

Combine the DSL and tableOf() when calling basicPublish. The AMQPChannel interface accepts a Properties instance:

import dev.kourier.amqp.*

suspend fun publishOrderEvent(channel: AMQPChannel, orderId: String) {
    val payload = """{"orderId":"$orderId"}""".toByteArray()
    
    val props = properties {
        contentType = "application/json"
        deliveryMode = 2u                // Persistent
        messageId = orderId
        timestamp = System.currentTimeMillis()
        headers = tableOf(
            "x-event-type" to "order.created",
            "x-service" to "payment-service"
        )
    }

    channel.basicPublish(
        body = payload,
        exchange = "orders.exchange",
        routingKey = "order.created",
        properties = props
    )
}

Source references: Properties.kt for the data class, Extensions.kt for properties {} and tableOf(), and AMQPChannel.basicPublish in the client module.

Consuming and Reading Properties/Headers

When consuming, the Delivery object contains an AMQPMessage (defined in amqp-client/src/commonMain/kotlin/dev/kourier/amqp/AMQPMessage.kt) which exposes the Properties:

suspend fun consumeWithHeaders(channel: AMQPChannel) {
    val consumer = channel.basicConsume("my-queue", noAck = true)
    
    for (delivery in consumer) {
        val msg = delivery.message
        val props = msg.properties
        
        // Standard properties
        println("ContentType: ${props.contentType}")
        println("MessageId: ${props.messageId}")
        
        // Custom headers
        val headers = props.headers?.toMap() ?: emptyMap()
        val traceId = headers["x-trace-id"] as? String
        val retryCount = headers["x-retry-count"] as? Int ?: 0
        
        println("Processing with trace=$traceId, retries=$retryCount")
    }
}

The Table.toMap() extension (in Extensions.kt) recursively unwraps Field instances back to native Kotlin types.

Real-World Example: RPC Pattern

The RPC tutorial (docs/tutorials/rpc.md) demonstrates practical usage of replyTo and correlationId. The client builds properties that tell the server where to respond:

val requestProps = properties {
    replyTo = callbackQueueName
    correlationId = UUID.randomUUID().toString()
}

channel.basicPublish(
    body = requestBytes,
    exchange = "",
    routingKey = "rpc_queue",
    properties = requestProps
)

The server then reads these properties to route the response correctly, illustrating how standard AMQP properties enable request-reply patterns without custom headers.

Summary

• Properties are modeled by the dev.kourier.amqp.Properties data class in Properties.kt, exposing all 14 AMQP 0-9-1 fields including headers.
• DSL Builder: Use properties { } from Extensions.kt for type-safe, readable construction of property sets.
• Headers: Represented as a Table ( Map<String, Field> ); use tableOf() to convert Kotlin values to AMQP field types automatically.
• Publishing: Pass the built Properties instance to channel.basicPublish().
• Consuming: Access delivery.message.properties and unwrap headers with headers?.toMap().

Frequently Asked Questions

How do I set a message as persistent in Kourier?

Set the deliveryMode property to 2u (unsigned byte value 2). According to the AMQP specification and the Properties data class in Properties.kt, a value of 2 indicates persistent delivery mode, ensuring the message survives broker restarts when the queue is durable.

Can I nest tables inside headers?

Yes. The Field sealed class in Field.kt includes a Table variant that holds a Map<String, Field>. When using tableOf(), you can nest another tableOf() call as a value, and the toField() extension will recursively wrap it into Field.Table, creating nested header structures compliant with AMQP 0-9-1.

What is the difference between standard properties and headers?

Standard properties are the 14 predefined fields in the AMQP basic.properties frame (e.g., contentType, replyTo, messageId), modeled as individual fields in the Properties class. Headers are an arbitrary key-value map stored in the headers field (type Table) within that same properties block, intended for application-defined metadata like trace IDs or retry counts.

How do I read headers when consuming messages?

Access the headers property on the Properties object, which returns a Table? (defined in Extensions.kt as Map<String, Field>). Use the toMap() extension function to convert it to a Map<String, Any?>, which unwraps the Field sealed class instances back to native Kotlin types like String, Int, or nested Map structures.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}