Merge pull request #501 from mlaventure/new-shim-continued

New shim continued

vendor/github.com/nats-io/go-nats-streaming/stan.go

@@ -0,0 +1,476 @@
+// Copyright 2016 Apcera Inc. All rights reserved.
+
+// Package stan is a Go client for the NATS Streaming messaging system (https://nats.io).
+package stan
+
+import (
+	"errors"
+	"fmt"
+	"runtime"
+	"sync"
+	"time"
+
+	"github.com/nats-io/go-nats"
+	"github.com/nats-io/go-nats-streaming/pb"
+	"github.com/nats-io/nuid"
+)
+
+// Version is the NATS Streaming Go Client version
+const Version = "0.3.4"
+
+const (
+	// DefaultNatsURL is the default URL the client connects to
+	DefaultNatsURL = "nats://localhost:4222"
+	// DefaultConnectWait is the default timeout used for the connect operation
+	DefaultConnectWait = 2 * time.Second
+	// DefaultDiscoverPrefix is the prefix subject used to connect to the NATS Streaming server
+	DefaultDiscoverPrefix = "_STAN.discover"
+	// DefaultACKPrefix is the prefix subject used to send ACKs to the NATS Streaming server
+	DefaultACKPrefix = "_STAN.acks"
+	// DefaultMaxPubAcksInflight is the default maximum number of published messages
+	// without outstanding ACKs from the server
+	DefaultMaxPubAcksInflight = 16384
+)
+
+// Conn represents a connection to the NATS Streaming subsystem. It can Publish and
+// Subscribe to messages within the NATS Streaming cluster.
+type Conn interface {
+	// Publish
+	Publish(subject string, data []byte) error
+	PublishAsync(subject string, data []byte, ah AckHandler) (string, error)
+
+	// Subscribe
+	Subscribe(subject string, cb MsgHandler, opts ...SubscriptionOption) (Subscription, error)
+
+	// QueueSubscribe
+	QueueSubscribe(subject, qgroup string, cb MsgHandler, opts ...SubscriptionOption) (Subscription, error)
+
+	// Close
+	Close() error
+
+	// NatsConn returns the underlying NATS conn. Use this with care. For
+	// example, closing the wrapped NATS conn will put the NATS Streaming Conn
+	// in an invalid state.
+	NatsConn() *nats.Conn
+}
+
+// Errors
+var (
+	ErrConnectReqTimeout = errors.New("stan: connect request timeout")
+	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")
+)
+
+// AckHandler is used for Async Publishing to provide status of the ack.
+// The func will be passed teh GUID and any error state. No error means the
+// message was successfully received by NATS Streaming.
+type AckHandler func(string, error)
+
+// Options can be used to a create a customized connection.
+type Options struct {
+	NatsURL            string
+	NatsConn           *nats.Conn
+	ConnectTimeout     time.Duration
+	AckTimeout         time.Duration
+	DiscoverPrefix     string
+	MaxPubAcksInflight int
+}
+
+// DefaultOptions are the NATS Streaming client's default options
+var DefaultOptions = Options{
+	NatsURL:            DefaultNatsURL,
+	ConnectTimeout:     DefaultConnectWait,
+	AckTimeout:         DefaultAckWait,
+	DiscoverPrefix:     DefaultDiscoverPrefix,
+	MaxPubAcksInflight: DefaultMaxPubAcksInflight,
+}
+
+// Option is a function on the options for a connection.
+type Option func(*Options) error
+
+// NatsURL is an Option to set the URL the client should connect to.
+func NatsURL(u string) Option {
+	return func(o *Options) error {
+		o.NatsURL = u
+		return nil
+	}
+}
+
+// ConnectWait is an Option to set the timeout for establishing a connection.
+func ConnectWait(t time.Duration) Option {
+	return func(o *Options) error {
+		o.ConnectTimeout = t
+		return nil
+	}
+}
+
+// PubAckWait is an Option to set the timeout for waiting for an ACK for a
+// published message.
+func PubAckWait(t time.Duration) Option {
+	return func(o *Options) error {
+		o.AckTimeout = t
+		return nil
+	}
+}
+
+// MaxPubAcksInflight is an Option to set the maximum number of published
+// messages without outstanding ACKs from the server.
+func MaxPubAcksInflight(max int) Option {
+	return func(o *Options) error {
+		o.MaxPubAcksInflight = max
+		return nil
+	}
+}
+
+// NatsConn is an Option to set the underlying NATS connection to be used
+// by a NATS Streaming Conn object.
+func NatsConn(nc *nats.Conn) Option {
+	return func(o *Options) error {
+		o.NatsConn = nc
+		return nil
+	}
+}
+
+// A conn represents a bare connection to a stan cluster.
+type conn struct {
+	sync.RWMutex
+	clientID         string
+	serverID         string
+	pubPrefix        string // Publish prefix set by stan, append our subject.
+	subRequests      string // Subject to send subscription requests.
+	unsubRequests    string // Subject to send unsubscribe requests.
+	subCloseRequests string // Subject to send subscription close requests.
+	closeRequests    string // Subject to send close requests.
+	ackSubject       string // publish acks
+	ackSubscription  *nats.Subscription
+	hbSubscription   *nats.Subscription
+	subMap           map[string]*subscription
+	pubAckMap        map[string]*ack
+	pubAckChan       chan (struct{})
+	opts             Options
+	nc               *nats.Conn
+	ncOwned          bool // NATS Streaming created the connection, so needs to close it.
+}
+
+// Closure for ack contexts.
+type ack struct {
+	t  *time.Timer
+	ah AckHandler
+	ch chan error
+}
+
+// Connect will form a connection to the NATS Streaming subsystem.
+func Connect(stanClusterID, clientID string, options ...Option) (Conn, error) {
+	// Process Options
+	c := conn{clientID: clientID, opts: DefaultOptions}
+	for _, opt := range options {
+		if err := opt(&c.opts); err != nil {
+			return nil, err
+		}
+	}
+	// Check if the user has provided a connection as an option
+	c.nc = c.opts.NatsConn
+	// Create a NATS connection if it doesn't exist.
+	if c.nc == nil {
+		nc, err := nats.Connect(c.opts.NatsURL)
+		if err != nil {
+			return nil, err
+		}
+		c.nc = nc
+		c.ncOwned = true
+	} else if !c.nc.IsConnected() {
+		// Bail if the custom NATS connection is disconnected
+		return nil, ErrBadConnection
+	}
+
+	// Create a heartbeat inbox
+	hbInbox := nats.NewInbox()
+	var err error
+	if c.hbSubscription, err = c.nc.Subscribe(hbInbox, c.processHeartBeat); err != nil {
+		c.Close()
+		return nil, err
+	}
+
+	// Send Request to discover the cluster
+	discoverSubject := c.opts.DiscoverPrefix + "." + stanClusterID
+	req := &pb.ConnectRequest{ClientID: clientID, HeartbeatInbox: hbInbox}
+	b, _ := req.Marshal()
+	reply, err := c.nc.Request(discoverSubject, b, c.opts.ConnectTimeout)
+	if err != nil {
+		c.Close()
+		if err == nats.ErrTimeout {
+			return nil, ErrConnectReqTimeout
+		}
+		return nil, err
+	}
+	// Process the response, grab server pubPrefix
+	cr := &pb.ConnectResponse{}
+	err = cr.Unmarshal(reply.Data)
+	if err != nil {
+		c.Close()
+		return nil, err
+	}
+	if cr.Error != "" {
+		c.Close()
+		return nil, errors.New(cr.Error)
+	}
+
+	// Capture cluster configuration endpoints to publish and subscribe/unsubscribe.
+	c.pubPrefix = cr.PubPrefix
+	c.subRequests = cr.SubRequests
+	c.unsubRequests = cr.UnsubRequests
+	c.subCloseRequests = cr.SubCloseRequests
+	c.closeRequests = cr.CloseRequests
+
+	// Setup the ACK subscription
+	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(1024*1024, 32*1024*1024)
+	c.pubAckMap = make(map[string]*ack)
+
+	// Create Subscription map
+	c.subMap = make(map[string]*subscription)
+
+	c.pubAckChan = make(chan struct{}, c.opts.MaxPubAcksInflight)
+
+	// Attach a finalizer
+	runtime.SetFinalizer(&c, func(sc *conn) { sc.Close() })
+
+	return &c, nil
+}
+
+// Close a connection to the stan system.
+func (sc *conn) Close() error {
+	if sc == nil {
+		return ErrBadConnection
+	}
+
+	sc.Lock()
+	defer sc.Unlock()
+
+	if sc.nc == nil {
+		// We are already closed.
+		return nil
+	}
+
+	// Capture for NATS calls below.
+	nc := sc.nc
+	if sc.ncOwned {
+		defer nc.Close()
+	}
+
+	// Signals we are closed.
+	sc.nc = nil
+
+	// Now close ourselves.
+	if sc.ackSubscription != nil {
+		sc.ackSubscription.Unsubscribe()
+	}
+
+	req := &pb.CloseRequest{ClientID: sc.clientID}
+	b, _ := req.Marshal()
+	reply, err := nc.Request(sc.closeRequests, b, sc.opts.ConnectTimeout)
+	if err != nil {
+		if err == nats.ErrTimeout {
+			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
+}
+
+// NatsConn returns the underlying NATS conn. Use this with care. For example,
+// closing the wrapped NATS conn will put the NATS Streaming Conn in an invalid
+// state.
+func (sc *conn) NatsConn() *nats.Conn {
+	return sc.nc
+}
+
+// Process a heartbeat from the NATS Streaming cluster
+func (sc *conn) processHeartBeat(m *nats.Msg) {
+	// No payload assumed, just reply.
+	sc.RLock()
+	nc := sc.nc
+	sc.RUnlock()
+	if nc != nil {
+		nc.Publish(m.Reply, nil)
+	}
+}
+
+// Process an ack from the NATS Streaming cluster
+func (sc *conn) processAck(m *nats.Msg) {
+	pa := &pb.PubAck{}
+	err := pa.Unmarshal(m.Data)
+	if err != nil {
+		// FIXME, make closure to have context?
+		fmt.Printf("Error processing unmarshal\n")
+		return
+	}
+
+	// Remove
+	a := sc.removeAck(pa.Guid)
+	if a != nil {
+		// Capture error if it exists.
+		if pa.Error != "" {
+			err = errors.New(pa.Error)
+		}
+		if a.ah != nil {
+			// Perform the ackHandler callback
+			a.ah(pa.Guid, err)
+		} else if a.ch != nil {
+			// Send to channel directly
+			a.ch <- err
+		}
+	}
+}
+
+// Publish will publish to the cluster and wait for an ACK.
+func (sc *conn) Publish(subject string, data []byte) error {
+	ch := make(chan error)
+	_, err := sc.publishAsync(subject, data, nil, ch)
+	if err == nil {
+		err = <-ch
+	}
+	return err
+}
+
+// PublishAsync will publish to the cluster on pubPrefix+subject and asynchronously
+// process the ACK or error state. It will return the GUID for the message being sent.
+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.nc == nil {
+		sc.Unlock()
+		return "", ErrConnectionClosed
+	}
+
+	subj := sc.pubPrefix + "." + subject
+	// This is only what we need from PubMsg in the timer below,
+	// so do this so that pe doesn't escape (and we same on new object)
+	peGUID := nuid.Next()
+	pe := &pb.PubMsg{ClientID: sc.clientID, Guid: peGUID, Subject: subject, Data: data}
+	b, _ := pe.Marshal()
+
+	// Map ack to guid.
+	sc.pubAckMap[peGUID] = a
+	// snapshot
+	ackSubject := sc.ackSubject
+	ackTimeout := sc.opts.AckTimeout
+	pac := sc.pubAckChan
+	sc.Unlock()
+
+	// Use the buffered channel to control the number of outstanding acks.
+	pac <- struct{}{}
+
+	err := sc.nc.PublishRequest(subj, ackSubject, b)
+	if err != nil {
+		sc.removeAck(peGUID)
+		return "", err
+	}
+
+	// Setup the timer for expiration.
+	sc.Lock()
+	a.t = time.AfterFunc(ackTimeout, func() {
+		sc.removeAck(peGUID)
+		if a.ah != nil {
+			ah(peGUID, ErrTimeout)
+		} else if a.ch != nil {
+			a.ch <- ErrTimeout
+		}
+	})
+	sc.Unlock()
+
+	return peGUID, nil
+}
+
+// removeAck removes the ack from the pubAckMap and cancels any state, e.g. timers
+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()
+
+	// Cancel timer if needed.
+	if t != nil {
+		t.Stop()
+	}
+
+	// Remove from channel to unblock PublishAsync
+	if a != nil && len(pac) > 0 {
+		<-pac
+	}
+	return a
+}
+
+// Process an msg from the NATS Streaming cluster
+func (sc *conn) processMsg(raw *nats.Msg) {
+	msg := &Msg{}
+	err := msg.Unmarshal(raw.Data)
+	if err != nil {
+		panic("Error processing unmarshal for msg")
+	}
+	// Lookup the subscription
+	sc.RLock()
+	nc := sc.nc
+	isClosed := nc == nil
+	sub := sc.subMap[raw.Subject]
+	sc.RUnlock()
+
+	// Check if sub is no longer valid or connection has been closed.
+	if sub == nil || isClosed {
+		return
+	}
+
+	// Store in msg for backlink
+	msg.Sub = sub
+
+	sub.RLock()
+	cb := sub.cb
+	ackSubject := sub.ackInbox
+	isManualAck := sub.opts.ManualAcks
+	subsc := sub.sc // Can be nil if sub has been unsubscribed.
+	sub.RUnlock()
+
+	// Perform the callback
+	if cb != nil && subsc != nil {
+		cb(msg)
+	}
+
+	// Proces auto-ack
+	if !isManualAck && nc != nil {
+		ack := &pb.Ack{Subject: msg.Subject, Sequence: msg.Sequence}
+		b, _ := ack.Marshal()
+		if err := nc.Publish(ackSubject, b); err != nil {
+			// FIXME(dlc) - Async error handler? Retry?
+		}
+	}
+}

