RabbitMQ #

// build.gradle.kts
dependencies {
    // RabbitMQ Java Client (AMQP)
    implementation("com.rabbitmq:amqp-client:5.21.0")

    // kotlinx.serialization for messages
    implementation("org.jetbrains.kotlinx:kotlinx-serialization-json:1.6.3")

    // Coroutines (optional)
    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.8.0")
}

import com.rabbitmq.client.*
import kotlinx.serialization.Serializable
import kotlinx.serialization.json.Json

object RabbitMQConnection {
    private val factory = ConnectionFactory().apply {
        host = System.getenv("RABBITMQ_HOST") ?: "localhost"
        port = System.getenv("RABBITMQ_PORT")?.toInt() ?: 5672
        username = System.getenv("RABBITMQ_USER") ?: "guest"
        password = System.getenv("RABBITMQ_PASSWORD") ?: "guest"
        virtualHost = System.getenv("RABBITMQ_VHOST") ?: "/"

        // Connection settings
        connectionTimeout = 30_000        // 30 second connection timeout
        requestedHeartbeat = 60           // heartbeat every 60 seconds
        isAutomaticRecoveryEnabled = true // automatically reconnect if the connection drops
        networkRecoveryInterval = 5_000   // try reconnecting every 5 seconds
    }

    // One connection can have many channels
    // A channel is the logical unit of communication, lightweight to create
    fun createConnection(): Connection = factory.newConnection()
}

// Usage
fun main() {
    RabbitMQConnection.createConnection().use { connection ->
        connection.createChannel().use { channel ->
            println("Connected to RabbitMQ: ${connection.serverProperties["version"]}")
        }
    }
}

@Serializable
data class OrderEvent(
    val orderId: String,
    val userId: Long,
    val total: Double,
    val status: String
)

class DirectMessaging {
    private val json = Json { ignoreUnknownKeys = true }
    private val EXCHANGE = "pesanan.direct"
    private val QUEUE_EMAIL = "pesanan.email"
    private val QUEUE_INVENTORY = "pesanan.inventori"
    private val KEY_NEW = "pesanan.baru"
    private val KEY_DONE = "pesanan.selesai"

    // Setup: declare the exchange and queues
    fun setupInfrastructure(channel: Channel) {
        // Declare the exchange — durable: survives a broker restart
        channel.exchangeDeclare(EXCHANGE, BuiltinExchangeType.DIRECT, true)

        // Declare the queues — durable: the queue survives, messages can also be durable
        channel.queueDeclare(QUEUE_EMAIL, true, false, false, null)
        channel.queueDeclare(QUEUE_INVENTORY, true, false, false, null)

        // Binding: connect queues to the exchange with routing keys
        channel.queueBind(QUEUE_EMAIL, EXCHANGE, KEY_NEW)      // email only receives new orders
        channel.queueBind(QUEUE_EMAIL, EXCHANGE, KEY_DONE)     // email also receives completed orders
        channel.queueBind(QUEUE_INVENTORY, EXCHANGE, KEY_NEW)  // inventory only new orders

        println("Messaging infrastructure ready")
    }

    // Publisher
    fun sendOrderEvent(channel: Channel, event: OrderEvent, routingKey: String) {
        val payload = json.encodeToString(OrderEvent.serializer(), event).toByteArray()

        val properties = AMQP.BasicProperties.Builder()
            .contentType("application/json")
            .deliveryMode(2)  // 2 = persistent (messages survive a broker restart)
            .messageId(event.orderId)
            .timestamp(java.util.Date())
            .build()

        channel.basicPublish(EXCHANGE, routingKey, properties, payload)
        println("Event sent to '$routingKey': ${event.orderId}")
    }

    // Consumer with manual acknowledgement
    fun startConsumer(channel: Channel, queueName: String, consumerName: String) {
        // prefetchCount = 1: take one message first, then ask for the next
        // This distributes the load evenly among consumers
        channel.basicQos(1)

        val deliverCallback = DeliverCallback { _, delivery ->
            try {
                val message = String(delivery.body)
                val event = json.decodeFromString(OrderEvent.serializer(), message)
                println("[$consumerName] Processing: ${event.orderId}")

                processEvent(consumerName, event)

                // Acknowledge — tell the broker the message was processed successfully
                channel.basicAck(delivery.envelope.deliveryTag, false)

            } catch (e: Exception) {
                println("[$consumerName] Failed: ${e.message}")
                // Nack with requeue=false → send to the dead letter (if configured)
                channel.basicNack(delivery.envelope.deliveryTag, false, false)
            }
        }

        val cancelCallback = CancelCallback { tag ->
            println("[$consumerName] Consumer cancelled: $tag")
        }

        // autoAck=false: we control when the ack is sent
        channel.basicConsume(queueName, false, deliverCallback, cancelCallback)
        println("[$consumerName] Listening on queue '$queueName'...")
    }

    private fun processEvent(consumer: String, event: OrderEvent) {
        Thread.sleep(500)  // simulate processing
        println("[$consumer] Done processing ${event.orderId}")
    }
}

