messaging-amqp/dev.kaccelero.commons.messaging/DelayedMessagingService

DelayedMessagingService

open class DelayedMessagingService(val host: String, val user: String, val password: String, val exchange: IMessagingExchange, val queue: IMessagingQueue, val keys: List<IMessagingKey>, val handleMessagingUseCaseFactory: () -> IHandleMessagingUseCase, val coroutineScope: CoroutineScope, val json: Json? = null, val autoConnect: Boolean = true, val autoListen: Boolean = true, val persistent: Boolean = false, val quorum: Boolean = false, val dead: Boolean = false, val maxXDeathCount: Int = 1, val connectionName: String = queue.queue) : MessagingService

Constructors

constructor(host: String, user: String, password: String, exchange: IMessagingExchange, queue: IMessagingQueue, keys: List<IMessagingKey>, handleMessagingUseCaseFactory: () -> IHandleMessagingUseCase, coroutineScope: CoroutineScope, json: Json? = null, autoConnect: Boolean = true, autoListen: Boolean = true, persistent: Boolean = false, quorum: Boolean = false, dead: Boolean = false, maxXDeathCount: Int = 1, connectionName: String = queue.queue)

Properties

autoConnect

open val autoConnect: Boolean = true

autoListen

open val autoListen: Boolean = true

channel

open override var channel: AMQPChannel?

connection

open override var connection: AMQPConnection?

connectionName

open val connectionName: String

coroutineScope

open val coroutineScope: CoroutineScope

dead

open val dead: Boolean = false

exchange

open val exchange: IMessagingExchange

handleMessagingUseCaseFactory

open val handleMessagingUseCaseFactory: () -> IHandleMessagingUseCase

host

open val host: String

json

open val json: Json? = null

keys

open val keys: List<IMessagingKey>

maxXDeathCount

open val maxXDeathCount: Int = 1

password

open val password: String

persistent

open val persistent: Boolean = false

queue

open val queue: IMessagingQueue

quorum

open val quorum: Boolean = false

user

open val user: String

Functions

connect

open suspend override fun connect()

exchangeDeclare

open suspend override fun exchangeDeclare(exchange: IMessagingExchange, type: String, arguments: Map<String, Field>)

handleException

open suspend fun handleException(delivery: AMQPResponse.Channel.Message.Delivery, exception: Exception)

handleFailedMessage

open suspend fun handleFailedMessage(delivery: AMQPResponse.Channel.Message.Delivery, exception: Exception)

listen

open suspend override fun listen()

publish

inline suspend fun <T> publish(routingKey: IMessagingKey, value: T, publishAt: Instant?, exchange: IMessagingExchange? = null, persistent: Boolean = false, attempts: Int = 3, delay: Long = 5000)

inline suspend fun <T> publish(routingKey: IMessagingKey, value: T, exchange: IMessagingExchange? = null, persistent: Boolean = false, attempts: Int = 3, delay: Long = 5000)

queueBind

open suspend override fun queueBind(queue: IMessagingQueue, exchange: IMessagingExchange, routingKey: IMessagingKey, arguments: Map<String, Field>)

queueDeclare

open suspend override fun queueDeclare(queue: IMessagingQueue, exchange: IMessagingExchange?, durable: Boolean, exclusive: Boolean, autoDelete: Boolean, arguments: Map<String, Field>)

routingKey

open fun routingKey(key: String): IMessagingKey

setup

open suspend override fun setup()

tryWithAttempts

open suspend fun tryWithAttempts(attempts: Int = 3, delay: Long = 5000, block: suspend () -> Unit)