func (sc *conn) removeAck(guid string) *ack {
	var t *time.Timer
	sc.Lock()
	a := sc.pubAckMap[guid]
	if a != nil {
		t = a.t
		delete(sc.pubAckMap, guid)
	}
	pac := sc.pubAckChan
	sc.Unlock()
	if t != nil {
		t.Stop()
	}
	if a != nil && len(pac) > 0 {
		<-pac
	}
	return a
}
removeAck removes the ack from the pubAckMap and cancels any state, e.g. timers
package stan
import (
	"errors"
	"fmt"
	"sync"
	"time"
	"github.com/nats-io/nats.go"
	"github.com/nats-io/nuid"
	"github.com/nats-io/stan.go/pb"
)
const Version = "0.10.0"
const (
	DefaultNatsURL = "nats://127.0.0.1:4222"
	DefaultConnectWait = 2 * time.Second
	DefaultDiscoverPrefix = "_STAN.discover"
	DefaultACKPrefix = "_STAN.acks"
	DefaultMaxPubAcksInflight = 16384
	DefaultPingInterval = 5
	DefaultPingMaxOut = 88
)
type Conn interface {
	Publish(subject string, data []byte) error
	PublishAsync(subject string, data []byte, ah AckHandler) (string, error)
	Subscribe(subject string, cb MsgHandler, opts ...SubscriptionOption) (Subscription, error)
	QueueSubscribe(subject, qgroup string, cb MsgHandler, opts ...SubscriptionOption) (Subscription, error)
	Close() error
	NatsConn() *nats.Conn
}
const (
	protocolOne = int32(1)
)
var (
	ErrConnectReqTimeout = errors.New("stan: connect request timeout (possibly wrong cluster ID?)")
	ErrCloseReqTimeout   = errors.New("stan: close request timeout")
	ErrSubReqTimeout     = errors.New("stan: subscribe request timeout")
	ErrUnsubReqTimeout   = errors.New("stan: unsubscribe request timeout")
	ErrConnectionClosed  = errors.New("stan: connection closed")
	ErrTimeout           = errors.New("stan: publish ack timeout")
	ErrBadAck            = errors.New("stan: malformed ack")
	ErrBadSubscription   = errors.New("stan: invalid subscription")
	ErrBadConnection     = errors.New("stan: invalid connection")
	ErrManualAck         = errors.New("stan: cannot manually ack in auto-ack mode")
	ErrNilMsg            = errors.New("stan: nil message")
	ErrNoServerSupport   = errors.New("stan: not supported by server")
	ErrMaxPings          = errors.New("stan: connection lost due to PING failure")
)
var testAllowMillisecInPings = false
type AckHandler func(string, error)
type ConnectionLostHandler func(Conn, error)
type Options struct {
	NatsURL string
	NatsConn *nats.Conn
	NatsOptions []nats.Option
	ConnectTimeout time.Duration
	AckTimeout time.Duration
	DiscoverPrefix string
	MaxPubAcksInflight int
	PingIterval int
	PingInterval int
	PingMaxOut int
	ConnectionLostCB ConnectionLostHandler
}
func GetDefaultOptions() Options {
	return getDefaultOptions()
}
func getDefaultOptions() Options {
	return Options{
		NatsURL:            DefaultNatsURL,
		ConnectTimeout:     DefaultConnectWait,
		AckTimeout:         DefaultAckWait,
		DiscoverPrefix:     DefaultDiscoverPrefix,
		MaxPubAcksInflight: DefaultMaxPubAcksInflight,
		PingInterval:       DefaultPingInterval,
		PingMaxOut:         DefaultPingMaxOut,
	}
}
var DefaultOptions = getDefaultOptions()
type Option func(*Options) error
func NatsURL(u string) Option {
	return func(o *Options) error {
		o.NatsURL = u
		return nil
	}
}
func ConnectWait(t time.Duration) Option {
	return func(o *Options) error {
		o.ConnectTimeout = t
		return nil
	}
}
func PubAckWait(t time.Duration) Option {
	return func(o *Options) error {
		o.AckTimeout = t
		return nil
	}
}
func MaxPubAcksInflight(max int) Option {
	return func(o *Options) error {
		o.MaxPubAcksInflight = max
		return nil
	}
}
func NatsConn(nc *nats.Conn) Option {
	return func(o *Options) error {
		o.NatsConn = nc
		return nil
	}
}
func NatsOptions(opts ...nats.Option) Option {
	return func(o *Options) error {
		o.NatsOptions = append([]nats.Option(nil), opts...)
		return nil
	}
}
func Pings(interval, maxOut int) Option {
	return func(o *Options) error {
		if !testAllowMillisecInPings {
			if interval < 1 || maxOut < 2 {
				return fmt.Errorf("invalid ping values: interval=%v (min>0) maxOut=%v (min=2)", interval, maxOut)
			}
		}
		o.PingInterval = interval
		o.PingMaxOut = maxOut
		return nil
	}
}
func SetConnectionLostHandler(handler ConnectionLostHandler) Option {
	return func(o *Options) error {
		o.ConnectionLostCB = handler
		return nil
	}
}
type conn struct {
	sync.RWMutex
	clientID         string
	connID           []byte  
	pubPrefix        string  
	subRequests      string  
	unsubRequests    string  
	subCloseRequests string  
	closeRequests    string  
	ackSubject       string  
	ackSubscription  *nats.Subscription
	hbSubscription   *nats.Subscription
	subMap           map[string]*subscription
	pubAckMap        map[string]*ack
	pubAckChan       chan (struct{})
	pubAckCloseChan  chan (struct{})
	opts             Options
	nc               *nats.Conn
	ncOwned          bool        
	pubNUID          *nuid.NUID  
	connLostCB       ConnectionLostHandler
	closed           bool
	ping             pingInfo
}
type pingInfo struct {
	mu       sync.Mutex
	sub      *nats.Subscription
	timer    *time.Timer
	proto    []byte
	requests string
	inbox    string
	interval time.Duration
	maxOut   int
	out      int
}
type ack struct {
	t  *time.Timer
	ah AckHandler
	ch chan error
}
func Connect(stanClusterID, clientID string, options ...Option) (Conn, error) {
	c := conn{
		clientID:        clientID,
		opts:            getDefaultOptions(),
		connID:          []byte(nuid.Next()),
		pubNUID:         nuid.New(),
		pubAckMap:       make(map[string]*ack),
		pubAckCloseChan: make(chan struct{}),
		subMap:          make(map[string]*subscription),
	}
	for _, opt := range options {
		if err := opt(&c.opts); err != nil {
			return nil, err
		}
	}
	c.nc = c.opts.NatsConn
	if c.nc == nil {
		nopts := c.opts.NatsOptions
		nopts = append(nopts, nats.MaxReconnects(-1), nats.ReconnectBufSize(-1))
		var do nats.Options
		for _, o := range nopts {
			o(&do)
		}
		if do.Name == "" {
			nopts = append(nopts, nats.Name(clientID))
		}
		nc, err := nats.Connect(c.opts.NatsURL, nopts...)
		if err != nil {
			return nil, err
		}
		c.nc = nc
		c.ncOwned = true
	} else if !c.nc.IsConnected() {
		return nil, ErrBadConnection
	}
	hbInbox := nats.NewInbox()
	var err error
	if c.hbSubscription, err = c.nc.Subscribe(hbInbox, c.processHeartBeat); err != nil {
		c.failConnect(err)
		return nil, err
	}
	p := &c.ping
	if p.sub, err = c.nc.Subscribe(nats.NewInbox(), c.processPingResponse); err != nil {
		c.failConnect(err)
		return nil, err
	}
	discoverSubject := c.opts.DiscoverPrefix + "." + stanClusterID
	req := &pb.ConnectRequest{
		ClientID:       clientID,
		HeartbeatInbox: hbInbox,
		ConnID:         c.connID,
		Protocol:       protocolOne,
		PingInterval:   int32(c.opts.PingInterval),
		PingMaxOut:     int32(c.opts.PingMaxOut),
	}
	b, _ := req.Marshal()
	reply, err := c.nc.Request(discoverSubject, b, c.opts.ConnectTimeout)
	if err != nil {
		c.failConnect(err)
		if err == nats.ErrTimeout || err == nats.ErrNoResponders {
			return nil, ErrConnectReqTimeout
		}
		return nil, err
	}
	cr := &pb.ConnectResponse{}
	err = cr.Unmarshal(reply.Data)
	if err != nil {
		c.failConnect(err)
		return nil, err
	}
	if cr.Error != "" {
		c.failConnect(err)
		return nil, errors.New(cr.Error)
	}
	c.pubPrefix = cr.PubPrefix
	c.subRequests = cr.SubRequests
	c.unsubRequests = cr.UnsubRequests
	c.subCloseRequests = cr.SubCloseRequests
	c.closeRequests = cr.CloseRequests
	c.ackSubject = DefaultACKPrefix + "." + nuid.Next()
	if c.ackSubscription, err = c.nc.Subscribe(c.ackSubject, c.processAck); err != nil {
		c.Close()
		return nil, err
	}
	c.ackSubscription.SetPendingLimits(-1, -1)
	c.pubAckChan = make(chan struct{}, c.opts.MaxPubAcksInflight)
	c.connLostCB = c.opts.ConnectionLostCB
	unsubPingSub := true
	if cr.Protocol >= protocolOne {
		if cr.PingInterval != 0 {
			unsubPingSub = false
			p.requests = cr.PingRequests
			p.inbox = p.sub.Subject
			if testAllowMillisecInPings && cr.PingInterval < 0 {
				p.interval = time.Duration(cr.PingInterval*-1) * time.Millisecond
			} else {
				p.interval = time.Duration(cr.PingInterval) * time.Second
			}
			p.maxOut = int(cr.PingMaxOut)
			p.proto, _ = (&pb.Ping{ConnID: c.connID}).Marshal()
			p.mu.Lock()
			p.timer = time.AfterFunc(p.interval, c.pingServer)
			p.mu.Unlock()
		}
	}
	if unsubPingSub {
		p.sub.Unsubscribe()
		p.sub = nil
	}
	return &c, nil
}
func (sc *conn) failConnect(err error) {
	sc.cleanupOnClose(err)
	if sc.nc != nil && sc.ncOwned {
		sc.nc.Close()
	}
}
func (sc *conn) pingServer() {
	p := &sc.ping
	p.mu.Lock()
	if p.timer == nil {
		p.mu.Unlock()
		return
	}
	p.out++
	if p.out > p.maxOut {
		p.mu.Unlock()
		sc.closeDueToPing(ErrMaxPings)
		return
	}
	p.timer.Reset(p.interval)
	p.mu.Unlock()
	if err := sc.nc.PublishRequest(p.requests, p.inbox, p.proto); err == nats.ErrConnectionClosed {
		sc.closeDueToPing(err)
	}
}
func (sc *conn) processPingResponse(m *nats.Msg) {
	if len(m.Data) > 0 {
		pingResp := &pb.PingResponse{}
		if err := pingResp.Unmarshal(m.Data); err != nil {
			return
		}
		if pingResp.Error != "" {
			sc.closeDueToPing(errors.New(pingResp.Error))
			return
		}
	} else if m.Header.Get("Status") == "503" {
		return
	}
	p := &sc.ping
	p.mu.Lock()
	p.out = 0
	p.mu.Unlock()
}
func (sc *conn) closeDueToPing(err error) {
	sc.Lock()
	if sc.closed {
		sc.Unlock()
		return
	}
	sc.cleanupOnClose(err)
	if sc.ncOwned && !sc.nc.IsClosed() {
		sc.nc.Close()
	}
	sc.closed = true
	cb := sc.connLostCB
	sc.Unlock()
	if cb != nil {
		go cb(sc, err)
	}
}
func (sc *conn) cleanupOnClose(err error) {
	p := &sc.ping
	p.mu.Lock()
	if p.timer != nil {
		p.timer.Stop()
		p.timer = nil
	}
	p.mu.Unlock()
	if !sc.ncOwned && !sc.nc.IsClosed() {
		if sc.hbSubscription != nil {
			sc.hbSubscription.Unsubscribe()
		}
		if p.sub != nil {
			p.sub.Unsubscribe()
		}
		if sc.ackSubscription != nil {
			sc.ackSubscription.Unsubscribe()
		}
	}
	if len(sc.pubAckMap) > 0 {
		acks := map[string]*ack{}
		for guid, pubAck := range sc.pubAckMap {
			if pubAck.t != nil && pubAck.t.Stop() {
				delete(sc.pubAckMap, guid)
				acks[guid] = pubAck
			}
		}
		if len(acks) > 0 {
			go func() {
				for guid, a := range acks {
					if a.ah != nil {
						a.ah(guid, ErrConnectionClosed)
					} else if a.ch != nil {
						a.ch <- ErrConnectionClosed
					}
				}
			}()
		}
	}
	close(sc.pubAckCloseChan)
}
func (sc *conn) Close() error {
	sc.Lock()
	defer sc.Unlock()
	if sc.closed {
		return nil
	}
	sc.closed = true
	if sc.ncOwned {
		defer sc.nc.Close()
	}
	sc.cleanupOnClose(ErrConnectionClosed)
	req := &pb.CloseRequest{ClientID: sc.clientID}
	b, _ := req.Marshal()
	reply, err := sc.nc.Request(sc.closeRequests, b, sc.opts.ConnectTimeout)
	if err != nil {
		if err == nats.ErrTimeout || err == nats.ErrNoResponders {
			return ErrCloseReqTimeout
		}
		return err
	}
	cr := &pb.CloseResponse{}
	err = cr.Unmarshal(reply.Data)
	if err != nil {
		return err
	}
	if cr.Error != "" {
		return errors.New(cr.Error)
	}
	return nil
}
func (sc *conn) NatsConn() *nats.Conn {
	sc.RLock()
	nc := sc.nc
	if sc.closed {
		nc = nil
	}
	sc.RUnlock()
	return nc
}
func (sc *conn) processHeartBeat(m *nats.Msg) {
	sc.nc.Publish(m.Reply, nil)
}
func (sc *conn) processAck(m *nats.Msg) {
	pa := &pb.PubAck{}
	err := pa.Unmarshal(m.Data)
	if err != nil {
		fmt.Printf("error during ack unmarshal: %v\n", err)
		return
	}
	a := sc.removeAck(pa.Guid)
	if a != nil {
		if pa.Error != "" {
			err = errors.New(pa.Error)
		}
		if a.ah != nil {
			a.ah(pa.Guid, err)
		} else if a.ch != nil {
			a.ch <- err
		}
	}
}
func (sc *conn) Publish(subject string, data []byte) error {
	ch := make(chan error, 1)
	_, err := sc.publishAsync(subject, data, nil, ch)
	if err == nil {
		err = <-ch
	}
	return err
}
func (sc *conn) PublishAsync(subject string, data []byte, ah AckHandler) (string, error) {
	return sc.publishAsync(subject, data, ah, nil)
}
func (sc *conn) publishAsync(subject string, data []byte, ah AckHandler, ch chan error) (string, error) {
	a := &ack{ah: ah, ch: ch}
	sc.Lock()
	if sc.closed {
		sc.Unlock()
		return "", ErrConnectionClosed
	}
	subj := sc.pubPrefix + "." + subject
	peGUID := sc.pubNUID.Next()
	pe := &pb.PubMsg{ClientID: sc.clientID, Guid: peGUID, Subject: subject, Data: data, ConnID: sc.connID}
	b, _ := pe.Marshal()
	sc.pubAckMap[peGUID] = a
	ackSubject := sc.ackSubject
	ackTimeout := sc.opts.AckTimeout
	sc.Unlock()
	select {
	case sc.pubAckChan <- struct{}{}:
	default:
		select {
		case sc.pubAckChan <- struct{}{}:
		case <-sc.pubAckCloseChan:
			return "", ErrConnectionClosed
		}
	}
	err := sc.nc.PublishRequest(subj, ackSubject, b)
	sc.Lock()
	if err != nil || sc.closed {
		sc.Unlock()
		removed := sc.removeAck(peGUID) != nil
		if removed || (ch == nil && ah == nil) {
			if err == nil {
				err = ErrConnectionClosed
			}
			return "", err
		}
		return peGUID, nil
	}
	a.t = time.AfterFunc(ackTimeout, func() {
		pubAck := sc.removeAck(peGUID)
		if pubAck == nil {
			return
		}
		if pubAck.ah != nil {
			pubAck.ah(peGUID, ErrTimeout)
		} else if a.ch != nil {
			pubAck.ch <- ErrTimeout
		}
	})
	sc.Unlock()
	return peGUID, nil
}