func (nc *Conn) waitForExits() {
	select {
	case nc.fch <- true:
	default:
	}
	nc.wg.Wait()
}
waitForExits will wait for all socket watcher Go routines to
be shutdown before proceeding.
package nats
import (
	"bufio"
	"bytes"
	"crypto/tls"
	"crypto/x509"
	"encoding/json"
	"errors"
	"fmt"
	"io/ioutil"
	"math/rand"
	"net"
	"net/url"
	"regexp"
	"runtime"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	"github.com/nats-io/nuid"
)
const (
	Version                 = "1.2.2"
	DefaultURL              = "nats://localhost:4222"
	DefaultPort             = 4222
	DefaultMaxReconnect     = 60
	DefaultReconnectWait    = 2 * time.Second
	DefaultTimeout          = 2 * time.Second
	DefaultPingInterval     = 2 * time.Minute
	DefaultMaxPingOut       = 2
	DefaultMaxChanLen       = 8192             
	DefaultReconnectBufSize = 8 * 1024 * 1024  
	RequestChanLen          = 8
	LangString              = "go"
)
const STALE_CONNECTION = "stale connection"
const PERMISSIONS_ERR = "permissions violation"
var (
	ErrConnectionClosed     = errors.New("nats: connection closed")
	ErrSecureConnRequired   = errors.New("nats: secure connection required")
	ErrSecureConnWanted     = errors.New("nats: secure connection not available")
	ErrBadSubscription      = errors.New("nats: invalid subscription")
	ErrTypeSubscription     = errors.New("nats: invalid subscription type")
	ErrBadSubject           = errors.New("nats: invalid subject")
	ErrSlowConsumer         = errors.New("nats: slow consumer, messages dropped")
	ErrTimeout              = errors.New("nats: timeout")
	ErrBadTimeout           = errors.New("nats: timeout invalid")
	ErrAuthorization        = errors.New("nats: authorization violation")
	ErrNoServers            = errors.New("nats: no servers available for connection")
	ErrJsonParse            = errors.New("nats: connect message, json parse error")
	ErrChanArg              = errors.New("nats: argument needs to be a channel type")
	ErrMaxPayload           = errors.New("nats: maximum payload exceeded")
	ErrMaxMessages          = errors.New("nats: maximum messages delivered")
	ErrSyncSubRequired      = errors.New("nats: illegal call on an async subscription")
	ErrMultipleTLSConfigs   = errors.New("nats: multiple tls.Configs not allowed")
	ErrNoInfoReceived       = errors.New("nats: protocol exception, INFO not received")
	ErrReconnectBufExceeded = errors.New("nats: outbound buffer limit exceeded")
	ErrInvalidConnection    = errors.New("nats: invalid connection")
	ErrInvalidMsg           = errors.New("nats: invalid message or message nil")
	ErrInvalidArg           = errors.New("nats: invalid argument")
	ErrStaleConnection      = errors.New("nats: " + STALE_CONNECTION)
)
var DefaultOptions = Options{
	AllowReconnect:   true,
	MaxReconnect:     DefaultMaxReconnect,
	ReconnectWait:    DefaultReconnectWait,
	Timeout:          DefaultTimeout,
	PingInterval:     DefaultPingInterval,
	MaxPingsOut:      DefaultMaxPingOut,
	SubChanLen:       DefaultMaxChanLen,
	ReconnectBufSize: DefaultReconnectBufSize,
}
type Status int
const (
	DISCONNECTED = Status(iota)
	CONNECTED
	CLOSED
	RECONNECTING
	CONNECTING
)
type ConnHandler func(*Conn)
type ErrHandler func(*Conn, *Subscription, error)
type asyncCB func()
type Option func(*Options) error
type Options struct {
	Url            string
	Servers        []string
	NoRandomize    bool
	Name           string
	Verbose        bool
	Pedantic       bool
	Secure         bool
	TLSConfig      *tls.Config
	AllowReconnect bool
	MaxReconnect   int
	ReconnectWait  time.Duration
	Timeout        time.Duration
	PingInterval   time.Duration  
	MaxPingsOut    int
	ClosedCB       ConnHandler
	DisconnectedCB ConnHandler
	ReconnectedCB  ConnHandler
	AsyncErrorCB   ErrHandler
	ReconnectBufSize int
	SubChanLen int
}
const (
	scratchSize = 512
	defaultBufSize = 32768
	flushChanSize = 1024
	srvPoolSize = 4
	asyncCBChanSize = 32
)
type Conn struct {
	ssid int64
	Statistics
	mu      sync.Mutex
	Opts    Options
	wg      sync.WaitGroup
	url     *url.URL
	conn    net.Conn
	srvPool []*srv
	bw      *bufio.Writer
	pending *bytes.Buffer
	fch     chan bool
	info    serverInfo
	subs    map[int64]*Subscription
	mch     chan *Msg
	ach     chan asyncCB
	pongs   []chan bool
	scratch [scratchSize]byte
	status  Status
	err     error
	ps      *parseState
	ptmr    *time.Timer
	pout    int
}
type Subscription struct {
	mu  sync.Mutex
	sid int64
	Subject string
	Queue string
	delivered  uint64
	max        uint64
	conn       *Conn
	mcb        MsgHandler
	mch        chan *Msg
	closed     bool
	sc         bool
	connClosed bool
	typ SubscriptionType
	pHead *Msg
	pTail *Msg
	pCond *sync.Cond
	pMsgs       int
	pBytes      int
	pMsgsMax    int
	pBytesMax   int
	pMsgsLimit  int
	pBytesLimit int
	dropped     int
}
type Msg struct {
	Subject string
	Reply   string
	Data    []byte
	Sub     *Subscription
	next    *Msg
}
type Statistics struct {
	InMsgs     uint64
	OutMsgs    uint64
	InBytes    uint64
	OutBytes   uint64
	Reconnects uint64
}
type srv struct {
	url         *url.URL
	didConnect  bool
	reconnects  int
	lastAttempt time.Time
}
type serverInfo struct {
	Id           string `json:"server_id"`
	Host         string `json:"host"`
	Port         uint   `json:"port"`
	Version      string `json:"version"`
	AuthRequired bool   `json:"auth_required"`
	TLSRequired  bool   `json:"tls_required"`
	MaxPayload   int64  `json:"max_payload"`
}
type connectInfo struct {
	Verbose  bool   `json:"verbose"`
	Pedantic bool   `json:"pedantic"`
	User     string `json:"user,omitempty"`
	Pass     string `json:"pass,omitempty"`
	Token    string `json:"auth_token,omitempty"`
	TLS      bool   `json:"tls_required"`
	Name     string `json:"name"`
	Lang     string `json:"lang"`
	Version  string `json:"version"`
}
type MsgHandler func(msg *Msg)
func Connect(url string, options ...Option) (*Conn, error) {
	opts := DefaultOptions
	opts.Servers = processUrlString(url)
	for _, opt := range options {
		if err := opt(&opts); err != nil {
			return nil, err
		}
	}
	return opts.Connect()
}
func Name(name string) Option {
	return func(o *Options) error {
		o.Name = name
		return nil
	}
}
func Secure(tls ...*tls.Config) Option {
	return func(o *Options) error {
		o.Secure = true
		if len(tls) > 1 {
			return ErrMultipleTLSConfigs
		}
		if len(tls) == 1 {
			o.TLSConfig = tls[0]
		}
		return nil
	}
}
func RootCAs(file ...string) Option {
	return func(o *Options) error {
		pool := x509.NewCertPool()
		for _, f := range file {
			rootPEM, err := ioutil.ReadFile(f)
			if err != nil || rootPEM == nil {
				return fmt.Errorf("nats: error loading or parsing rootCA file: %v", err)
			}
			ok := pool.AppendCertsFromPEM([]byte(rootPEM))
			if !ok {
				return fmt.Errorf("nats: failed to parse root certificate from %q", f)
			}
		}
		if o.TLSConfig == nil {
			o.TLSConfig = &tls.Config{MinVersion: tls.VersionTLS12}
		}
		o.TLSConfig.RootCAs = pool
		o.Secure = true
		return nil
	}
}
func ClientCert(certFile, keyFile string) Option {
	return func(o *Options) error {
		cert, err := tls.LoadX509KeyPair(certFile, keyFile)
		if err != nil {
			return fmt.Errorf("nats: error loading client certificate: %v", err)
		}
		cert.Leaf, err = x509.ParseCertificate(cert.Certificate[0])
		if err != nil {
			return fmt.Errorf("nats: error parsing client certificate: %v", err)
		}
		if o.TLSConfig == nil {
			o.TLSConfig = &tls.Config{MinVersion: tls.VersionTLS12}
		}
		o.TLSConfig.Certificates = []tls.Certificate{cert}
		o.Secure = true
		return nil
	}
}
func NoReconnect() Option {
	return func(o *Options) error {
		o.AllowReconnect = false
		return nil
	}
}
func DontRandomize() Option {
	return func(o *Options) error {
		o.NoRandomize = true
		return nil
	}
}
func ReconnectWait(t time.Duration) Option {
	return func(o *Options) error {
		o.ReconnectWait = t
		return nil
	}
}
func MaxReconnects(max int) Option {
	return func(o *Options) error {
		o.MaxReconnect = max
		return nil
	}
}
func Timeout(t time.Duration) Option {
	return func(o *Options) error {
		o.Timeout = t
		return nil
	}
}
func DisconnectHandler(cb ConnHandler) Option {
	return func(o *Options) error {
		o.DisconnectedCB = cb
		return nil
	}
}
func ReconnectHandler(cb ConnHandler) Option {
	return func(o *Options) error {
		o.ReconnectedCB = cb
		return nil
	}
}
func ClosedHandler(cb ConnHandler) Option {
	return func(o *Options) error {
		o.ClosedCB = cb
		return nil
	}
}
func ErrorHandler(cb ErrHandler) Option {
	return func(o *Options) error {
		o.AsyncErrorCB = cb
		return nil
	}
}
func (nc *Conn) SetDisconnectHandler(dcb ConnHandler) {
	if nc == nil {
		return
	}
	nc.mu.Lock()
	defer nc.mu.Unlock()
	nc.Opts.DisconnectedCB = dcb
}
func (nc *Conn) SetReconnectHandler(rcb ConnHandler) {
	if nc == nil {
		return
	}
	nc.mu.Lock()
	defer nc.mu.Unlock()
	nc.Opts.ReconnectedCB = rcb
}
func (nc *Conn) SetClosedHandler(cb ConnHandler) {
	if nc == nil {
		return
	}
	nc.mu.Lock()
	defer nc.mu.Unlock()
	nc.Opts.ClosedCB = cb
}
func (nc *Conn) SetErrorHandler(cb ErrHandler) {
	if nc == nil {
		return
	}
	nc.mu.Lock()
	defer nc.mu.Unlock()
	nc.Opts.AsyncErrorCB = cb
}
func processUrlString(url string) []string {
	urls := strings.Split(url, ",")
	for i, s := range urls {
		urls[i] = strings.TrimSpace(s)
	}
	return urls
}
func (o Options) Connect() (*Conn, error) {
	nc := &Conn{Opts: o}
	if nc.Opts.MaxPingsOut == 0 {
		nc.Opts.MaxPingsOut = DefaultMaxPingOut
	}
	if nc.Opts.SubChanLen == 0 {
		nc.Opts.SubChanLen = DefaultMaxChanLen
	}
	if nc.Opts.ReconnectBufSize == 0 {
		nc.Opts.ReconnectBufSize = DefaultReconnectBufSize
	}
	if nc.Opts.Timeout == 0 {
		nc.Opts.Timeout = DefaultTimeout
	}
	if err := nc.setupServerPool(); err != nil {
		return nil, err
	}
	nc.ach = make(chan asyncCB, asyncCBChanSize)
	if err := nc.connect(); err != nil {
		return nil, err
	}
	go nc.asyncDispatch()
	return nc, nil
}
const (
	_CRLF_  = "\r\n"
	_EMPTY_ = ""
	_SPC_   = " "
	_PUB_P_ = "PUB "
)
const (
	_OK_OP_   = "+OK"
	_ERR_OP_  = "-ERR"
	_MSG_OP_  = "MSG"
	_PING_OP_ = "PING"
	_PONG_OP_ = "PONG"
	_INFO_OP_ = "INFO"
)
const (
	conProto   = "CONNECT %s" + _CRLF_
	pingProto  = "PING" + _CRLF_
	pongProto  = "PONG" + _CRLF_
	pubProto   = "PUB %s %s %d" + _CRLF_
	subProto   = "SUB %s %s %d" + _CRLF_
	unsubProto = "UNSUB %d %s" + _CRLF_
	okProto    = _OK_OP_ + _CRLF_
)
func (nc *Conn) currentServer() (int, *srv) {
	for i, s := range nc.srvPool {
		if s == nil {
			continue
		}
		if s.url == nc.url {
			return i, s
		}
	}
	return -1, nil
}
func (nc *Conn) selectNextServer() (*srv, error) {
	i, s := nc.currentServer()
	if i < 0 {
		return nil, ErrNoServers
	}
	sp := nc.srvPool
	num := len(sp)
	copy(sp[i:num-1], sp[i+1:num])
	maxReconnect := nc.Opts.MaxReconnect
	if maxReconnect < 0 || s.reconnects < maxReconnect {
		nc.srvPool[num-1] = s
	} else {
		nc.srvPool = sp[0 : num-1]
	}
	if len(nc.srvPool) <= 0 {
		nc.url = nil
		return nil, ErrNoServers
	}
	nc.url = nc.srvPool[0].url
	return nc.srvPool[0], nil
}
func (nc *Conn) pickServer() error {
	nc.url = nil
	if len(nc.srvPool) <= 0 {
		return ErrNoServers
	}
	for _, s := range nc.srvPool {
		if s != nil {
			nc.url = s.url
			return nil
		}
	}
	return ErrNoServers
}
const tlsScheme = "tls"
func (nc *Conn) setupServerPool() error {
	nc.srvPool = make([]*srv, 0, srvPoolSize)
	if nc.Opts.Url != _EMPTY_ {
		u, err := url.Parse(nc.Opts.Url)
		if err != nil {
			return err
		}
		s := &srv{url: u}
		nc.srvPool = append(nc.srvPool, s)
	}
	var srvrs []string
	source := rand.NewSource(time.Now().UnixNano())
	r := rand.New(source)
	if nc.Opts.NoRandomize {
		srvrs = nc.Opts.Servers
	} else {
		in := r.Perm(len(nc.Opts.Servers))
		for _, i := range in {
			srvrs = append(srvrs, nc.Opts.Servers[i])
		}
	}
	for _, urlString := range srvrs {
		u, err := url.Parse(urlString)
		if err != nil {
			return err
		}
		s := &srv{url: u}
		nc.srvPool = append(nc.srvPool, s)
	}
	if len(nc.srvPool) <= 0 {
		u, err := url.Parse(DefaultURL)
		if err != nil {
			return err
		}
		s := &srv{url: u}
		nc.srvPool = append(nc.srvPool, s)
	}
	for _, srv := range nc.srvPool {
		if srv.url.Scheme == tlsScheme {
			nc.Opts.Secure = true
			if nc.Opts.TLSConfig == nil {
				nc.Opts.TLSConfig = &tls.Config{MinVersion: tls.VersionTLS12}
			}
		}
	}
	return nc.pickServer()
}
func (nc *Conn) createConn() (err error) {
	if nc.Opts.Timeout < 0 {
		return ErrBadTimeout
	}
	if _, cur := nc.currentServer(); cur == nil {
		return ErrNoServers
	} else {
		cur.lastAttempt = time.Now()
	}
	nc.conn, err = net.DialTimeout("tcp", nc.url.Host, nc.Opts.Timeout)
	if err != nil {
		return err
	}
	if nc.pending != nil && nc.bw != nil {
		nc.bw.Flush()
	}
	nc.bw = bufio.NewWriterSize(nc.conn, defaultBufSize)
	return nil
}
func (nc *Conn) makeTLSConn() {
	if nc.Opts.TLSConfig != nil {
		tlsCopy := *nc.Opts.TLSConfig
		if tlsCopy.ServerName == _EMPTY_ {
			h, _, _ := net.SplitHostPort(nc.url.Host)
			tlsCopy.ServerName = h
		}
		nc.conn = tls.Client(nc.conn, &tlsCopy)
	} else {
		nc.conn = tls.Client(nc.conn, &tls.Config{InsecureSkipVerify: true})
	}
	conn := nc.conn.(*tls.Conn)
	conn.Handshake()
	nc.bw = bufio.NewWriterSize(nc.conn, defaultBufSize)
}