stan

README

NATS Streaming

NATS Streaming is an extremely performant, lightweight reliable streaming platform powered by NATS.

NATS Streaming provides the following high-level feature set:

• Log based persistence
• At-Least-Once Delivery model, giving reliable message delivery
• Rate matched on a per subscription basis
• Replay/Restart
• Last Value Semantics

Notes

• Please raise questions/issues via the Issue Tracker or via the #stan-preview channel on natsio.slack.com (contact larry@apcera.com or brian@apcera.com for access)

Known Issues

• Time- and sequence-based subscriptions are exact. Requesting a time or seqno before the earliest stored message for a subject will result in an error (in SubscriptionRequest.Error)

Installation

# Go client
go get github.com/nats-io/go-nats-streaming

For the Preview, stan-server is provided in binary form for Linux and Mac here

Basic Usage

sc, _ := stan.Connect(clusterID, clientID)

// Simple Synchronous Publisher
sc.Publish("foo", []byte("Hello World")) // does not return until an ack has been received from NATS Streaming

// Simple Async Subscriber
sub, _ := sc.Subscribe("foo", func(m *stan.Msg) {
    fmt.Printf("Received a message: %s\n", string(m.Data))
})

// Unsubscribe
sub.Unsubscribe()

// Close connection
sc.Close()

Subscription Start (i.e. Replay) Options

NATS Streaming subscriptions are similar to NATS subscriptions, but clients may start their subscription at an earlier point in the message stream, allowing them to receive messages that were published before this client registered interest.

The options are described with examples below:

// Subscribe starting with most recently published value
sub, err := sc.Subscribe("foo", func(m *stan.Msg) {
    fmt.Printf("Received a message: %s\n", string(m.Data))
}, StartWithLastReceived())

// Receive all stored values in order
sub, err := sc.Subscribe("foo", func(m *stan.Msg) {
    fmt.Printf("Received a message: %s\n", string(m.Data))
}, DeliverAllAvailable())

// Receive messages starting at a specific sequence number
sub, err := sc.Subscribe("foo", func(m *stan.Msg) {
    fmt.Printf("Received a message: %s\n", string(m.Data))
}, StartAtSequence(22))

// Subscribe starting at a specific time
var startTime time.Time
...
sub, err := sc.Subscribe("foo", func(m *stan.Msg) {
    fmt.Printf("Received a message: %s\n", string(m.Data))
}, StartAtTime(startTime))

// Subscribe starting a specific amount of time in the past (e.g. 30 seconds ago)
sub, err := sc.Subscribe("foo", func(m *stan.Msg) {
    fmt.Printf("Received a message: %s\n", string(m.Data))
}, StartAtTimeDelta(time.ParseDuration("30s")))

Durable Subscriptions

Replay of messages offers great flexibility for clients wishing to begin processing at some earlier point in the data stream. However, some clients just need to pick up where they left off from an earlier session, without having to manually track their position in the stream of messages. Durable subscriptions allow clients to assign a durable name to a subscription when it is created. Doing this causes the NATS Streaming server to track the last acknowledged message for that clientID + durable name, so that only messages since the last acknowledged message will be delivered to the client.

sc, _ := stan.Connect("test-cluster", "client-123")

// Subscribe with durable name
sc.Subscribe("foo", func(m *stan.Msg) {
    fmt.Printf("Received a message: %s\n", string(m.Data))
}, stan.DurableName("my-durable"))
...
// client receives message sequence 1-40
...
// client disconnects for an hour
...
// client reconnects with same clientID "client-123"
sc, _ := stan.Connect("test-cluster", "client-123")

// client re-subscribes to "foo" with same durable name "my-durable"
sc.Subscribe("foo", func(m *stan.Msg) {
    fmt.Printf("Received a message: %s\n", string(m.Data))
}, stan.DurableName("my-durable"))
...
// client receives messages 41-current

Wildcard Subscriptions

NATS Streaming subscriptions do not support wildcards.

Advanced Usage

Asynchronous Publishing

The basic publish API (Publish(subject, payload)) is synchronous; it does not return control to the caller until the NATS Streaming server has acknowledged receipt of the message. To accomplish this, a NUID is generated for the message on creation, and the client library waits for a publish acknowledgement from the server with a matching NUID before it returns control to the caller, possibly with an error indicating that the operation was not successful due to some server problem or authorization error.

Advanced users may wish to process these publish acknowledgements manually to achieve higher publish throughput by not waiting on individual acknowledgements during the publish operation. An asynchronous publish API is provided for this purpose:

    ackHandler := func(ackedNuid string, err error) {
        if err != nil {
            log.Printf("Warning: error publishing msg id %s: %v\n", ackedNuid, err.Error())
        } else {
            log.Printf("Received ack for msg id %s\n", ackedNuid)
        }
    }

    // can also use PublishAsyncWithReply(subj, replysubj, payload, ah)
    nuid, err := sc.PublishAsync("foo", []byte("Hello World"), ackHandler) // returns immediately
    if err != nil {
        log.Printf("Error publishing msg %s: %v\n", nuid, err.Error())
    }