fun main() {
    val messaging = DirectMessaging()
    val json = Json { ignoreUnknownKeys = true }

    RabbitMQConnection.createConnection().use { connection ->
        connection.createChannel().use { channel ->
            messaging.setupInfrastructure(channel)

            // Publish some messages
            listOf(
                OrderEvent("ORD-001", 1L, 150_000.0, "NEW") to "pesanan.baru",
                OrderEvent("ORD-002", 2L, 250_000.0, "NEW") to "pesanan.baru",
                OrderEvent("ORD-001", 1L, 150_000.0, "DONE") to "pesanan.selesai"
            ).forEach { (event, key) ->
                messaging.sendOrderEvent(channel, event, key)
            }
        }

        // Consumer on a separate channel
        val emailChannel = connection.createChannel()
        messaging.startConsumer(emailChannel, "pesanan.email", "Email Service")

        Thread.sleep(5000)
    }
}

class FanoutMessaging {
    private val json = Json { ignoreUnknownKeys = true }
    private val EXCHANGE = "notifikasi.fanout"

    fun setup(channel: Channel) {
        channel.exchangeDeclare(EXCHANGE, BuiltinExchangeType.FANOUT, true)

        // Each service creates its own queue and binds to the fanout exchange
        // Queues can be temporary (exclusive, autodelete)
        val emailQueue = channel.queueDeclare("notif.email", true, false, false, null).queue
        val smsQueue   = channel.queueDeclare("notif.sms", true, false, false, null).queue
        val pushQueue  = channel.queueDeclare("notif.push", true, false, false, null).queue

        // Binding to fanout — the routing key is ignored
        channel.queueBind(emailQueue, EXCHANGE, "")
        channel.queueBind(smsQueue,   EXCHANGE, "")
        channel.queueBind(pushQueue,  EXCHANGE, "")
    }

    fun broadcast(channel: Channel, message: String) {
        val payload = message.toByteArray()
        val props = AMQP.BasicProperties.Builder().deliveryMode(2).build()

        // Empty routing key for fanout
        channel.basicPublish(EXCHANGE, "", props, payload)
        println("Broadcast sent to all subscribers: $message")
    }
}

class TopicMessaging {
    private val json = Json { ignoreUnknownKeys = true }
    private val EXCHANGE = "sistem.topic"

    fun setup(channel: Channel) {
        channel.exchangeDeclare(EXCHANGE, BuiltinExchangeType.TOPIC, true)

        // Queue for all order events
        channel.queueDeclare("audit.semua", true, false, false, null)
        channel.queueBind("audit.semua", EXCHANGE, "pesanan.#")
        // Receives: pesanan.baru, pesanan.bayar, pesanan.kirim, pesanan.selesai, etc.

        // Queue only for any errors
        channel.queueDeclare("alert.error", true, false, false, null)
        channel.queueBind("alert.error", EXCHANGE, "*.error")
        // Receives: pesanan.error, pembayaran.error, pengiriman.error, etc.

        // Queue for general monitoring
        channel.queueDeclare("monitoring.semua", true, false, false, null)
        channel.queueBind("monitoring.semua", EXCHANGE, "#")
        // Receives: ALL messages regardless of routing key
    }

    fun send(channel: Channel, routingKey: String, message: String) {
        val props = AMQP.BasicProperties.Builder().deliveryMode(2).build()
        channel.basicPublish(EXCHANGE, routingKey, props, message.toByteArray())
        println("Message '$message' sent with key '$routingKey'")
    }
}

fun setupWithDlx(channel: Channel) {
    // Create the DLX exchange and queue first
    channel.exchangeDeclare("dlx.exchange", BuiltinExchangeType.DIRECT, true)
    channel.queueDeclare("dlx.queue", true, false, false, null)
    channel.queueBind("dlx.queue", "dlx.exchange", "dead")

    // Create the main queue with DLX arguments
    val args = mapOf(
        "x-dead-letter-exchange" to "dlx.exchange",    // the DLX destination exchange
        "x-dead-letter-routing-key" to "dead",          // the routing key for the DLX
        "x-message-ttl" to 30_000,                      // messages expire after 30 seconds
        "x-max-length" to 1000                           // max 1000 messages in the queue
    )

    channel.exchangeDeclare("pesanan.exchange", BuiltinExchangeType.DIRECT, true)
    channel.queueDeclare("pesanan.queue", true, false, false, args)
    channel.queueBind("pesanan.queue", "pesanan.exchange", "pesanan")

    println("Queue with DLX configured successfully")
}

// A consumer that sends to the DLX on failure
fun consumerWithDlx(channel: Channel) {
    channel.basicQos(1)

    channel.basicConsume("pesanan.queue", false,
        { _, delivery ->
            val message = String(delivery.body)
            try {
                processMessage(message)
                channel.basicAck(delivery.envelope.deliveryTag, false)
            } catch (e: Exception) {
                println("Failed: ${e.message} — sending to DLX")
                // requeue=false: don't return to the main queue, send to the DLX
                channel.basicNack(delivery.envelope.deliveryTag, false, false)
            }
        },
        { tag -> println("Consumer cancelled: $tag") }
    )
}

