func (me *Connection) demux(f frame) {
	if f.channel() == 0 {
		me.dispatch0(f)
	} else {
		me.dispatchN(f)
	}
}
All methods sent to the connection channel should be synchronous so we
can handle them directly without a framing component
package amqp
import (
	"bufio"
	"crypto/tls"
	"io"
	"net"
	"reflect"
	"strconv"
	"strings"
	"sync"
	"time"
)
const (
	maxChannelMax = (2 << 15) - 1
	defaultHeartbeat         = 10 * time.Second
	defaultConnectionTimeout = 30 * time.Second
	defaultProduct           = "https://github.com/streadway/amqp"
	defaultVersion           = "β"
	defaultChannelMax        = maxChannelMax
)
type Config struct {
	SASL []Authentication
	Vhost string
	ChannelMax int            
	FrameSize  int            
	Heartbeat  time.Duration  
	TLSClientConfig *tls.Config
	Properties Table
	Dial func(network, addr string) (net.Conn, error)
}
type Connection struct {
	destructor sync.Once   
	sendM      sync.Mutex  
	m          sync.Mutex  
	conn io.ReadWriteCloser
	rpc       chan message
	writer    *writer
	sends     chan time.Time      
	deadlines chan readDeadliner  
	allocator *allocator  
	channels  map[uint16]*Channel
	noNotify bool  
	closes   []chan *Error
	blocks   []chan Blocking
	errors chan *Error
	Config Config  
	Major      int    
	Minor      int    
	Properties Table  
}
type readDeadliner interface {
	SetReadDeadline(time.Time) error
}
type localNetAddr interface {
	LocalAddr() net.Addr
}
func defaultDial(network, addr string) (net.Conn, error) {
	conn, err := net.DialTimeout(network, addr, defaultConnectionTimeout)
	if err != nil {
		return nil, err
	}
	if err := conn.SetReadDeadline(time.Now().Add(defaultConnectionTimeout)); err != nil {
		return nil, err
	}
	return conn, nil
}
func Dial(url string) (*Connection, error) {
	return DialConfig(url, Config{
		Heartbeat: defaultHeartbeat,
	})
}
func DialTLS(url string, amqps *tls.Config) (*Connection, error) {
	return DialConfig(url, Config{
		Heartbeat:       defaultHeartbeat,
		TLSClientConfig: amqps,
	})
}
func DialConfig(url string, config Config) (*Connection, error) {
	var err error
	var conn net.Conn
	uri, err := ParseURI(url)
	if err != nil {
		return nil, err
	}
	if config.SASL == nil {
		config.SASL = []Authentication{uri.PlainAuth()}
	}
	if config.Vhost == "" {
		config.Vhost = uri.Vhost
	}
	if uri.Scheme == "amqps" && config.TLSClientConfig == nil {
		config.TLSClientConfig = new(tls.Config)
	}
	addr := net.JoinHostPort(uri.Host, strconv.FormatInt(int64(uri.Port), 10))
	dialer := config.Dial
	if dialer == nil {
		dialer = defaultDial
	}
	conn, err = dialer("tcp", addr)
	if err != nil {
		return nil, err
	}
	if config.TLSClientConfig != nil {
		if config.TLSClientConfig.ServerName == "" {
			c := *config.TLSClientConfig
			c.ServerName = uri.Host
			config.TLSClientConfig = &c
		}
		client := tls.Client(conn, config.TLSClientConfig)
		if err := client.Handshake(); err != nil {
			conn.Close()
			return nil, err
		}
		conn = client
	}
	return Open(conn, config)
}
func Open(conn io.ReadWriteCloser, config Config) (*Connection, error) {
	me := &Connection{
		conn:      conn,
		writer:    &writer{bufio.NewWriter(conn)},
		channels:  make(map[uint16]*Channel),
		rpc:       make(chan message),
		sends:     make(chan time.Time),
		errors:    make(chan *Error, 1),
		deadlines: make(chan readDeadliner, 1),
	}
	go me.reader(conn)
	return me, me.open(config)
}
func (me *Connection) LocalAddr() net.Addr {
	if c, ok := me.conn.(localNetAddr); ok {
		return c.LocalAddr()
	}
	return &net.TCPAddr{}
}
func (me *Connection) 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 *Connection) NotifyBlocked(c chan Blocking) chan Blocking {
	me.m.Lock()
	defer me.m.Unlock()
	if me.noNotify {
		close(c)
	} else {
		me.blocks = append(me.blocks, c)
	}
	return c
}
func (me *Connection) Close() error {
	defer me.shutdown(nil)
	return me.call(
		&connectionClose{
			ReplyCode: replySuccess,
			ReplyText: "kthxbai",
		},
		&connectionCloseOk{},
	)
}
func (me *Connection) closeWith(err *Error) error {
	defer me.shutdown(err)
	return me.call(
		&connectionClose{
			ReplyCode: uint16(err.Code),
			ReplyText: err.Reason,
		},
		&connectionCloseOk{},
	)
}
func (me *Connection) send(f frame) error {
	me.sendM.Lock()
	err := me.writer.WriteFrame(f)
	me.sendM.Unlock()
	if err != nil {
		go me.shutdown(&Error{
			Code:   FrameError,
			Reason: err.Error(),
		})
	} else {
		select {
		case me.sends <- time.Now():
		default:
		}
	}
	return err
}
func (me *Connection) shutdown(err *Error) {
	me.destructor.Do(func() {
		if err != nil {
			for _, c := range me.closes {
				c <- err
			}
		}
		for _, ch := range me.channels {
			me.closeChannel(ch, err)
		}
		if err != nil {
			me.errors <- err
		}
		me.conn.Close()
		for _, c := range me.closes {
			close(c)
		}
		for _, c := range me.blocks {
			close(c)
		}
		me.m.Lock()
		me.noNotify = true
		me.m.Unlock()
	})
}