amqp091-go/channel.go at main · dominikbraun/amqp091-go

Repository navigation

10BC8

// Use of this source code is governed by a BSD-style

// license that can be found in the LICENSE file.

package amqp091

import (

"reflect"

"sync"

"sync/atomic"

)

// 0 1 3 7 size+7 size+8

// +------+---------+-------------+ +------------+ +-----------+

// +------+---------+-------------+ +------------+ +-----------+

// octet short long size octets octet

const frameHeaderSize = 1 + 2 + 4 + 1

Channel represents an AMQP channel. Used as a context for valid message

exchange. Errors on methods with this Channel as a receiver means this channel

should be discarded and a new channel established.

type Channel struct {

destructor sync.Once

m sync.Mutex // struct field mutex

confirmM sync.Mutex // publisher confirms state mutex

notifyM sync.RWMutex

connection *Connection

rpc chan message

consumers *consumers

id uint16

// closed is set to 1 when the channel has been closed - see Channel.send()

closed int32

// true when we will never notify again

noNotify bool

// Channel and Connection exceptions will be broadcast on these listeners.

closes []chan *Error

// Listeners for active=true flow control. When true is sent to a listener,

// publishing should pause until false is sent to listeners.

flows []chan bool

// Listeners for returned publishings for unroutable messages on mandatory

// publishings or undeliverable messages on immediate publishings.

returns []chan Return

// Listeners for when the server notifies the client that

// a consumer has been cancelled.

cancels []chan string

// Allocated when in confirm mode in order to track publish counter and order confirms

confirms *confirms

confirming bool

// Selects on any errors from shutdown during RPC

errors chan *Error

// State machine that manages frame order, must only be mutated by the connection

recv func(*Channel, frame)

// Current state for frame re-assembly, only mutated from recv

message messageWithContent

header *headerFrame

body []byte

}

// Constructs a new channel with the given framing rules

