How to Properly Close Connections and Channels in Kourier to Release Resources

Always close channels before connections using the suspend functions channel.close() and connection.close() to ensure AMQP handshake completion and full resource release.

Kourier is a Kotlin Multiplatform AMQP client that implements the RabbitMQ AMQP 0-9-1 protocol. Properly managing the lifecycle of connections and channels is critical to prevent resource leaks, dangling coroutines, and broker-side channel accumulation. This guide explains how to close connections and channels in Kourier using the actual source implementation from the guimauvedigital/kourier repository.

Understanding Kourier's Resource Hierarchy

In Kourier's architecture, a connection owns the TCP socket, heartbeat logic, and a collection of channels. Each channel maintains its own state, pending deliveries, and flow-control handlers.

Because channels depend on the connection's transport layer, you must close every channel explicitly before closing the connection. Closing the connection first triggers a broker-side cleanup, but explicit channel closure ensures deterministic resource release and proper AMQP handshake completion.

The Proper Shutdown Sequence

Kourier implements graceful shutdown through two primary classes: DefaultAMQPChannel and DefaultAMQPConnection. Both use suspend functions that perform full AMQP handshakes before releasing underlying resources.

Closing Channels

The DefaultAMQPChannel.close method (located in amqp-client/src/commonMain/kotlin/dev/kourier/amqp/channel/DefaultAMQPChannel.kt) sends a Channel.Close frame to the broker and awaits the Channel.Closed response:

suspend fun close(reason: String = "", code: UShort = 200u): AMQPResponse.Channel.Closed

This method performs three critical actions:

• Sends the close frame using writeAndWaitForResponse
• Invokes cancelAll to complete the channelClosed deferred object
• Releases all coroutines attached to the channel's scope

Closing Connections

After all channels are closed, invoke DefaultAMQPConnection.close (found in amqp-client/src/commonMain/kotlin/dev/kourier/amqp/connection/DefaultAMQPConnection.kt):

suspend fun close(reason: String, code: UShort): AMQPResponse.Connection.Closed

The connection close sequence:

• Sends Connection.Close frame to the broker
• Awaits the broker's Connection.Closed frame
• Calls cancelAll which:
  • Cancels the socket subscription and heartbeat subscription
  • Closes the underlying TCP socket
  • Completes the connectionClosed deferred to resume any awaiting coroutines

Practical Implementation Examples

Manual Resource Management

For explicit control over connection and channel lifecycles, create resources in a try-finally block:

import dev.kourier.amqp.connection.createAMQPConnection
import dev.kourier.amqp.channel.AMQPChannel
import kotlinx.coroutines.runBlocking

runBlocking {
    val connection = createAMQPConnection(this) {}
    val channel: AMQPChannel = connection.openChannel()
    
    try {
        // Perform publishing or consuming operations
        channel.basicPublish(exchange, routingKey, message)
    } finally {
        // Always close channel before connection
        channel.close(reason = "Work completed", code = 200u)
        connection.close(reason = "Application shutdown", code = 200u)
    }
}

This pattern ensures that even if an exception occurs during processing, both channel.close() and connection.close() are invoked, releasing the TCP socket and canceling all related coroutines.

Using the withConnection Helper

Kourier provides a convenience wrapper in amqp-client/src/commonTest/kotlin/dev/kourier/amqp/withConnection.kt that guarantees connection closure:

import dev.kourier.amqp.withConnection
import dev.kourier.amqp.channel.AMQPChannel

withConnection { conn ->
    val channel = conn.openChannel()
    
    // Use the channel for operations
    channel.basicConsume(queue) { delivery ->
        // Process message
    }
    
    // Explicitly close the channel before block exits
    channel.close()
    // Connection closes automatically in finally block
}

The withConnection helper executes connection.close() in a finally clause, ensuring resource release even if the block throws an exception. However, you must still manually close any channels opened within the block before the connection closes.

Handling Multiple Channels

When working with multiple channels, close each one individually before closing the parent connection:

runBlocking {
    val connection = createAMQPConnection(this) {}
    
    // Open multiple channels
    val channels = (1..3).map { connection.openChannel() }
    
    try {
        channels.forEachIndexed { index, channel ->
            channel.basicPublish("logs", "routing.$index", message)
        }
    } finally {
        // Close all channels first
        channels.forEach { 
            it.close(reason = "Batch complete", code = 200u) 
        }
        // Then close the connection
        connection.close(reason = "Cleanup", code = 200u)
    }
}

This pattern prevents resource leaks when managing high-throughput scenarios with multiple concurrent channels.

Key Implementation Files

Understanding the source structure helps when debugging resource management issues:

File	Purpose
amqp-client/src/commonMain/kotlin/dev/kourier/amqp/channel/DefaultAMQPChannel.kt	Implements channel lifecycle, including the close method that sends Channel.Close frames and releases coroutines.
amqp-client/src/commonMain/kotlin/dev/kourier/amqp/connection/DefaultAMQPConnection.kt	Implements connection lifecycle, socket management, and the close method that shuts down the TCP socket and heartbeat mechanisms.
amqp-client/src/commonMain/kotlin/dev/kourier/amqp/channel/AMQPChannel.kt	Public interface declaring the close suspend function that all channel implementations must provide.
amqp-client/src/commonMain/kotlin/dev/kourier/amqp/connection/AMQPConnection.kt	Public interface exposing the close suspend function for connection management.
amqp-client/src/commonTest/kotlin/dev/kourier/amqp/withConnection.kt	Test utility providing the withConnection helper for automatic resource cleanup.

Summary

• Close channels before connections to ensure proper AMQP handshake completion and deterministic resource release.
• Always await close methods since both DefaultAMQPChannel.close and DefaultAMQPConnection.close are suspend functions that perform network I/O.
• Use try-finally blocks or the withConnection helper to guarantee closure even when exceptions occur.
• Close all channels individually when managing multiple channels before shutting down the parent connection.
• Reference source files DefaultAMQPChannel.kt and DefaultAMQPConnection.kt when implementing custom resource management logic.

Frequently Asked Questions

What happens if I close the connection before closing channels?

If you invoke connection.close() while channels are still open, the broker sends Channel.Close frames for each open channel as part of the connection shutdown sequence. However, you lose the ability to inspect per-channel AMQPResponse.Channel.Closed results and may encounter race conditions in client-side coroutine scopes. Always close channels explicitly first to ensure deterministic cleanup.

Why are the close methods suspend functions?

Both DefaultAMQPChannel.close and DefaultAMQPConnection.close are marked suspend because they perform network I/O operations. They send Channel.Close or Connection.Close frames to the broker and await the corresponding Closed response frames. This asynchronous handshake ensures the broker acknowledges the shutdown before the client releases socket resources and cancels related coroutines.

Can I reuse a connection or channel after calling close?

No. Once close() completes successfully, the underlying TCP socket is shut down and the coroutine scope is canceled. The DefaultAMQPConnection and DefaultAMQPChannel implementations set their internal state to CLOSED and short-circuit subsequent operations. Attempting to publish or consume on a closed channel throws an exception; attempting to open new channels on a closed connection fails immediately.

How do I ensure resources are closed when exceptions occur?

Wrap your channel and connection usage in a try-finally block, calling channel.close() and connection.close() in the finally clause. Alternatively, use the withConnection helper function from the test utilities, which automatically closes the connection in a finally block. For multiple channels, maintain a list and close each one in the finally block before closing the parent connection.

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