amqp-client-opentelemetry/dev.kourier.amqp.opentelemetry/OpenTelemetryAMQPChannel

OpenTelemetryAMQPChannel

class OpenTelemetryAMQPChannel(delegate: AMQPChannel, tracer: Tracer, config: TracingConfig = TracingConfig.default()) : AMQPChannel

OpenTelemetry-instrumented wrapper around AMQPChannel.

This class automatically creates spans for message publishing and consumption operations, and propagates trace context through AMQP message headers using W3C Trace Context format.

Constructors

constructor(delegate: AMQPChannel, tracer: Tracer, config: TracingConfig = TracingConfig.default())

Properties

channelClosed

open override val channelClosed: Deferred<AMQPException.ChannelClosed>

closedResponses

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

flowResponses

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

open override val id: ChannelId

isConfirmMode

open override val isConfirmMode: Boolean

isTxMode

open override val isTxMode: Boolean

openedResponses

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

publishConfirmResponses

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

returnResponses

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

state

open override val state: ConnectionState

Functions

basicAck

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

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

basicCancel

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

basicConsume

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

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

basicGet

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

basicNack

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

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

basicPublish

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

basicQos

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

basicRecover

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

basicReject

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

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

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

confirmSelect

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

confirmSelectNoWait

open suspend override fun confirmSelectNoWait()

consumerCount

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

exchangeBind

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

exchangeBindNoWait

open suspend override fun exchangeBindNoWait(destination: String, source: String, routingKey: String, arguments: Table)

exchangeDeclare

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

exchangeDeclareNoWait

open suspend override fun exchangeDeclareNoWait(name: String, type: String, durable: Boolean, autoDelete: Boolean, internal: Boolean, arguments: Table)

exchangeDeclarePassive

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

exchangeDeclarePassiveNoWait

open suspend override fun exchangeDeclarePassiveNoWait(name: String)

exchangeDelete

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

exchangeDeleteNoWait

open suspend override fun exchangeDeleteNoWait(name: String, ifUnused: Boolean)

exchangeUnbind

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

exchangeUnbindNoWait

open suspend override fun exchangeUnbindNoWait(destination: String, source: String, routingKey: String, arguments: Table)

flow

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

messageCount

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

open

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

queueBind

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

queueBindNoWait

open suspend override fun queueBindNoWait(queue: String, exchange: String, routingKey: String, arguments: Table)

queueDeclare

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

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

queueDeclareNoWait

open suspend override fun queueDeclareNoWait(name: String, durable: Boolean, exclusive: Boolean, autoDelete: Boolean, arguments: Table)

queueDeclarePassive

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

queueDeclarePassiveNoWait

open suspend override fun queueDeclarePassiveNoWait(name: String)

queueDelete

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

queueDeleteNoWait

open suspend override fun queueDeleteNoWait(name: String, ifUnused: Boolean, ifEmpty: Boolean)

queuePurge

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

queuePurgeNoWait

open suspend override fun queuePurgeNoWait(name: String)

queueUnbind

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

txCommit

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

txRollback

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

txSelect

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

withTracing

fun AMQPChannel.withTracing(tracer: Tracer, tracingConfig: TracingConfig = TracingConfig.default()): AMQPChannel

Wraps an existing channel with OpenTelemetry instrumentation.

write

open suspend override fun write(vararg frames: Frame)