func (me *Channel) QueuePurge(name string, noWait bool) (int, error) {
	req := &queuePurge{
		Queue:  name,
		NoWait: noWait,
	}
	res := &queuePurgeOk{}
	err := me.call(req, res)
	return int(res.MessageCount), err
}
/*
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.
package amqp
import (
	"reflect"
	"sync"
)
const frameHeaderSize = 1 + 2 + 4 + 1
type Channel struct {
	destructor sync.Once
	sendM      sync.Mutex  
	m          sync.Mutex  
	connection *Connection
	rpc       chan message
	consumers *consumers
	id uint16
	noNotify bool
	closes []chan *Error
	flows []chan bool
	returns []chan Return
	cancels []chan string
	confirms   *confirms
	confirming bool
	errors chan *Error
	recv func(*Channel, frame) error
	send func(*Channel, message) error
	message messageWithContent
	header  *headerFrame
	body    []byte
}
func newChannel(c *Connection, id uint16) *Channel {
	return &Channel{
		connection: c,
		id:         id,
		rpc:        make(chan message),
		consumers:  makeConsumers(),
		confirms:   newConfirms(),
		recv:       (*Channel).recvMethod,
		send:       (*Channel).sendOpen,
		errors:     make(chan *Error, 1),
	}
}
func (me *Channel) shutdown(e *Error) {
	me.destructor.Do(func() {
		me.m.Lock()
		defer me.m.Unlock()
		if e != nil {
			for _, c := range me.closes {
				c <- e
			}
		}
		me.send = (*Channel).sendClosed
		if e != nil {
			me.errors <- e
		}
		me.consumers.closeAll()
		for _, c := range me.closes {
			close(c)
		}
		for _, c := range me.flows {
			close(c)
		}
		for _, c := range me.returns {
			close(c)
		}
		for _, c := range me.cancels {
			close(c)
		}
		if me.confirms != nil {
			me.confirms.Close()
		}
		me.noNotify = true
	})
}
func (me *Channel) open() error {
	return me.call(&channelOpen{}, &channelOpenOk{})
}
func (me *Channel) call(req message, res ...message) error {
	if err := me.send(me, req); err != nil {
		return err
	}
	if req.wait() {
		select {
		case e := <-me.errors:
			return e
		case msg := <-me.rpc:
			if msg != nil {
				for _, try := range res {
					if reflect.TypeOf(msg) == reflect.TypeOf(try) {
						vres := reflect.ValueOf(try).Elem()
						vmsg := reflect.ValueOf(msg).Elem()
						vres.Set(vmsg)
						return nil
					}
				}
				return ErrCommandInvalid
			} else {
				return ErrClosed
			}
		}
	}
	return nil
}
func (me *Channel) sendClosed(msg message) (err error) {
	me.sendM.Lock()
	defer me.sendM.Unlock()
	if _, ok := msg.(*channelCloseOk); ok {
		return me.connection.send(&methodFrame{
			ChannelId: me.id,
			Method:    msg,
		})
	}
	return ErrClosed
}
func (me *Channel) sendOpen(msg message) (err error) {
	me.sendM.Lock()
	defer me.sendM.Unlock()
	if content, ok := msg.(messageWithContent); ok {
		props, body := content.getContent()
		class, _ := content.id()
		var size int
		if me.connection.Config.FrameSize > 0 {
			size = me.connection.Config.FrameSize - frameHeaderSize
		} else {
			size = len(body)
		}
		if err = me.connection.send(&methodFrame{
			ChannelId: me.id,
			Method:    content,
		}); err != nil {
			return
		}
		if err = me.connection.send(&headerFrame{
			ChannelId:  me.id,
			ClassId:    class,
			Size:       uint64(len(body)),
			Properties: props,
		}); err != nil {
			return
		}
		for i, j := 0, size; i < len(body); i, j = j, j+size {
			if j > len(body) {
				j = len(body)
			}
			if err = me.connection.send(&bodyFrame{
				ChannelId: me.id,
				Body:      body[i:j],
			}); err != nil {
				return
			}
		}
	} else {
		err = me.connection.send(&methodFrame{
			ChannelId: me.id,
			Method:    msg,
		})
	}
	return
}
func (me *Channel) dispatch(msg message) {
	switch m := msg.(type) {
	case *channelClose:
		me.connection.closeChannel(me, newError(m.ReplyCode, m.ReplyText))
		me.send(me, &channelCloseOk{})
	case *channelFlow:
		for _, c := range me.flows {
			c <- m.Active
		}
		me.send(me, &channelFlowOk{Active: m.Active})
	case *basicCancel:
		for _, c := range me.cancels {
			c <- m.ConsumerTag
		}
		me.send(me, &basicCancelOk{ConsumerTag: m.ConsumerTag})
	case *basicReturn:
		ret := newReturn(*m)
		for _, c := range me.returns {
			c <- *ret
		}
	case *basicAck:
		if me.confirming {
			if m.Multiple {
				me.confirms.Multiple(Confirmation{m.DeliveryTag, true})
			} else {
				me.confirms.One(Confirmation{m.DeliveryTag, true})
			}
		}
	case *basicNack:
		if me.confirming {
			if m.Multiple {
				me.confirms.Multiple(Confirmation{m.DeliveryTag, false})
			} else {
				me.confirms.One(Confirmation{m.DeliveryTag, false})
			}
		}
	case *basicDeliver:
		me.consumers.send(m.ConsumerTag, newDelivery(me, m))
	default:
		me.rpc <- msg
	}
}
func (me *Channel) transition(f func(*Channel, frame) error) error {
	me.recv = f
	return nil
}
func (me *Channel) recvMethod(f frame) error {
	switch frame := f.(type) {
	case *methodFrame:
		if msg, ok := frame.Method.(messageWithContent); ok {
			me.body = make([]byte, 0)
			me.message = msg
			return me.transition((*Channel).recvHeader)
		}
		me.dispatch(frame.Method)  
		return me.transition((*Channel).recvMethod)
	case *headerFrame:
		return me.transition((*Channel).recvMethod)
	case *bodyFrame:
		return me.transition((*Channel).recvMethod)
	default:
		panic("unexpected frame type")
	}
	panic("unreachable")
}
func (me *Channel) recvHeader(f frame) error {
	switch frame := f.(type) {
	case *methodFrame:
		return me.recvMethod(f)
	case *headerFrame:
		me.header = frame
		if frame.Size == 0 {
			me.message.setContent(me.header.Properties, me.body)
			me.dispatch(me.message)  
			return me.transition((*Channel).recvMethod)
		} else {
			return me.transition((*Channel).recvContent)
		}
	case *bodyFrame:
		return me.transition((*Channel).recvMethod)
	default:
		panic("unexpected frame type")
	}
	panic("unreachable")
}
func (me *Channel) recvContent(f frame) error {
	switch frame := f.(type) {
	case *methodFrame:
		return me.recvMethod(f)
	case *headerFrame:
		return me.transition((*Channel).recvMethod)
	case *bodyFrame:
		me.body = append(me.body, frame.Body...)
		if uint64(len(me.body)) >= me.header.Size {
			me.message.setContent(me.header.Properties, me.body)
			me.dispatch(me.message)  
			return me.transition((*Channel).recvMethod)
		}
		return me.transition((*Channel).recvContent)
	default:
		panic("unexpected frame type")
	}
	panic("unreachable")
}
func (me *Channel) Close() error {
	defer me.connection.closeChannel(me, nil)
	return me.call(
		&channelClose{ReplyCode: replySuccess},
		&channelCloseOk{},
	)
}
func (me *Channel) NotifyClose(c chan *Error) chan *Error {
	me.m.Lock()
	defer me.m.Unlock()
	if me.noNotify {
		close(c)
	} else {
		me.closes = append(me.closes, c)
	}
	return c
}
func (me *Channel) NotifyFlow(c chan bool) chan bool {
	me.m.Lock()
	defer me.m.Unlock()
	if me.noNotify {
		close(c)
	} else {
		me.flows = append(me.flows, c)
	}
	return c
}
func (me *Channel) NotifyReturn(c chan Return) chan Return {
	me.m.Lock()
	defer me.m.Unlock()
	if me.noNotify {
		close(c)
	} else {
		me.returns = append(me.returns, c)
	}
	return c
}
func (me *Channel) NotifyCancel(c chan string) chan string {
	me.m.Lock()
	defer me.m.Unlock()
	if me.noNotify {
		close(c)
	} else {
		me.cancels = append(me.cancels, c)
	}
	return c
}
func (me *Channel) NotifyConfirm(ack, nack chan uint64) (chan uint64, chan uint64) {
	confirms := me.NotifyPublish(make(chan Confirmation, len(ack)+len(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
}
func (me *Channel) NotifyPublish(confirm chan Confirmation) chan Confirmation {
	me.m.Lock()
	defer me.m.Unlock()
	if me.noNotify {
		close(confirm)
	} else {
		me.confirms.Listen(confirm)
	}
	return confirm
}
func (me *Channel) Qos(prefetchCount, prefetchSize int, global bool) error {
	return me.call(
		&basicQos{
			PrefetchCount: uint16(prefetchCount),
			PrefetchSize:  uint32(prefetchSize),
			Global:        global,
		},
		&basicQosOk{},
	)
}
func (me *Channel) Cancel(consumer string, noWait bool) error {
	req := &basicCancel{
		ConsumerTag: consumer,
		NoWait:      noWait,
	}
	res := &basicCancelOk{}
	if err := me.call(req, res); err != nil {
		return err
	}
	if req.wait() {
		me.consumers.close(res.ConsumerTag)
	} else {
		me.consumers.close(consumer)
	}
	return nil
}
func (me *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 := me.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
	panic("unreachable")
}
func (me *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 := me.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
	panic("unreachable")
}
func (me *Channel) QueueInspect(name string) (Queue, error) {
	req := &queueDeclare{
		Queue:   name,
		Passive: true,
	}
	res := &queueDeclareOk{}
	err := me.call(req, res)
	state := Queue{
		Name:      name,
		Messages:  int(res.MessageCount),
		Consumers: int(res.ConsumerCount),
	}
	return state, err
}
func (me *Channel) QueueBind(name, key, exchange string, noWait bool, args Table) error {
	if err := args.Validate(); err != nil {
		return err
	}
	return me.call(
		&queueBind{
			Queue:      name,
			Exchange:   exchange,
			RoutingKey: key,
			NoWait:     noWait,
			Arguments:  args,
		},
		&queueBindOk{},
	)
}
func (me *Channel) QueueUnbind(name, key, exchange string, args Table) error {
	if err := args.Validate(); err != nil {
		return err
	}
	return me.call(
		&queueUnbind{
			Queue:      name,
			Exchange:   exchange,
			RoutingKey: key,
			Arguments:  args,
		},
		&queueUnbindOk{},
	)
}