Message Acknowledgements and Redelivery

NATS Streaming offers At-Least-Once delivery semantics, meaning that once a message has been delivered to an eligible subscriber, if an acknowledgement is not received within the configured timeout interval, NATS Streaming will attempt redelivery of the message. This timeout interval is specified by the subscription option AckWait, which defaults to 30 seconds.

By default, messages are automatically acknowledged by the NATS Streaming client library after the subscriber's message handler is invoked. However, there may be cases in which the subscribing client wishes to accelerate or defer acknowledgement of the message. To do this, the client must set manual acknowledgement mode on the subscription, and invoke Ack() on the Msg. ex:

// Subscribe with manual ack mode, and set AckWait to 60 seconds
aw, _ := time.ParseDuration("60s")
sub, err := sc.Subscribe("foo", func(m *stan.Msg) {
  m.Ack() // ack message before performing I/O intensive operation
  ///...
  fmt.Printf("Received a message: %s\n", string(m.Data))
}, stan.SetManualAckMode(), stan.AckWait(aw))

Rate limiting/matching

A classic problem of publish-subscribe messaging is matching the rate of message producers with the rate of message consumers. Message producers can often outpace the speed of the subscribers that are consuming their messages. This mismatch is commonly called a "fast producer/slow consumer" problem, and may result in dramatic resource utilization spikes in the underlying messaging system as it tries to buffer messages until the slow consumer(s) can catch up.

Publisher rate limiting

NATS Streaming provides a connection option called MaxPubAcksInflight that effectively limits the number of unacknowledged messages that a publisher may have in-flight at any given time. When this maximum is reached, further PublishAsync() calls will block until the number of unacknowledged messages falls below the specified limit. ex:

sc, _ := stan.Connect(clusterID, clientID, MaxPubAcksInflight(25))

ah := func(nuid string, err error) {
    // process the ack
    ...
}

for i := 1; i < 1000; i++ {
    // If the server is unable to keep up with the publisher, the number of oustanding acks will eventually
    // reach the max and this call will block
    guid, _ := sc.PublishAsync("foo", []byte("Hello World"), ah)
}

Subscriber rate limiting

Rate limiting may also be accomplished on the subscriber side, on a per-subscription basis, using a subscription option called MaxInflight. This option specifies the maximum number of outstanding acknowledgements (messages that have been delivered but not acknowledged) that NATS Streaming will allow for a given subscription. When this limit is reached, NATS Streaming will suspend delivery of messages to this subscription until the number of unacknowledged messages falls below the specified limit. ex:

// Subscribe with manual ack mode and a max in-flight limit of 25
sc.Subscribe("foo", func(m *stan.Msg) {
  fmt.Printf("Received message #: %s\n", string(m.Data))
  ...
  // Does not ack, or takes a very long time to ack
  ...
  // Message delivery will suspend when the number of unacknowledged messages reaches 25
}, stan.SetManualAckMode(), stan.MaxInflight(25))

License

(The MIT License)

Copyright (c) 2012-2016 Apcera Inc.

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

Documentation

Overview

Package stan is a Go client for the NATS Streaming messaging system (https://nats.io).

Package stan is a Go client for the NATS Streaming messaging system (https://nats.io).

Index

• Constants
• Variables
• type AckHandler
• type Conn
  • func Connect(stanClusterID, clientID string, options ...Option) (Conn, error)
• type Msg
  • func (msg *Msg) Ack() error
• type MsgHandler
• type Option
  • func ConnectWait(t time.Duration) Option
  • func MaxPubAcksInflight(max int) Option
  • func NatsConn(nc *nats.Conn) Option
  • func NatsURL(u string) Option
  • func PubAckWait(t time.Duration) Option
• type Options
• type Subscription
• type SubscriptionOption
  • func AckWait(t time.Duration) SubscriptionOption
  • func DeliverAllAvailable() SubscriptionOption
  • func DurableName(name string) SubscriptionOption
  • func MaxInflight(m int) SubscriptionOption
  • func SetManualAckMode() SubscriptionOption
  • func StartAt(sp pb.StartPosition) SubscriptionOption
  • func StartAtSequence(seq uint64) SubscriptionOption
  • func StartAtTime(start time.Time) SubscriptionOption
  • func StartAtTimeDelta(ago time.Duration) SubscriptionOption
  • func StartWithLastReceived() SubscriptionOption
• type SubscriptionOptions

Constants

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
)

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
)

const Version = "0.2.0"

Version is the NATS Streaming Go Client version

Variables