fun processMessage(message: String) {
    if (Math.random() < 0.3) throw RuntimeException("Processing failed!")
    println("Message processed: $message")
}

import java.util.UUID
import java.util.concurrent.ArrayBlockingQueue

class RpcClient(private val channel: Channel) {
    private val QUEUE_REQUEST = "rpc.hitung"

    // Temporary queue for receiving replies
    private val replyQueue = channel.queueDeclare().queue

    // Correlation ID → queue for storing temporary results
    private val pending = java.util.concurrent.ConcurrentHashMap<
        String, ArrayBlockingQueue<String>
    >()

    init {
        // Set up the consumer for the reply queue
        channel.basicConsume(replyQueue, true,
            { _, delivery ->
                val correlationId = delivery.properties.correlationId
                val result = String(delivery.body)
                pending[correlationId]?.offer(result)
            },
            { _ -> }
        )
    }

    fun callRpc(input: String, timeoutMs: Long = 5000): String? {
        val correlationId = UUID.randomUUID().toString()
        val resultQueue = ArrayBlockingQueue<String>(1)
        pending[correlationId] = resultQueue

        try {
            val props = AMQP.BasicProperties.Builder()
                .correlationId(correlationId)
                .replyTo(replyQueue)        // tell the server where to send the reply
                .build()

            channel.basicPublish("", QUEUE_REQUEST, props, input.toByteArray())
            println("RPC request '$input' sent with correlationId: $correlationId")

            // Wait for the reply with a timeout
            return resultQueue.poll(timeoutMs, java.util.concurrent.TimeUnit.MILLISECONDS)
        } finally {
            pending.remove(correlationId)
        }
    }
}

class RpcServer(private val channel: Channel) {
    private val QUEUE_REQUEST = "rpc.hitung"

    fun start() {
        channel.queueDeclare(QUEUE_REQUEST, false, false, false, null)
        channel.basicQos(1)

        channel.basicConsume(QUEUE_REQUEST, false,
            { _, delivery ->
                val input = String(delivery.body)
                println("RPC request received: $input")

                // Process the request
                val result = processRequest(input)

                // Send the reply to the replyTo queue with the same correlationId
                val props = AMQP.BasicProperties.Builder()
                    .correlationId(delivery.properties.correlationId)
                    .build()

                channel.basicPublish(
                    "", delivery.properties.replyTo,
                    props, result.toByteArray()
                )

                channel.basicAck(delivery.envelope.deliveryTag, false)
                println("Reply '$result' sent")
            },
            { _ -> }
        )
        println("RPC Server listening on queue '$QUEUE_REQUEST'")
    }

    private fun processRequest(input: String): String {
        val number = input.toIntOrNull() ?: return "Error: not a number"
        return (number * number).toString()  // square the number
    }
}

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.Channel as KotlinChannel

fun rabbitConsumerFlow(
    rabbitChannel: Channel,
    queueName: String
): kotlinx.coroutines.flow.Flow<String> = kotlinx.coroutines.flow.callbackFlow {
    rabbitChannel.basicQos(10)

    val tag = rabbitChannel.basicConsume(queueName, false,
        { _, delivery ->
            val message = String(delivery.body)
            trySend(message)
            rabbitChannel.basicAck(delivery.envelope.deliveryTag, false)
        },
        { _ -> close() }
    )

    awaitClose {
        runCatching { rabbitChannel.basicCancel(tag) }
    }
}

fun main() = runBlocking {
    val connection = RabbitMQConnection.createConnection()
    val channel = connection.createChannel()

    rabbitConsumerFlow(channel, "pesanan.email")
        .collect { message ->
            println("Coroutine consumer: $message")
            // process the message in a suspend way if needed
        }
}

// 1. Always declare queues and exchanges as durable
channel.exchangeDeclare("name", BuiltinExchangeType.DIRECT, /* durable= */ true)
channel.queueDeclare("name", /* durable= */ true, false, false, null)

// 2. Always set messages as persistent
val properties = AMQP.BasicProperties.Builder()
    .deliveryMode(2)  // 2 = persistent
    .build()

// 3. Set the right prefetchCount for consumers
channel.basicQos(10)  // process max 10 messages at once, before acknowledging

// 4. Automatic recovery is configured in the factory, but make sure
//    channels are recreated after reconnect because channels aren't recovered automatically
factory.isAutomaticRecoveryEnabled = true

// 5. Monitoring: use the RabbitMQ Management Plugin (port 15672)
// GET http://localhost:15672/api/queues to see queue status via REST

// 6. Configure consumer timeouts
channel.basicConsume(queue, false, deliverCallback, cancelCallback)
// If a consumer doesn't process within x seconds, the connection will be closed
// Set consumer_timeout in rabbitmq.conf