feat: add deliveryMode (1: non-persistent, 2: persistent) to AMQP configuration with default of 1 (non-persistent)

Author: rudwna

keycloak-webhook-provider-amqp/src/main/kotlin/AmqpWebhookHandler.kt

@@ -15,6 +15,7 @@ class AmqpWebhookHandler : WebhookHandler {
     private lateinit var connection: Connection
     private lateinit var exchange: String
     private lateinit var connectionFactory: ConnectionFactory
+    private var deliveryMode: Int = 1
 
     companion object {
         const val PROVIDER_ID = "webhook-amqp"
@@ -25,23 +26,23 @@ class AmqpWebhookHandler : WebhookHandler {
         @JvmStatic
         private val logger = LoggerFactory.getLogger(AmqpWebhookHandler::class.java)
 
-        @JvmStatic
-        private fun getMessageProps(className: String): BasicProperties {
-            val headers: MutableMap<String, Any> = HashMap()
-            headers["__TypeId__"] = className
-            return BasicProperties.Builder()
-                .appId("Keycloak/Kotlin")
-                .headers(headers)
-                .contentType(MediaType.APPLICATION_JSON)
-                .contentEncoding("UTF-8")
-                .build()
-        }
-
         @JvmStatic
         private fun genRoutingKey(request: WebhookPayload): String =
             "KC_CLIENT.${request.realmId}.${request.clientId ?: "xxx"}.${request.userId ?: "xxx"}.${request.type}"
     }
 
+    private fun getMessageProps(className: String): BasicProperties {
+        val headers: MutableMap<String, Any> = HashMap()
+        headers["__TypeId__"] = className
+        return BasicProperties.Builder()
+            .appId("Keycloak/Kotlin")
+            .headers(headers)
+            .contentType(MediaType.APPLICATION_JSON)
+            .contentEncoding("UTF-8")
+            .deliveryMode(this.deliveryMode)
+            .build()
+    }
+
     /**
      * Ensures that the connection and channel are open.
      * If either is closed, it will try to reinitialize them up to 3 times.
@@ -122,6 +123,7 @@ class AmqpWebhookHandler : WebhookHandler {
         val amqp = AmqpConfig.fromEnv()
 
         exchange = amqp.exchange
+        deliveryMode = amqp.deliveryMode
 
         if (this::connection.isInitialized && this::channel.isInitialized && connection.isOpen && channel.isOpen) {
             logger.debug("Connection is already open")

keycloak-webhook-provider-amqp/src/main/kotlin/models/AmqpConfig.kt

@@ -10,6 +10,7 @@ data class AmqpConfig(
     val vHost: String?,
     val ssl: Boolean,
     val exchange: String,
+    val deliveryMode: Int
 ) {
     companion object {
         fun fromEnv(): AmqpConfig = AmqpConfig(
@@ -19,7 +20,8 @@ data class AmqpConfig(
             port = amqpPortKey.cff(),
             vHost = amqpVHostKey.cf(),
             ssl = amqpSsl.bf(),
-            exchange = amqpExchangeKey.cff()
+            exchange = amqpExchangeKey.cff(),
+            deliveryMode = amqpDeliveryMode.cfe(fun() = "1").toInt()
         )
     }
 }

keycloak-webhook-provider-core/src/main/kotlin/helper/Config.kt

@@ -13,6 +13,7 @@ const val amqpPortKey = "WEBHOOK_AMQP_PORT"
 const val amqpVHostKey = "WEBHOOK_AMQP_VHOST"
 const val amqpExchangeKey = "WEBHOOK_AMQP_EXCHANGE"
 const val amqpSsl = "WEBHOOK_AMQP_SSL"
+const val amqpDeliveryMode = "WEBHOOK_AMQP_DELIVERY_MODE"
 
 const val syslogProtocol = "WEBHOOK_SYSLOG_PROTOCOL"
 const val syslogHostname = "WEBHOOK_SYSLOG_HOSTNAME"