func (nc *Conn) SetErrorHandler(cb ErrHandler) {
	if nc == nil {
		return
	}
	nc.mu.Lock()
	defer nc.mu.Unlock()
	nc.Opts.AsyncErrorCB = cb
}
SetErrorHandler will set the async error handler.
package nats
import (
	"bufio"
	"bytes"
	"crypto/tls"
	"crypto/x509"
	"encoding/base64"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"io/ioutil"
	"math/rand"
	"net"
	"net/url"
	"os"
	"path/filepath"
	"runtime"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	"github.com/nats-io/jwt"
	"github.com/nats-io/nats.go/util"
	"github.com/nats-io/nkeys"
	"github.com/nats-io/nuid"
)
const (
	Version                 = "1.9.1"
	DefaultURL              = "nats://127.0.0.1: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
	DefaultDrainTimeout     = 30 * time.Second
	LangString              = "go"
)
const (
	STALE_CONNECTION = "stale connection"
	PERMISSIONS_ERR = "permissions violation"
	AUTHORIZATION_ERR = "authorization violation"
	AUTHENTICATION_EXPIRED_ERR = "user authentication expired"
)
var (
	ErrConnectionClosed       = errors.New("nats: connection closed")
	ErrConnectionDraining     = errors.New("nats: connection draining")
	ErrDrainTimeout           = errors.New("nats: draining connection timed out")
	ErrConnectionReconnecting = errors.New("nats: connection reconnecting")
	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")
	ErrBadQueueName           = errors.New("nats: invalid queue name")
	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")
	ErrAuthExpired            = errors.New("nats: authentication expired")
	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")
	ErrInvalidContext         = errors.New("nats: invalid context")
	ErrNoDeadlineContext      = errors.New("nats: context requires a deadline")
	ErrNoEchoNotSupported     = errors.New("nats: no echo option not supported by this server")
	ErrClientIDNotSupported   = errors.New("nats: client ID not supported by this server")
	ErrUserButNoSigCB         = errors.New("nats: user callback defined without a signature handler")
	ErrNkeyButNoSigCB         = errors.New("nats: nkey defined without a signature handler")
	ErrNoUserCB               = errors.New("nats: user callback not defined")
	ErrNkeyAndUser            = errors.New("nats: user callback and nkey defined")
	ErrNkeysNotSupported      = errors.New("nats: nkeys not supported by the server")
	ErrStaleConnection        = errors.New("nats: " + STALE_CONNECTION)
	ErrTokenAlreadySet        = errors.New("nats: token and token handler both set")
	ErrMsgNotBound            = errors.New("nats: message is not bound to subscription/connection")
	ErrMsgNoReply             = errors.New("nats: message does not have a reply")
)
func init() {
	rand.Seed(time.Now().UnixNano())
}
func GetDefaultOptions() Options {
	return Options{
		AllowReconnect:   true,
		MaxReconnect:     DefaultMaxReconnect,
		ReconnectWait:    DefaultReconnectWait,
		Timeout:          DefaultTimeout,
		PingInterval:     DefaultPingInterval,
		MaxPingsOut:      DefaultMaxPingOut,
		SubChanLen:       DefaultMaxChanLen,
		ReconnectBufSize: DefaultReconnectBufSize,
		DrainTimeout:     DefaultDrainTimeout,
	}
}
var DefaultOptions = GetDefaultOptions()
type Status int
const (
	DISCONNECTED = Status(iota)
	CONNECTED
	CLOSED
	RECONNECTING
	CONNECTING
	DRAINING_SUBS
	DRAINING_PUBS
)
type ConnHandler func(*Conn)
type ConnErrHandler func(*Conn, error)
type ErrHandler func(*Conn, *Subscription, error)
type UserJWTHandler func() (string, error)
type SignatureHandler func([]byte) ([]byte, error)
type AuthTokenHandler func() string
type asyncCB struct {
	f    func()
	next *asyncCB
}
type asyncCallbacksHandler struct {
	mu   sync.Mutex
	cond *sync.Cond
	head *asyncCB
	tail *asyncCB
}
type Option func(*Options) error
type CustomDialer interface {
	Dial(network, address string) (net.Conn, error)
}
type Options struct {
	Url string
	Servers []string
	NoRandomize bool
	NoEcho bool
	Name string
	Verbose bool
	Pedantic bool
	Secure bool
	TLSConfig *tls.Config
	AllowReconnect bool
	MaxReconnect int
	ReconnectWait time.Duration
	Timeout time.Duration
	DrainTimeout time.Duration
	FlusherTimeout time.Duration
	PingInterval time.Duration
	MaxPingsOut int
	ClosedCB ConnHandler
	DisconnectedCB ConnHandler
	DisconnectedErrCB ConnErrHandler
	ReconnectedCB ConnHandler
	DiscoveredServersCB ConnHandler
	AsyncErrorCB ErrHandler
	ReconnectBufSize int
	SubChanLen int
	UserJWT UserJWTHandler
	Nkey string
	SignatureCB SignatureHandler
	User string
	Password string
	Token string
	TokenHandler AuthTokenHandler
	Dialer *net.Dialer
	CustomDialer CustomDialer
	UseOldRequestStyle bool
	NoCallbacksAfterClientClose bool
}
const (
	scratchSize = 512
	defaultBufSize = 32768
	flushChanSize = 1
	srvPoolSize = 4
	nuidSize = 22
	defaultPortString = "4222"
)
type Conn struct {
	Statistics
	mu sync.RWMutex
	Opts    Options
	wg      sync.WaitGroup
	srvPool []*srv
	current *srv
	urls    map[string]struct{}  
	conn    net.Conn
	bw      *bufio.Writer
	pending *bytes.Buffer
	fch     chan struct{}
	info    serverInfo
	ssid    int64
	subsMu  sync.RWMutex
	subs    map[int64]*Subscription
	ach     *asyncCallbacksHandler
	pongs   []chan struct{}
	scratch [scratchSize]byte
	status  Status
	initc   bool  
	err     error
	ps      *parseState
	ptmr    *time.Timer
	pout    int
	ar      bool  
	respSub   string                
	respScanf string                
	respMux   *Subscription         
	respMap   map[string]chan *Msg  
	respSetup sync.Once             
	respRand  *rand.Rand            
}
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
	barrier *barrierInfo
}
type barrierInfo struct {
	refs int64
	f    func()
}
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
	lastErr     error
	isImplicit  bool
	tlsName     string
}
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"`
	ConnectURLs  []string `json:"connect_urls,omitempty"`
	Proto        int      `json:"proto,omitempty"`
	CID          uint64   `json:"client_id,omitempty"`
	Nonce        string   `json:"nonce,omitempty"`
}
const (
	  _ = iota
	clientProtoInfo
)
type connectInfo struct {
	Verbose   bool   `json:"verbose"`
	Pedantic  bool   `json:"pedantic"`
	UserJWT   string `json:"jwt,omitempty"`
	Nkey      string `json:"nkey,omitempty"`
	Signature string `json:"sig,omitempty"`
	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"`
	Protocol  int    `json:"protocol"`
	Echo      bool   `json:"echo"`
}
type MsgHandler func(msg *Msg)
func Connect(url string, options ...Option) (*Conn, error) {
	opts := GetDefaultOptions()
	opts.Servers = processUrlString(url)
	for _, opt := range options {
		if opt != nil {
			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(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 NoEcho() Option {
	return func(o *Options) error {
		o.NoEcho = 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 PingInterval(t time.Duration) Option {
	return func(o *Options) error {
		o.PingInterval = t
		return nil
	}
}
func MaxPingsOutstanding(max int) Option {
	return func(o *Options) error {
		o.MaxPingsOut = max
		return nil
	}
}
func ReconnectBufSize(size int) Option {
	return func(o *Options) error {
		o.ReconnectBufSize = size
		return nil
	}
}
func Timeout(t time.Duration) Option {
	return func(o *Options) error {
		o.Timeout = t
		return nil
	}
}
func FlusherTimeout(t time.Duration) Option {
	return func(o *Options) error {
		o.FlusherTimeout = t
		return nil
	}
}
func DrainTimeout(t time.Duration) Option {
	return func(o *Options) error {
		o.DrainTimeout = t
		return nil
	}
}
func DisconnectErrHandler(cb ConnErrHandler) Option {
	return func(o *Options) error {
		o.DisconnectedErrCB = cb
		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 DiscoveredServersHandler(cb ConnHandler) Option {
	return func(o *Options) error {
		o.DiscoveredServersCB = cb
		return nil
	}
}
func ErrorHandler(cb ErrHandler) Option {
	return func(o *Options) error {
		o.AsyncErrorCB = cb
		return nil
	}
}
func UserInfo(user, password string) Option {
	return func(o *Options) error {
		o.User = user
		o.Password = password
		return nil
	}
}
func Token(token string) Option {
	return func(o *Options) error {
		if o.TokenHandler != nil {
			return ErrTokenAlreadySet
		}
		o.Token = token
		return nil
	}
}
func TokenHandler(cb AuthTokenHandler) Option {
	return func(o *Options) error {
		if o.Token != "" {
			return ErrTokenAlreadySet
		}
		o.TokenHandler = cb
		return nil
	}
}
func UserCredentials(userOrChainedFile string, seedFiles ...string) Option {
	userCB := func() (string, error) {
		return userFromFile(userOrChainedFile)
	}
	var keyFile string
	if len(seedFiles) > 0 {
		keyFile = seedFiles[0]
	} else {
		keyFile = userOrChainedFile
	}
	sigCB := func(nonce []byte) ([]byte, error) {
		return sigHandler(nonce, keyFile)
	}
	return UserJWT(userCB, sigCB)
}
func UserJWT(userCB UserJWTHandler, sigCB SignatureHandler) Option {
	return func(o *Options) error {
		if userCB == nil {
			return ErrNoUserCB
		}
		if sigCB == nil {
			return ErrUserButNoSigCB
		}
		o.UserJWT = userCB
		o.SignatureCB = sigCB
		return nil
	}
}
func Nkey(pubKey string, sigCB SignatureHandler) Option {
	return func(o *Options) error {
		o.Nkey = pubKey
		o.SignatureCB = sigCB
		if pubKey != "" && sigCB == nil {
			return ErrNkeyButNoSigCB
		}
		return nil
	}
}
func SyncQueueLen(max int) Option {
	return func(o *Options) error {
		o.SubChanLen = max
		return nil
	}
}
func Dialer(dialer *net.Dialer) Option {
	return func(o *Options) error {
		o.Dialer = dialer
		return nil
	}
}
func SetCustomDialer(dialer CustomDialer) Option {
	return func(o *Options) error {
		o.CustomDialer = dialer
		return nil
	}
}
func UseOldRequestStyle() Option {
	return func(o *Options) error {
		o.UseOldRequestStyle = true
		return nil
	}
}
func NoCallbacksAfterClientClose() Option {
	return func(o *Options) error {
		o.NoCallbacksAfterClientClose = true
		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) SetDisconnectErrHandler(dcb ConnErrHandler) {
	if nc == nil {
		return
	}
	nc.mu.Lock()
	defer nc.mu.Unlock()
	nc.Opts.DisconnectedErrCB = 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) SetDiscoveredServersHandler(dscb ConnHandler) {
	if nc == nil {
		return
	}
	nc.mu.Lock()
	defer nc.mu.Unlock()
	nc.Opts.DiscoveredServersCB = dscb
}
func (nc *Conn) SetClosedHandler(cb ConnHandler) {
	if nc == nil {
		return
	}
	nc.mu.Lock()
	defer nc.mu.Unlock()
	nc.Opts.ClosedCB = cb
}