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.close(res.ConsumerTag)
	} else {
		ch.consumers.close(consumer)
	}
	return nil
}
/*
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.
package amqp
import (
	"reflect"
	"sync"
	"sync/atomic"
)
const frameHeaderSize = 1 + 2 + 4 + 1
type Channel struct {
	destructor sync.Once
	m          sync.Mutex  
	confirmM   sync.Mutex  
	notifyM    sync.RWMutex
	connection *Connection
	rpc       chan message
	consumers *consumers
	id uint16
	closed int32
	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
	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,
		errors:     make(chan *Error, 1),
	}
}
func (ch *Channel) shutdown(e *Error) {
	ch.destructor.Do(func() {
		ch.m.Lock()
		defer ch.m.Unlock()
		ch.notifyM.Lock()
		defer ch.notifyM.Unlock()
		if e != nil {
			for _, c := range ch.closes {
				c <- e
			}
		}
		atomic.StoreInt32(&ch.closed, 1)
		if e != nil {
			ch.errors <- e
		}
		ch.consumers.closeAll()
		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)
		}
		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
	})
}
func (ch *Channel) send(msg message) (err error) {
	if atomic.LoadInt32(&ch.closed) == 1 {
		return ch.sendClosed(msg)
	}
	return ch.sendOpen(msg)
}
func (ch *Channel) open() error {
	return ch.call(&channelOpen{}, &channelOpenOk{})
}
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) {
						vres := reflect.ValueOf(try).Elem()
						vmsg := reflect.ValueOf(msg).Elem()
						vres.Set(vmsg)
						return nil
					}
				}
				return ErrCommandInvalid
			}
			return ErrClosed
		}
	}
	return nil
}
func (ch *Channel) sendClosed(msg message) (err error) {
	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()
		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
		}
		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
}
func (ch *Channel) dispatch(msg message) {
	switch m := msg.(type) {
	case *channelClose:
		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.close(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))
	default:
		ch.rpc <- msg
	}
}
func (ch *Channel) transition(f func(*Channel, frame) error) error {
	ch.recv = f
	return nil
}
func (ch *Channel) recvMethod(f frame) error {
	switch frame := f.(type) {
	case *methodFrame:
		if msg, ok := frame.Method.(messageWithContent); ok {
			ch.body = make([]byte, 0)
			ch.message = msg
			return ch.transition((*Channel).recvHeader)
		}
		ch.dispatch(frame.Method)  
		return ch.transition((*Channel).recvMethod)
	case *headerFrame:
		return ch.transition((*Channel).recvMethod)
	case *bodyFrame:
		return ch.transition((*Channel).recvMethod)
	}
	panic("unexpected frame type")
}
func (ch *Channel) recvHeader(f frame) error {
	switch frame := f.(type) {
	case *methodFrame:
		return ch.recvMethod(f)
	case *headerFrame:
		ch.header = frame
		if frame.Size == 0 {
			ch.message.setContent(ch.header.Properties, ch.body)
			ch.dispatch(ch.message)  
			return ch.transition((*Channel).recvMethod)
		}
		return ch.transition((*Channel).recvContent)
	case *bodyFrame:
		return ch.transition((*Channel).recvMethod)
	}
	panic("unexpected frame type")
}
func (ch *Channel) recvContent(f frame) error {
	switch frame := f.(type) {
	case *methodFrame:
		return ch.recvMethod(f)
	case *headerFrame:
		return ch.transition((*Channel).recvMethod)
	case *bodyFrame:
		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)  
			return ch.transition((*Channel).recvMethod)
		}
		return ch.transition((*Channel).recvContent)
	}
	panic("unexpected frame type")
}
func (ch *Channel) Close() error {
	defer ch.connection.closeChannel(ch, nil)
	return ch.call(
		&channelClose{ReplyCode: replySuccess},
		&channelCloseOk{},
	)
}
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
}
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
}
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
}
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
}
func (ch *Channel) NotifyConfirm(ack, nack chan uint64) (chan uint64, chan uint64) {
	confirms := ch.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 (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
}
func (ch *Channel) Qos(prefetchCount, prefetchSize int, global bool) error {
	return ch.call(
		&basicQos{
			PrefetchCount: uint16(prefetchCount),
			PrefetchSize:  uint32(prefetchSize),
			Global:        global,
		},
		&basicQosOk{},
	)
}