amqp-client/dev.kourier.amqp.channel/DefaultAMQPChannel

DefaultAMQPChannel

open class DefaultAMQPChannel(val connection: DefaultAMQPConnection, val id: ChannelId, val frameMax: UInt) : AMQPChannel

Constructors

DefaultAMQPChannel

constructor(connection: DefaultAMQPConnection, id: ChannelId, frameMax: UInt)

Properties

channelClosed

open override val channelClosed: CompletableDeferred<AMQPException.ChannelClosed>

A deferred that completes when the channek is closed.

channelResponses

val channelResponses: MutableSharedFlow<AMQPResponse>

closedResponses

open override val closedResponses: Flow<AMQPResponse.Channel.Closed>

A flow of closed responses from the channel.

connection

open val connection: DefaultAMQPConnection

flowResponses

open override val flowResponses: Flow<AMQPResponse.Channel.Flowed>

A flow of basic delivery messages.

frameMax

val frameMax: UInt

open override val id: ChannelId

Unique identifier of the channel.

isConfirmMode

open override var isConfirmMode: Boolean

True when the channel is in confirm mode.

isTxMode

open override var isTxMode: Boolean

True when the channel is in transaction mode.

nextMessage

var nextMessage: PartialDelivery?

openedResponses

open override val openedResponses: Flow<AMQPResponse.Channel.Opened>

A flow of opened responses from the channel.

publishConfirmResponses

open override val publishConfirmResponses: Flow<AMQPResponse.Channel.Basic.PublishConfirm>

A flow of basic publish confirm responses.

returnResponses

open override val returnResponses: Flow<AMQPResponse.Channel.Message.Return>

A flow of basic return responses.

state

open override var state: ConnectionState

The channel connection state.

writeMutex

val writeMutex: Mutex

Functions

basicAck

open suspend override fun basicAck(message: AMQPMessage, multiple: Boolean)

open suspend override fun basicAck(deliveryTag: ULong, multiple: Boolean)

Acknowledge a message.

basicCancel

open suspend override fun basicCancel(consumerTag: String): AMQPResponse.Channel.Basic.Canceled

Cancel sending messages from server to consumer.

basicConsume

open suspend override fun basicConsume(queue: String, consumerTag: String, noAck: Boolean, exclusive: Boolean, arguments: Table): AMQPReceiveChannel

Consume messages from a queue by sending them to the channel. The consumer is automatically canceled when the channel is closed.

open suspend override fun basicConsume(queue: String, consumerTag: String, noAck: Boolean, exclusive: Boolean, arguments: Table, onDelivery: suspend (AMQPResponse.Channel.Message.Delivery) -> Unit, onCanceled: suspend (AMQPResponse.Channel) -> Unit): AMQPResponse.Channel.Basic.ConsumeOk

Consume messages from a queue by sending them to registered consume listeners. The caller is responsible for cancelling the consumer when done.

basicGet

open suspend override fun basicGet(queue: String, noAck: Boolean): AMQPResponse.Channel.Message.Get

Get a single message from a queue.

basicNack

open suspend override fun basicNack(message: AMQPMessage, multiple: Boolean, requeue: Boolean)

open suspend override fun basicNack(deliveryTag: ULong, multiple: Boolean, requeue: Boolean)

Reject a message.

basicPublish

open suspend override fun basicPublish(body: ByteArray, exchange: String, routingKey: String, mandatory: Boolean, immediate: Boolean, properties: Properties): AMQPResponse.Channel.Basic.Published

Publish a ByteArray message to exchange or queue.

basicQos

open suspend override fun basicQos(count: UShort, global: Boolean): AMQPResponse.Channel.Basic.QosOk

Sets a prefetch limit when consuming messages. No more messages will be delivered to the consumer until one or more messages have been acknowledged or rejected.

basicRecover

open suspend override fun basicRecover(requeue: Boolean): AMQPResponse.Channel.Basic.Recovered

Tell the broker what to do with all unacknowledged messages. Unacknowledged messages retrieved by basicGet are requeued regardless.

