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

MessagingService

open class MessagingService(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 prefetchCount: UShort, val maxXDeathCount: Int = 1, val connectionName: String = queue.queue) : IMessagingService

A robust messaging service that handles connection, setup, publishing, and listening for messages.

Inheritors

DelayedMessagingService

Constructors

MessagingService

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, prefetchCount: UShort, 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?

The raw AMQP channel instance.

channelReady

open override val channelReady: CompletableDeferred<Unit>

A Deferred that completes when the connection and channels are ready.

connection

open override var connection: AMQPConnection?

The raw AMQP connection instance.

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

prefetchCount

open val prefetchCount: UShort

queue

open val queue: IMessagingQueue

quorum

open val quorum: Boolean = false

setupCompleted

open override val setupCompleted: CompletableDeferred<Unit>

A Deferred that completes when the setup process is finished.

user

open val user: String

Functions

connect

open suspend override fun connect()

Connects to the messaging server.

exchangeDeclare

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

Declares an exchange with the specified parameters.

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()

Starts listening for incoming messages in a coroutine.

publish

inline suspend fun <T> publish(routingKey: IMessagingKey, value: T, exchange: IMessagingExchange? = null, persistent: Boolean = false, attempts: Int = 3, delay: Long = 5000, crossinline configureProperties: PropertiesBuilder.() -> Unit = {})

queueBind

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

Binds a queue to an exchange with the specified routing key and arguments.

queueDeclare

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

Declares a queue with the specified parameters.

routingKey

open fun routingKey(key: String): IMessagingKey

setup

open suspend override fun setup()

Sets up the necessary exchanges, queues, and bindings.