func (nc *Conn) spinUpSocketWatchers() {
	nc.waitForExits()
	nc.wg.Add(2)
	go nc.readLoop()
	go nc.flusher()
	nc.mu.Lock()
	nc.pout = 0
	if nc.Opts.PingInterval > 0 {
		if nc.ptmr == nil {
			nc.ptmr = time.AfterFunc(nc.Opts.PingInterval, nc.processPingTimer)
		} else {
			nc.ptmr.Reset(nc.Opts.PingInterval)
		}
	}
	nc.mu.Unlock()
}
spinUpSocketWatchers will launch the Go routines responsible for
reading and writing to the socket. This will be launched via a
go routine itself to release any locks that may be held.
We also use a WaitGroup to make sure we only start them on a
reconnect when the previous ones have exited.
package nats
import (
	"bufio"
	"bytes"
	"crypto/rand"
	"crypto/tls"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net"
	"net/url"
	"regexp"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	"unsafe"
	mrand "math/rand"
)
const (
	Version              = "1.0.8"
	DefaultURL           = "nats://localhost:4222"
	DefaultPort          = 4222
	DefaultMaxReconnect  = 60
	DefaultReconnectWait = 2 * time.Second
	DefaultTimeout       = 2 * time.Second
	DefaultPingInterval  = 2 * time.Minute
	DefaultMaxPingOut    = 2
	DefaultMaxChanLen    = 65536
	RequestChanLen       = 4
)
const STALE_CONNECTION = "Stale Connection"
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")
	ErrSlowConsumer       = errors.New("nats: Slow Consumer, messages dropped")
	ErrTimeout            = errors.New("nats: Timeout")
	ErrBadTimeout         = errors.New("nats: Timeout Invalid")
	ErrAuthorization      = errors.New("nats: Authorization Failed")
	ErrNoServers          = errors.New("nats: No servers available for connection")
	ErrJsonParse          = errors.New("nats: Connect message, json parse err")
	ErrChanArg            = errors.New("nats: Argument needs to be a channel type")
	ErrStaleConnection    = errors.New("nats: " + STALE_CONNECTION)
)
var DefaultOptions = Options{
	AllowReconnect: true,
	MaxReconnect:   DefaultMaxReconnect,
	ReconnectWait:  DefaultReconnectWait,
	Timeout:        DefaultTimeout,
	PingInterval:   DefaultPingInterval,
	MaxPingsOut:    DefaultMaxPingOut,
	SubChanLen:     DefaultMaxChanLen,
}
type Status int
const (
	DISCONNECTED = Status(iota)
	CONNECTED
	CLOSED
	RECONNECTING
)
type ConnHandler func(*Conn)
type ErrHandler func(*Conn, *Subscription, error)
type Options struct {
	Url            string
	Servers        []string
	NoRandomize    bool
	Name           string
	Verbose        bool
	Pedantic       bool
	Secure         bool
	AllowReconnect bool
	MaxReconnect   int
	ReconnectWait  time.Duration
	Timeout        time.Duration
	ClosedCB       ConnHandler
	DisconnectedCB ConnHandler
	ReconnectedCB  ConnHandler
	AsyncErrorCB   ErrHandler
	PingInterval time.Duration  
	MaxPingsOut  int
	SubChanLen int
}
const (
	scratchSize = 512
	defaultBufSize = 32768
	defaultPendingSize = 1024 * 1024
	flushChanSize = 1024
	srvPoolSize = 4
)
type Conn struct {
	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
	ssid    int64
	subs    map[int64]*Subscription
	mch     chan *Msg
	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
	msgs      uint64
	delivered uint64
	bytes     uint64
	max       uint64
	conn      *Conn
	mcb       MsgHandler
	mch       chan *Msg
	sc        bool
}
type Msg struct {
	Subject string
	Reply   string
	Data    []byte
	Sub     *Subscription
}
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"`
	SslRequired  bool   `json:"ssl_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"`
	Ssl      bool   `json:"ssl_required"`
	Name     string `json:"name"`
}
type MsgHandler func(msg *Msg)
func Connect(url string) (*Conn, error) {
	opts := DefaultOptions
	opts.Url = url
	return opts.Connect()
}
func SecureConnect(url string) (*Conn, error) {
	opts := DefaultOptions
	opts.Url = url
	opts.Secure = true
	return opts.Connect()
}
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 err := nc.setupServerPool(); err != nil {
		return nil, err
	}
	if err := nc.connect(); err != nil {
		return nil, err
	}
	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_
)
func (nc *Conn) serversAvailable() bool {
	for _, s := range nc.srvPool {
		if s != nil {
			return true
		}
	}
	return false
}
func (nc *Conn) debugPool(str string) {
	_, cur := nc.currentServer()
	fmt.Printf("%s\n", str)
	for i, s := range nc.srvPool {
		if s == cur {
			fmt.Printf("\t*%d: %v\n", i+1, s.url)
		} else {
			fmt.Printf("\t%d: %v\n", i+1, s.url)
		}
	}
}
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])
	max_reconnect := nc.Opts.MaxReconnect
	if max_reconnect < 0 || s.reconnects < max_reconnect {
		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
}
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 := mrand.NewSource(time.Now().UnixNano())
	r := mrand.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)
	}
	return nc.pickServer()
}
func (nc *Conn) createConn() error {
	if nc.Opts.Timeout < 0 {
		return ErrBadTimeout
	}
	if _, cur := nc.currentServer(); cur == nil {
		return ErrNoServers
	} else {
		cur.lastAttempt = time.Now()
	}
	nc.conn, nc.err = net.DialTimeout("tcp", nc.url.Host, nc.Opts.Timeout)
	if nc.err != nil {
		return nc.err
	}
	if nc.pending != nil && nc.bw != nil {
		nc.bw.Flush()
	}
	nc.bw = bufio.NewWriterSize(nc.conn, defaultBufSize)
	return nil
}
func (nc *Conn) makeTLSConn() {
	nc.conn = tls.Client(nc.conn, &tls.Config{InsecureSkipVerify: true})
	nc.bw = bufio.NewWriterSize(nc.conn, defaultBufSize)
}
func (nc *Conn) waitForExits() {
	nc.fch <- true
	nc.wg.Wait()
}