vendor/github.com/nats-io/go-nats-streaming/sub.go

@@ -0,0 +1,376 @@
+// Copyright 2016 Apcera Inc. All rights reserved.
+
+// Package stan is a Go client for the NATS Streaming messaging system (https://nats.io).
+package stan
+
+import (
+	"errors"
+	"sync"
+	"time"
+
+	"github.com/nats-io/go-nats"
+	"github.com/nats-io/go-nats-streaming/pb"
+)
+
+const (
+	// DefaultAckWait indicates how long the server should wait for an ACK before resending a message
+	DefaultAckWait = 30 * time.Second
+	// DefaultMaxInflight indicates how many messages with outstanding ACKs the server can send
+	DefaultMaxInflight = 1024
+)
+
+// Msg is the client defined message, which includes proto, then back link to subscription.
+type Msg struct {
+	pb.MsgProto // MsgProto: Seq, Subject, Reply[opt], Data, Timestamp, CRC32[opt]
+	Sub         Subscription
+}
+
+// Subscriptions and Options
+
+// Subscription represents a subscription within the NATS Streaming cluster. Subscriptions
+// will be rate matched and follow at-least delivery semantics.
+type Subscription interface {
+	// Unsubscribe removes interest in the subscription.
+	// For durables, it means that the durable interest is also removed from
+	// the server. Restarting a durable with the same name will not resume
+	// the subscription, it will be considered a new one.
+	Unsubscribe() error
+
+	// Close removes this subscriber from the server, but unlike Unsubscribe(),
+	// the durable interest is not removed. If the client has connected to a server
+	// for which this feature is not available, Close() will return a ErrNoServerSupport
+	// error.
+	Close() error
+}
+
+// A subscription represents a subscription to a stan cluster.
+type subscription struct {
+	sync.RWMutex
+	sc       *conn
+	subject  string
+	qgroup   string
+	inbox    string
+	ackInbox string
+	inboxSub *nats.Subscription
+	opts     SubscriptionOptions
+	cb       MsgHandler
+}
+
+// SubscriptionOption is a function on the options for a subscription.
+type SubscriptionOption func(*SubscriptionOptions) error
+
+// MsgHandler is a callback function that processes messages delivered to
+// asynchronous subscribers.
+type MsgHandler func(msg *Msg)
+
+// SubscriptionOptions are used to control the Subscription's behavior.
+type SubscriptionOptions struct {
+	// DurableName, if set will survive client restarts.
+	DurableName string
+	// Controls the number of messages the cluster will have inflight without an ACK.
+	MaxInflight int
+	// Controls the time the cluster will wait for an ACK for a given message.
+	AckWait time.Duration
+	// StartPosition enum from proto.
+	StartAt pb.StartPosition
+	// Optional start sequence number.
+	StartSequence uint64
+	// Optional start time.
+	StartTime time.Time
+	// Option to do Manual Acks
+	ManualAcks bool
+}
+
+// DefaultSubscriptionOptions are the default subscriptions' options
+var DefaultSubscriptionOptions = SubscriptionOptions{
+	MaxInflight: DefaultMaxInflight,
+	AckWait:     DefaultAckWait,
+}
+
+// MaxInflight is an Option to set the maximum number of messages the cluster will send
+// without an ACK.
+func MaxInflight(m int) SubscriptionOption {
+	return func(o *SubscriptionOptions) error {
+		o.MaxInflight = m
+		return nil
+	}
+}
+
+// AckWait is an Option to set the timeout for waiting for an ACK from the cluster's
+// point of view for delivered messages.
+func AckWait(t time.Duration) SubscriptionOption {
+	return func(o *SubscriptionOptions) error {
+		o.AckWait = t
+		return nil
+	}
+}
+
+// StartAt sets the desired start position for the message stream.
+func StartAt(sp pb.StartPosition) SubscriptionOption {
+	return func(o *SubscriptionOptions) error {
+		o.StartAt = sp
+		return nil
+	}
+}
+
+// StartAtSequence sets the desired start sequence position and state.
+func StartAtSequence(seq uint64) SubscriptionOption {
+	return func(o *SubscriptionOptions) error {
+		o.StartAt = pb.StartPosition_SequenceStart
+		o.StartSequence = seq
+		return nil
+	}
+}
+
+// StartAtTime sets the desired start time position and state.
+func StartAtTime(start time.Time) SubscriptionOption {
+	return func(o *SubscriptionOptions) error {
+		o.StartAt = pb.StartPosition_TimeDeltaStart
+		o.StartTime = start
+		return nil
+	}
+}
+
+// StartAtTimeDelta sets the desired start time position and state using the delta.
+func StartAtTimeDelta(ago time.Duration) SubscriptionOption {
+	return func(o *SubscriptionOptions) error {
+		o.StartAt = pb.StartPosition_TimeDeltaStart
+		o.StartTime = time.Now().Add(-ago)
+		return nil
+	}
+}
+
+// StartWithLastReceived is a helper function to set start position to last received.
+func StartWithLastReceived() SubscriptionOption {
+	return func(o *SubscriptionOptions) error {
+		o.StartAt = pb.StartPosition_LastReceived
+		return nil
+	}
+}
+
+// DeliverAllAvailable will deliver all messages available.
+func DeliverAllAvailable() SubscriptionOption {
+	return func(o *SubscriptionOptions) error {
+		o.StartAt = pb.StartPosition_First
+		return nil
+	}
+}
+
+// SetManualAckMode will allow clients to control their own acks to delivered messages.
+func SetManualAckMode() SubscriptionOption {
+	return func(o *SubscriptionOptions) error {
+		o.ManualAcks = true
+		return nil
+	}
+}
+
+// DurableName sets the DurableName for the subcriber.
+func DurableName(name string) SubscriptionOption {
+	return func(o *SubscriptionOptions) error {
+		o.DurableName = name
+		return nil
+	}
+}
+
+// Subscribe will perform a subscription with the given options to the NATS Streaming cluster.
+func (sc *conn) Subscribe(subject string, cb MsgHandler, options ...SubscriptionOption) (Subscription, error) {
+	return sc.subscribe(subject, "", cb, options...)
+}
+
+// QueueSubscribe will perform a queue subscription with the given options to the NATS Streaming cluster.
+func (sc *conn) QueueSubscribe(subject, qgroup string, cb MsgHandler, options ...SubscriptionOption) (Subscription, error) {
+	return sc.subscribe(subject, qgroup, cb, options...)
+}
+
+// subscribe will perform a subscription with the given options to the NATS Streaming cluster.
+func (sc *conn) subscribe(subject, qgroup string, cb MsgHandler, options ...SubscriptionOption) (Subscription, error) {
+	sub := &subscription{subject: subject, qgroup: qgroup, inbox: nats.NewInbox(), cb: cb, sc: sc, opts: DefaultSubscriptionOptions}
+	for _, opt := range options {
+		if err := opt(&sub.opts); err != nil {
+			return nil, err
+		}
+	}
+	sc.Lock()
+	if sc.nc == nil {
+		sc.Unlock()
+		return nil, ErrConnectionClosed
+	}
+
+	// Register subscription.
+	sc.subMap[sub.inbox] = sub
+	nc := sc.nc
+	sc.Unlock()
+
+	// Hold lock throughout.
+	sub.Lock()
+	defer sub.Unlock()
+
+	// Listen for actual messages.
+	nsub, err := nc.Subscribe(sub.inbox, sc.processMsg)
+	if err != nil {
+		return nil, err
+	}
+	sub.inboxSub = nsub
+
+	// Create a subscription request
+	// FIXME(dlc) add others.
+	sr := &pb.SubscriptionRequest{
+		ClientID:      sc.clientID,
+		Subject:       subject,
+		QGroup:        qgroup,
+		Inbox:         sub.inbox,
+		MaxInFlight:   int32(sub.opts.MaxInflight),
+		AckWaitInSecs: int32(sub.opts.AckWait / time.Second),
+		StartPosition: sub.opts.StartAt,
+		DurableName:   sub.opts.DurableName,
+	}
+
+	// Conditionals
+	switch sr.StartPosition {
+	case pb.StartPosition_TimeDeltaStart:
+		sr.StartTimeDelta = time.Now().UnixNano() - sub.opts.StartTime.UnixNano()
+	case pb.StartPosition_SequenceStart:
+		sr.StartSequence = sub.opts.StartSequence
+	}
+
+	b, _ := sr.Marshal()
+	reply, err := sc.nc.Request(sc.subRequests, b, sc.opts.ConnectTimeout)
+	if err != nil {
+		sub.inboxSub.Unsubscribe()
+		if err == nats.ErrTimeout {
+			err = ErrSubReqTimeout
+		}
+		return nil, err
+	}
+	r := &pb.SubscriptionResponse{}
+	if err := r.Unmarshal(reply.Data); err != nil {
+		sub.inboxSub.Unsubscribe()
+		return nil, err
+	}
+	if r.Error != "" {
+		sub.inboxSub.Unsubscribe()
+		return nil, errors.New(r.Error)
+	}
+	sub.ackInbox = r.AckInbox
+
+	return sub, nil
+}
+
+// closeOrUnsubscribe performs either close or unsubsribe based on
+// given boolean.
+func (sub *subscription) closeOrUnsubscribe(doClose bool) error {
+	if sub == nil {
+		return ErrBadSubscription
+	}
+	sub.Lock()
+	sc := sub.sc
+	if sc == nil {
+		// Already closed.
+		sub.Unlock()
+		return ErrBadSubscription
+	}
+	sub.sc = nil
+	sub.inboxSub.Unsubscribe()
+	sub.inboxSub = nil
+	sub.Unlock()
+
+	if sc == nil {
+		return ErrBadSubscription
+	}
+
+	sc.Lock()
+	if sc.nc == nil {
+		sc.Unlock()
+		return ErrConnectionClosed
+	}
+
+	delete(sc.subMap, sub.inbox)
+	reqSubject := sc.unsubRequests
+	if doClose {
+		reqSubject = sc.subCloseRequests
+		if reqSubject == "" {
+			sc.Unlock()
+			return ErrNoServerSupport
+		}
+	}
+
+	// Snapshot connection to avoid data race, since the connection may be
+	// closing while we try to send the request
+	nc := sc.nc
+	sc.Unlock()
+
+	usr := &pb.UnsubscribeRequest{
+		ClientID: sc.clientID,
+		Subject:  sub.subject,
+		Inbox:    sub.ackInbox,
+	}
+	b, _ := usr.Marshal()
+	reply, err := nc.Request(reqSubject, b, sc.opts.ConnectTimeout)
+	if err != nil {
+		if err == nats.ErrTimeout {
+			if doClose {
+				return ErrCloseReqTimeout
+			}
+			return ErrUnsubReqTimeout
+		}
+		return err
+	}
+	r := &pb.SubscriptionResponse{}
+	if err := r.Unmarshal(reply.Data); err != nil {
+		return err
+	}
+	if r.Error != "" {
+		return errors.New(r.Error)
+	}
+
+	return nil
+}
+
+// Unsubscribe implements the Subscription interface
+func (sub *subscription) Unsubscribe() error {
+	return sub.closeOrUnsubscribe(false)
+}
+
+// Close implements the Subscription interface
+func (sub *subscription) Close() error {
+	return sub.closeOrUnsubscribe(true)
+}
+
+// Ack manually acknowledges a message.
+// The subscriber had to be created with SetManualAckMode() option.
+func (msg *Msg) Ack() error {
+	if msg == nil {
+		return ErrNilMsg
+	}
+	// Look up subscription
+	sub := msg.Sub.(*subscription)
+	if sub == nil {
+		return ErrBadSubscription
+	}
+
+	sub.RLock()
+	ackSubject := sub.ackInbox
+	isManualAck := sub.opts.ManualAcks
+	sc := sub.sc
+	sub.RUnlock()
+
+	// Check for error conditions.
+	if sc == nil {
+		return ErrBadSubscription
+	}
+	// Get nc from the connection (needs locking to avoid race)
+	sc.RLock()
+	nc := sc.nc
+	sc.RUnlock()
+	if nc == nil {
+		return ErrBadConnection
+	}
+	if !isManualAck {
+		return ErrManualAck
+	}
+
+	// Ack here.
+	ack := &pb.Ack{Subject: msg.Subject, Sequence: msg.Sequence}
+	b, _ := ack.Marshal()
+	return nc.Publish(ackSubject, b)
+}