func newChannel(c *Connection, id uint16) *Channel {

return &Channel{

connection: c,

id: id,

rpc: make(chan message),

consumers: makeConsumers(),

confirms: newConfirms(),

recv: (*Channel).recvMethod,

errors: make(chan *Error, 1),

}

// shutdown is called by Connection after the channel has been removed from the

// connection registry.

func (ch *Channel) shutdown(e *Error) {

ch.destructor.Do(func() {

ch.m.Lock()

defer ch.m.Unlock()

// Grab an exclusive lock for the notify channels

ch.notifyM.Lock()

defer ch.notifyM.Unlock()

// Broadcast abnormal shutdown

if e != nil {

for _, c := range ch.closes {

c <- e

}

// Signal that from now on, Channel.send() should call

// Channel.sendClosed()

atomic.StoreInt32(&ch.closed, 1)

// Notify RPC if we're selecting

if e != nil {

ch.errors <- e

}

ch.consumers.close()

for _, c := range ch.closes {

close(c)

}

for _, c := range ch.flows {

close(c)

}

for _, c := range ch.returns {

close(c)

}

for _, c := range ch.cancels {

close(c)

}

// Set the slices to nil to prevent the dispatch() range from sending on

// the now closed channels after we release the notifyM mutex

ch.flows = nil

ch.closes = nil

ch.returns = nil

ch.cancels = nil

if ch.confirms != nil {

ch.confirms.Close()

}

close(ch.errors)

ch.noNotify = true

})

}

// send calls Channel.sendOpen() during normal operation.

// After the channel has been closed, send calls Channel.sendClosed(), ensuring

// only 'channel.close' is sent to the server.

func (ch *Channel) send(msg message) (err error) {

// If the channel is closed, use Channel.sendClosed()

if atomic.LoadInt32(&ch.closed) == 1 {

return ch.sendClosed(msg)

}

return ch.sendOpen(msg)

}

func (ch *Channel) open() error {

return ch.call(&channelOpen{}, &channelOpenOk{})

}

// Performs a request/response call for when the message is not NoWait and is

// specified as Synchronous.

func (ch *Channel) call(req message, res ...message) error {

if err := ch.send(req); err != nil {

return err

}

if req.wait() {

select {

case e, ok := <-ch.errors:

if ok {

return e

}

return ErrClosed

case msg := <-ch.rpc:

if msg != nil {

for _, try := range res {

if reflect.TypeOf(msg) == reflect.TypeOf(try) {

// *res = *msg

vres := reflect.ValueOf(try).Elem()

vmsg := reflect.ValueOf(msg).Elem()

vres.Set(vmsg)

return nil

}

return ErrCommandInvalid

}

// RPC channel has been closed without an error, likely due to a hard

// error on the Connection. This indicates we have already been

// shutdown and if were waiting, will have returned from the errors chan.

return ErrClosed

}

return nil

}

func (ch *Channel) sendClosed(msg message) (err error) {

// After a 'channel.close' is sent or received the on 10BC8 ly valid response is

// channel.close-ok

if _, ok := msg.(*channelCloseOk); ok {

return ch.connection.send(&methodFrame{

ChannelId: ch.id,

Method: msg,

})

}

return ErrClosed

}

func (ch *Channel) sendOpen(msg message) (err error) {

if content, ok := msg.(messageWithContent); ok {

props, body := content.getContent()

class, _ := content.id()

// catch client max frame size==0 and server max frame size==0

// set size to length of what we're trying to publish

var size int

if ch.connection.Config.FrameSize > 0 {

size = ch.connection.Config.FrameSize - frameHeaderSize

} else {

size = len(body)

}

if err = ch.connection.send(&methodFrame{

ChannelId: ch.id,

Method: content,

}); err != nil {

return

}

if err = ch.connection.send(&headerFrame{

ChannelId: ch.id,

ClassId: class,

Size: uint64(len(body)),

Properties: props,

}); err != nil {

return

}

// chunk body into size (max frame size - frame header size)

for i, j := 0, size; i < len(body); i, j = j, j+size {

if j > len(body) {

j = len(body)

}

if err = ch.connection.send(&bodyFrame{

ChannelId: ch.id,

Body: body[i:j],

}); err != nil {

return

}

} else {

err = ch.connection.send(&methodFrame{

ChannelId: ch.id,

Method: msg,

})

}

return

}

// Eventually called via the state machine from the connection's reader

// goroutine, so assumes serialized access.

func (ch *Channel) dispatch(msg message) {

switch m := msg.(type) {

case *channelClose:

// lock before sending connection.close-ok

// to avoid unexpected interleaving with basic.publish frames if

// publishing is happening concurrently

ch.m.Lock()

ch.send(&channelCloseOk{})

ch.m.Unlock()

ch.connection.closeChannel(ch, newError(m.ReplyCode, m.ReplyText))

case *channelFlow:

ch.notifyM.RLock()

for _, c := range ch.flows {

c <- m.Active

}

ch.notifyM.RUnlock()

ch.send(&channelFlowOk{Active: m.Active})

case *basicCancel:

ch.notifyM.RLock()

for _, c := range ch.cancels {

c <- m.ConsumerTag

}

ch.notifyM.RUnlock()

ch.consumers.cancel(m.ConsumerTag)

case *basicReturn:

ret := newReturn(*m)

ch.notifyM.RLock()

for _, c := range ch.returns {

c <- *ret

}

ch.notifyM.RUnlock()

case *basicAck:

if ch.confirming {

if m.Multiple {

ch.confirms.Multiple(Confirmation{m.DeliveryTag, true})

} else {

ch.confirms.One(Confirmation{m.DeliveryTag, true})

}

case *basicNack:

if ch.confirming {

if m.Multiple {

ch.confirms.Multiple(Confirmation{m.DeliveryTag, false})

} else {

ch.confirms.One(Confirmation{m.DeliveryTag, false})

}

case *basicDeliver:

ch.consumers.send(m.ConsumerTag, newDelivery(ch, m))

// TODO log failed consumer and close channel, this can happen when

// deliveries are in flight and a no-wait cancel has happened

default:

ch.rpc <- msg

}

func (ch *Channel) transition(f func(*Channel, frame)) {

ch.recv = f

}

func (ch *Channel) recvMethod(f frame) {

switch frame := f.(type) {

case *methodFrame:

if msg, ok := frame.Method.(messageWithContent); ok {

ch.body = make([]byte, 0)

ch.message = msg

ch.transition((*Channel).recvHeader)

return

}

ch.dispatch(frame.Method) // termination state

ch.transition((*Channel).recvMethod)

case *headerFrame:

// drop

ch.transition((*Channel).recvMethod)

case *bodyFrame:

// drop

ch.transition((*Channel).recvMethod)

default:

panic("unexpected frame type")

}

func (ch *Channel) recvHeader(f frame) {

switch frame := f.(type) {

case *methodFrame:

// interrupt content and handle method

ch.recvMethod(f)

case *headerFrame:

// start collecting if we expect body frames

ch.header = frame

if frame.Size == 0 {

ch.message.setContent(ch.header.Properties, ch.body)

ch.dispatch(ch.message) // termination state

ch.transition((*Channel).recvMethod)

return

}

ch.transition((*Channel).recvContent)

case *bodyFrame:

// drop and reset

ch.transition((*Channel).recvMethod)

default:

panic("unexpected frame type")

}

// state after method + header and before the length

// defined by the header has been reached

func (ch *Channel) recvContent(f frame) {

switch frame := f.(type) {

case *methodFrame:

// interrupt content and handle method

ch.recvMethod(f)

case *headerFrame:

// drop and reset

ch.transition((*Channel).recvMethod)

case *bodyFrame:

if cap(ch.body) == 0 {

ch.body = make([]byte, 0, ch.header.Size)

}

ch.body = append(ch.body, frame.Body...)

if uint64(len(ch.body)) >= ch.header.Size {

ch.message.setContent(ch.header.Properties, ch.body)

ch.dispatch(ch.message) // termination state

ch.transition((*Channel).recvMethod)

return

}

ch.transition((*Channel).recvContent)

default:

panic("unexpected frame type")

}

Close initiate a clean channel closure by sending a close message with the error

code set to '200'.

It is safe to call this method multiple times.

func (ch *Channel) Close() error {

defer ch.connection.closeChannel(ch, nil)

return ch.call(

&channelClose{ReplyCode: replySuccess},

&channelCloseOk{},

)

}

// IsClosed returns true if the channel is marked as closed, otherwise false

// is returned.

func (ch *Channel) IsClosed() bool {

return atomic.LoadInt32(&ch.closed) == 1

}

NotifyClose registers a listener for when the server sends a channel or

connection exception in the form of a Connection.Close or Channel.Close method.

Connection exceptions will be broadcast to all open channels and all channels

will be closed, where channel exceptions will only be broadcast to listeners to

this channel.

The chan provided will be closed when the Channel is closed and on a

graceful close, no error will be sent.

In case of a non graceful close the error will be notified synchronously by the library

so that it will be necessary to consume the Channel from the caller in order to avoid deadlocks

func (ch *Channel) NotifyClose(c chan *Error) chan *Error {

ch.notifyM.Lock()

defer ch.notifyM.Unlock()

if ch.noNotify {

close(c)

} else {

ch.closes = append(ch.closes, c)

}

return c

}

NotifyFlow registers a listener for basic.flow methods sent by the server.

When `false` is sent on one of the listener channels, all publishers should

pause until a `true` is sent.

The server may ask the producer to pause or restart the flow of Publishings

sent by on a channel. This is a simple flow-control mechanism that a server can

use to avoid overflowing its queues or otherwise finding itself receiving more

messages than it can process. Note that this method is not intended for window

control. It does not affect contents returned by basic.get-ok methods.

When a new channel is opened, it is active (flow is active). Some

applications assume that channels are inactive until started. To emulate

this behavior a client MAY open the channel, then pause it.

Publishers should respond to a flow messages as rapidly as possible and the

server may disconnect over producing channels that do not respect these

messages.

basic.flow-ok methods will always be returned to the server regardless of

the number of listeners there are.

To control the flow of deliveries from the server, use the Channel.Flow()

method instead.

Note: RabbitMQ will rather use TCP pushback on the network connection instead

of sending basic.flow. This means that if a single channel is producing too

much on the same connection, all channels using that connection will suffer,

including acknowledgments from deliveries. Use different Connections if you

desire to interleave consumers and producers in the same process to avoid your

basic.ack messages from getting rate limited with your basic.publish messages.

func (ch *Channel) NotifyFlow(c chan bool) chan bool {

ch.notifyM.Lock()

defer ch.notifyM.Unlock()

if ch.noNotify {

close(c)

} else {

ch.flows = append(ch.flows, c)

}

return c

}

NotifyReturn registers a listener for basic.return methods. These can be sent

from the server when a publish is undeliverable either from the mandatory or

immediate flags.

A return struct has a copy of the Publishing along with some error

information about why the publishing failed.

func (ch *Channel) NotifyReturn(c chan Return) chan Return {

ch.notifyM.Lock()

defer ch.notifyM.Unlock()

if ch.noNotify {

close(c)

} else {

ch.returns = append(ch.returns, c)

}

return c

}

NotifyCancel registers a listener for basic.cancel methods. These can be sent

from the server when a queue is deleted or when consuming from a mirrored queue

where the master has just failed (and was moved to another node).

The subscription tag is returned to the listener.

func (ch *Channel) NotifyCancel(c chan string) chan string {

ch.notifyM.Lock()

defer ch.notifyM.Unlock()

if ch.noNotify {

close(c)

} else {

ch.cancels = append(ch.cancels, c)

}

return c

}

NotifyConfirm calls NotifyPublish and starts a goroutine sending

ordered Ack and Nack DeliveryTag to the respective channels.

For strict ordering, use NotifyPublish instead.

func (ch *Channel) NotifyConfirm(ack, nack chan uint64) (chan uint64, chan uint64) {

confirms := ch.NotifyPublish(make(chan Confirmation, cap(ack)+cap(nack)))

go func() {

for c := range confirms {

if c.Ack {

ack <- c.DeliveryTag

} else {

nack <- c.DeliveryTag

}

close(ack)

if nack != ack {

close(nack)

}

}()

return ack, nack

}

NotifyPublish registers a listener for reliable publishing. Receives from this

chan for every publish after Channel.Confirm will be in order starting with

DeliveryTag 1.

There will be one and only one Confirmation Publishing starting with the

delivery tag of 1 and progressing sequentially until the total number of

Publishings have been seen by the server.

Acknowledgments will be received in the order of delivery from the

NotifyPublish channels even if the server acknowledges them out of order.

The listener chan will be closed when the Channel is closed.

The capacity of the chan Confirmation must be at least as large as the

number of outstanding publishings. Not having enough buffered chans will

create a deadlock if you attempt to perform other operations on the Connection

or Channel while confirms are in-flight.

It's advisable to wait for all Confirmations to arrive before calling

Channel.Close() or Connection.Close().

It is also advisable for the caller to consume from the channel returned till it is closed

to avoid possible deadlocks

func (ch *Channel) NotifyPublish(confirm chan Confirmation) chan Confirmation {

ch.notifyM.Lock()

defer ch.notifyM.Unlock()

if ch.noNotify {

close(confirm)

} else {

ch.confirms.Listen(confirm)

}

return confirm

}

Qos controls how many messages or how many bytes the server will try to keep on

the network for consumers before receiving delivery acks. The intent of Qos is

to make sure the network buffers stay full between the server and client.

With a prefetch count greater than zero, the server will deliver that many

messages to consumers before acknowledgments are received. The server ignores

this option when consumers are started with noAck because no acknowledgments

are expected or sent.

With a prefetch size greater than zero, the server will try to keep at least

that many bytes of deliveries flushed to the network before receiving

acknowledgments from the consumers. This option is ignored when consumers are

started with noAck.

When global is true, these Qos settings apply to all existing and future

consumers on all channels on the same connection. When false, the Channel.Qos

settings will apply to all existing and future consumers on this channel.

Please see the RabbitMQ Consumer Prefetch documentation for an explanation of

how the global flag is implemented in RabbitMQ, as it differs from the

AMQP 0.9.1 specification in that global Qos settings are limited in scope to

channels, not connections (https://www.rabbitmq.com/consumer-prefetch.html).

To get round-robin behavior between consumers consuming from the same queue on

different connections, set the prefetch count to 1, and the next available

message on the server will be delivered to the next available consumer.

If your consumer work time is reasonably consistent and not much greater

than two times your network round trip time, you will see significant

throughput improvements starting with a prefetch count of 2 or slightly

greater as described by benchmarks on RabbitMQ.

http://www.rabbitmq.com/blog/2012/04/25/rabbitmq-performance-measurements-part-2/

func (ch *Channel) Qos(prefetchCount, prefetchSize int, global bool) error {

return ch.call(

&basicQos{

PrefetchCount: uint16(prefetchCount),

PrefetchSize: uint32(prefetchSize),

Global: global,

&basicQosOk{},

)

}

Cancel stops deliveries to the consumer chan established in Channel.Consume and

identified by consumer.

Only use this method to cleanly stop receiving deliveries from the server and

cleanly shut down the consumer chan identified by this tag. Using this method

and waiting for remaining messages to flush from the consumer chan will ensure

all messages received on the network will be delivered to the receiver of your

consumer chan.

Continue consuming from the chan Delivery provided by Channel.Consume until the

chan closes.

When noWait is true, do not wait for the server to acknowledge the cancel.

Only use this when you are certain there are no deliveries in flight that

require an acknowledgment, otherwise they will arrive and be dropped in the

client without an ack, and will not be redelivered to other consumers.

func (ch *Channel) Cancel(consumer string, noWait bool) error {

req := &basicCancel{

ConsumerTag: consumer,

NoWait: noWait,

}

res := &basicCancelOk{}

if err := ch.call(req, res); err != nil {

return err

}

if req.wait() {

ch.consumers.cancel(res.ConsumerTag)

} else {

// Potentially could drop deliveries in flight

ch.consumers.cancel(consumer)

}

return nil

}

QueueDeclare declares a queue to hold messages and deliver to consumers.

Declaring creates a queue if it doesn't already exist, or ensures that an

existing queue matches the same parameters.

Every queue declared gets a default binding to the empty exchange "" which has

the type "direct" with the routing key matching the queue's name. With this

default binding, it is possible to publish messages that route directly to

this queue by publishing to "" with the routing key of the queue name.

QueueDeclare("alerts", true, false, false, false, nil)

Publish("", "alerts", false, false, Publishing{Body: []byte("...")})

Delivery Exchange Key Queue

-----------------------------------------------

key: alerts -> "" -> alerts -> alerts

The queue name may be empty, in which case the server will generate a unique name

which will be returned in the Name field of Queue struct.

Durable and Non-Auto-Deleted queues will survive server restarts and remain

when there are no remaining consumers or bindings. Persistent publishings will

be restored in this queue on server restart. These queues are only able to be

bound to durable exchanges.

Non-Durable and Auto-Deleted queues will not be redeclared on server restart

and will be deleted by the server after a short time when the last consumer is

canceled or the last consumer's channel is closed. Queues with this lifetime

can also be deleted normally with QueueDelete. These durable queues can only

be bound to non-durable exchanges.

Non-Durable and Non-Auto-Deleted queues will remain declared as long as the

server is running regardless of how many consumers. This lifetime is useful

for temporary topologies that may have long delays between consumer activity.

These queues can only be bound to non-durable exchanges.

Durable and Auto-Deleted queues will be restored on server restart, but without< 10BC8 /div>

active consumers will not survive and be removed. This Lifetime is unlikely

to be useful.

Exclusive queues are only accessible by the connection that declares them and

will be deleted when the connection closes. Channels on other connections

will receive an error when attempting to declare, bind, consume, purge or

delete a queue with the same name.

When noWait is true, the queue will assume to be declared on the server. A

channel exception will arrive if the conditions are met for existing queues

or attempting to modify an existing queue from a different connection.

When the error return value is not nil, you can assume the queue could not be

declared with these parameters, and the channel will be closed.

func (ch *Channel) QueueDeclare(name string, durable, autoDelete, exclusive, noWait bool, args Table) (Queue, error) {

if err := args.Validate(); err != nil {

return Queue{}, err

}

req := &queueDeclare{

Queue: name,

Passive: false,

Durable: durable,

AutoDelete: autoDelete,

Exclusive: exclusive,

NoWait: noWait,

Arguments: args,

}

res := &queueDeclareOk{}

if err := ch.call(req, res); err != nil {

return Queue{}, err

}

if req.wait() {

return Queue{

Name: res.Queue,

Messages: int(res.MessageCount),

Consumers: int(res.ConsumerCount),

}, nil

}

return Queue{Name: name}, nil

}

QueueDeclarePassive is functionally and parametrically equivalent to

QueueDeclare, except that it sets the "passive" attribute to true. A passive

queue is assumed by RabbitMQ to already exist, and attempting to connect to a

non-existent queue will cause RabbitMQ to throw an exception. This function

can be used to test for the existence of a queue.

func (ch *Channel) QueueDeclarePassive(name string, durable, autoDelete, exclusive, noWait bool, args Table) (Queue, error) {

if err := args.Validate(); err != nil {

return Queue{}, err

}

req := &queueDeclare{

Queue: name,

Passive: true,

Durable: durable,

AutoDelete: autoDelete,

Exclusive: exclusive,

NoWait: noWait,

Arguments: args,

}

res := &queueDeclareOk{}

if err := ch.call(req, res); err != nil {

return Queue{}, err

}

if req.wait() {

return Queue{

Name: res.Queue,

Messages: int(res.MessageCount),

Consumers: int(res.ConsumerCount),

}, nil

}

return Queue{Name: name}, nil

}

QueueInspect passively declares a queue by name to inspect the current message

count and consumer count.

Use this method to check how many messages ready for delivery reside in the queue,

how many consumers are receiving deliveries, and whether a queue by this

name already exists.

If the queue by this name exists, use Channel.QueueDeclare check if it is

declared with specific parameters.

If a queue by this name does not exist, an error will be returned and the

channel will be closed.

func (ch *Channel) QueueInspect(name string) (Queue, error) {

req := &queueDeclare{

Queue: name,

Passive: true,

}

res := &queueDeclareOk{}

err := ch.call(req, res)

state := Queue{

Name: name,

Messages: int(res.MessageCount),

Consumers: int(res.ConsumerCount),

}

return state, err

}

QueueBind binds an exchange to a queue so that publishings to the exchange will

be routed to the queue when the publishing routing key matches the binding

routing key.

QueueBind("pagers", "alert", "log", false, nil)

QueueBind("emails", "info", "log", false, nil)

Delivery Exchange Key Queue

-----------------------------------------------

key: alert --> log ----> alert --> pagers

key: info ---> log ----> info ---> emails

key: debug --> log (none) (dropped)

If a binding with the same key and arguments already exists between the

exchange and queue, the attempt to rebind will be ignored and the existing

binding will be retained.

In the case that multiple bindings may cause the message to be routed to the

same queue, the server will only route the publishing once. This is possible

with topic exchanges.

QueueBind("pagers", "alert", "amq.topic", false, nil)

QueueBind("emails", "info", "amq.topic", false, nil)

QueueBind("emails", "#", "amq.topic", false, nil) // match everything

Delivery Exchange Key Queue

-----------------------------------------------

key: alert --> amq.topic ----> alert --> pagers

key: info ---> amq.topic ----> # ------> emails

\---> info ---/

key: debug --> amq.topic ----> # ------> emails

It is only possible to bind a durable queue to a durable exchange regardless of

whether the queue or exchange is auto-deleted. Bindings between durable queues

and exchanges will also be restored on server restart.

If the binding could not complete, an error will be returned and the channel

will be closed.

When noWait is false and the queue could not be bound, the channel will be

closed with an error.

func (ch *Channel) QueueBind(name, key, exchange string, noWait bool, args Table) error {

if err := args.Validate(); err != nil {

return err

}

return ch.call(

&queueBind{

Queue: name,

Exchange: exchange,

RoutingKey: key,

NoWait: noWait,

Arguments: args,

&queueBindOk{},

)

}

QueueUnbind removes a binding between an exchange and queue matching the key and

arguments.

It is possible to send and empty string for the exchange name which means to

unbind the queue from the default exchange.

func (ch *Channel) QueueUnbind(name, key, exchange string, args Table) error {

if err := args.Validate(); err != nil {

return err

}

return ch.call(

&queueUnbind{

Queue: name,

Exchange: exchange,

RoutingKey: key,

Arguments: args,

&queueUnbindOk{},

)

}

QueuePurge removes all messages from the named queue which are not waiting to

be acknowledged. Messages that have been delivered but have not yet been

acknowledged will not be removed.

When successful, returns the number of messages purged.

If noWait is true, do not wait for the server response and the number of

messages purged will not be meaningful.

func (ch *Channel) QueuePurge(name string, noWait bool) (int, error) {

req := &queuePurge{

Queue: name,

NoWait: noWait,

}

res := &queuePurgeOk{}

err := ch.call(req, res)

return int(res.MessageCount), err

}

QueueDelete removes the queue from the server including all bindings then

purges the messages based on server configuration, returning the number of

messages purged.

When ifUnused is true, the queue will not be deleted if there are any

consumers on the queue. If there are consumers, an error will be returned and

the channel will be closed.

When ifEmpty is true, the queue will not be deleted if there are any messages

remaining on the queue. If there are messages, an error will be returned and

the channel will be closed.

When noWait is true, the queue will be deleted without waiting for a response

from the server. The purged message count will not be meaningful. If the queue

could not be deleted, a channel exception will be raised and the channel will

be closed.

func (ch *Channel) QueueDelete(name string, ifUnused, ifEmpty, noWait bool) (int, error) {

req := &queueDelete{

Queue: name,

View remainder of file in raw view

FilesExpand file tree

channel.go

Latest commit

History

channel.go

File metadata and controls