basicReject

open suspend override fun basicReject(message: AMQPMessage, requeue: Boolean)

open suspend override fun basicReject(deliveryTag: ULong, requeue: Boolean)

Reject a message.

cancelAll

open suspend fun cancelAll(channelClosed: AMQPException.ChannelClosed)

open suspend override fun close(reason: String, code: UShort): AMQPResponse.Channel.Closed

Closes the channel.

confirmSelect

open suspend override fun confirmSelect(): AMQPResponse.Channel.Confirm.Selected

Set channel in publish confirm mode, each published message will be acked or nacked.

consumerCount

open suspend override fun consumerCount(name: String): UInt

Returns the number of consumers subscribed to a queue.

exchangeBind

open suspend override fun exchangeBind(destination: String, source: String, routingKey: String, arguments: Table): AMQPResponse.Channel.Exchange.Bound

Bind an exchange to another exchange.

exchangeDeclare

open suspend override fun exchangeDeclare(name: String, type: String, durable: Boolean, autoDelete: Boolean, internal: Boolean, arguments: Table): AMQPResponse.Channel.Exchange.Declared

Declare an exchange.

exchangeDeclarePassive

open suspend override fun exchangeDeclarePassive(name: String): AMQPResponse.Channel.Exchange.Declared

Passively declare an exchange.

exchangeDelete

open suspend override fun exchangeDelete(name: String, ifUnused: Boolean): AMQPResponse.Channel.Exchange.Deleted

Delete an exchange.

exchangeUnbind

open suspend override fun exchangeUnbind(destination: String, source: String, routingKey: String, arguments: Table): AMQPResponse.Channel.Exchange.Unbound

Unbind an exchange from another exchange.

flow

open suspend override fun flow(active: Boolean): AMQPResponse.Channel.Flowed

Send a flow message to broker to start or stop sending messages to consumers. Warning: Not supported by all brokers.

messageCount

open suspend override fun messageCount(name: String): UInt

Returns the number of messages in a queue.

open

open suspend override fun open(): AMQPResponse.Channel.Opened

Opens the channel.

queueBind

open suspend override fun queueBind(queue: String, exchange: String, routingKey: String, arguments: Table): AMQPResponse.Channel.Queue.Bound

Binds a queue to an exchange.

queueDeclare

open suspend override fun queueDeclare(): AMQPResponse.Channel.Queue.Declared

Declare a server-named exclusive, autodelete, non-durable queue.

open suspend override fun queueDeclare(name: String, durable: Boolean, exclusive: Boolean, autoDelete: Boolean, arguments: Table): AMQPResponse.Channel.Queue.Declared

Declares a queue.

queueDeclarePassive

open suspend override fun queueDeclarePassive(name: String): AMQPResponse.Channel.Queue.Declared

Passively declares a queue.

queueDelete

open suspend override fun queueDelete(name: String, ifUnused: Boolean, ifEmpty: Boolean): AMQPResponse.Channel.Queue.Deleted

Deletes a queue.

queuePurge

open suspend override fun queuePurge(name: String): AMQPResponse.Channel.Queue.Purged

Deletes all messages from a queue.

queueUnbind

open suspend override fun queueUnbind(queue: String, exchange: String, routingKey: String, arguments: Table): AMQPResponse.Channel.Queue.Unbound

Unbinds a queue from an exchange.

txCommit

open suspend override fun txCommit(): AMQPResponse.Channel.Tx.Committed

Commit a transaction.

txRollback

open suspend override fun txRollback(): AMQPResponse.Channel.Tx.Rollbacked

Rollback a transaction.

txSelect

open suspend override fun txSelect(): AMQPResponse.Channel.Tx.Selected

Set channel in transaction mode.

write

open suspend override fun write(vararg frames: Frame)

Internal API to write raw frames to the channel.

writeAndWaitForResponse

inline suspend fun <T : AMQPResponse> writeAndWaitForResponse(vararg frames: Frame): T