var (
	ErrConnectReqTimeout = errors.New("stan: connect request timeout")
	ErrCloseReqTimeout   = errors.New("stan: close 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")
)

Errors

var DefaultOptions = Options{
	NatsURL:            DefaultNatsURL,
	ConnectTimeout:     DefaultConnectWait,
	AckTimeout:         DefaultAckWait,
	DiscoverPrefix:     DefaultDiscoverPrefix,
	MaxPubAcksInflight: DefaultMaxPubAcksInflight,
}

DefaultOptions are the NATS Streaming client's default options

var DefaultSubscriptionOptions = SubscriptionOptions{
	MaxInflight: DefaultMaxInflight,
	AckWait:     DefaultAckWait,
}

DefaultSubscriptionOptions are the default subscriptions' options

Types

type AckHandler

type AckHandler func(string, error)

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 Conn

type Conn interface {
	// Publish
	Publish(subject string, data []byte) error
	PublishAsync(subject string, data []byte, ah AckHandler) (string, error)
	// Publish with Reply
	PublishWithReply(subject, reply string, data []byte) error
	PublishAsyncWithReply(subject, reply 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
}

Conn represents a connection to the NATS Streaming subsystem. It can Publish and Subscribe to messages within the NATS Streaming cluster.

func Connect

func Connect(stanClusterID, clientID string, options ...Option) (Conn, error)

Connect will form a connection to the NATS Streaming subsystem.

type Msg

type Msg struct {
	pb.MsgProto // MsgProto: Seq, Subject, Reply[opt], Data, Timestamp, CRC32[opt]
	Sub         Subscription
}

Msg is the client defined message, which includes proto, then back link to subscription.

func (*Msg) Ack

func (msg *Msg) Ack() error

Ack manually acknowledges a message. The subscriber had to be created with SetManualAckMode() option.

type MsgHandler

type MsgHandler func(msg *Msg)

MsgHandler is a callback function that processes messages delivered to asynchronous subscribers.

type Option

type Option func(*Options) error

Option is a function on the options for a connection.

func ConnectWait

func ConnectWait(t time.Duration) Option

ConnectWait is an Option to set the timeout for establishing a connection.

func MaxPubAcksInflight

func MaxPubAcksInflight(max int) Option

MaxPubAcksInflight is an Option to set the maximum number of published messages without outstanding ACKs from the server.

func NatsConn

func NatsConn(nc *nats.Conn) Option

NatsConn is an Option to set the underlying NATS connection to be used by a NATS Streaming Conn object.

func NatsURL

func NatsURL(u string) Option

NatsURL is an Option to set the URL the client should connect to.

func PubAckWait

func PubAckWait(t time.Duration) Option

PubAckWait is an Option to set the timeout for waiting for an ACK for a published message.

type Options

type Options struct {
	NatsURL            string
	NatsConn           *nats.Conn
	ConnectTimeout     time.Duration
	AckTimeout         time.Duration
	DiscoverPrefix     string
	MaxPubAcksInflight int
}

Options can be used to a create a customized connection.

type Subscription

type Subscription interface {
	Unsubscribe() error
}

Subscription represents a subscription within the NATS Streaming cluster. Subscriptions will be rate matched and follow at-least delivery semantics.

type SubscriptionOption

type SubscriptionOption func(*SubscriptionOptions) error

SubscriptionOption is a function on the options for a subscription.

func AckWait

func AckWait(t time.Duration) SubscriptionOption

AckWait is an Option to set the timeout for waiting for an ACK from the cluster's point of view for delivered messages.

func DeliverAllAvailable

func DeliverAllAvailable() SubscriptionOption

DeliverAllAvailable will deliver all messages available.

func DurableName

func DurableName(name string) SubscriptionOption

DurableName sets the DurableName for the subcriber.

func MaxInflight

func MaxInflight(m int) SubscriptionOption

MaxInflight is an Option to set the maximum number of messages the cluster will send without an ACK.

func SetManualAckMode

func SetManualAckMode() SubscriptionOption

SetManualAckMode will allow clients to control their own acks to delivered messages.

func StartAt

func StartAt(sp pb.StartPosition) SubscriptionOption

StartAt sets the desired start position for the message stream.

func StartAtSequence

func StartAtSequence(seq uint64) SubscriptionOption

StartAtSequence sets the desired start sequence position and state.

func StartAtTime

func StartAtTime(start time.Time) SubscriptionOption

StartAtTime sets the desired start time position and state.

func StartAtTimeDelta

func StartAtTimeDelta(ago time.Duration) SubscriptionOption

StartAtTimeDelta sets the desired start time position and state using the delta.

func StartWithLastReceived

func StartWithLastReceived() SubscriptionOption

StartWithLastReceived is a helper function to set start position to last received.

type SubscriptionOptions

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
}

SubscriptionOptions are used to control the Subscription's behavior.