stan

package module
Version: v0.10.2 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Nov 17, 2021 License: Apache-2.0 Imports: 7 Imported by: 341

README

NATS Streaming

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

License Apache 2 Build Status Coverage Status GoDoc

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

Installation

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

When using or transitioning to Go modules support:

# Go client latest or explicit version
go get github.com/nats-io/stan.go/@latest
go get github.com/nats-io/stan.go/@v0.10.2

Important things to know about reconnections.

A common misunderstanding from users moving from NATS to NATS Streaming has to do with how reconnection works. It is important to understand how NATS Streaming relates to NATS "core". You can find some information in NATS Streaming Concepts/Relation to NATS.

The NATS Streaming library uses the NATS library to connect to a NATS Server and indirectly communicates with the NATS Streaming "server". To better understand the issues, you should assume that the server has no direct connection with the client, and the client may possibly never lose its TCP connection to a NATS Server and yet may not have access to a streaming server (streaming client connected to a NATS Server, that is cluster to another, to which the NATS Streaming "server" is connected to).

When the low-level NATS TCP connection is broken, the reconnection logic triggers in the core NATS library. Once the connection is re-established, the low level NATS subscriptions used by the Streaming library to communicate with the streaming servers will be resent. All of that activity could have happened and the Streaming server would not know (see topology example described above). But from the server that is not a problem, after all, if a message is delivered while the client is disconnected, the server won't receive an ACK and will redeliver that message after the AckWait interval.

At this point, it is worth noting that a frequent mistake made by new users is to run the NATS Streaming server with memory store (the default if no persistence mode is specified) and after the server is restarted, and the client "reconnects" fine, the application stops receiving messages and/or gets errors on publish. The note in NATS Streaming Concepts/Client Connections will detail what is happening in this case.

To maintain the streaming connection (a better name may have been "session"), both server and clients send heartbeats/PINGs. If the server misses a configured amount of heartbeats from the client, it will close the connection, which also means deleting all non-durable subscriptions. If the client was "network partitioned" from the server when that happened, even after the partition is resolved, the client would not know what happened. Again, to understand how that is possible, see the topology example above: the network partition happened between the two clustered NATS Servers, and no TCP connection between the streaming client and/or streaming server was ever lost.

To solve that, the client sends PINGs to the server, and if missing enough of them, will close the connection and report it as lost through the ConnectionLost handler. See Connection Status for more details. In the case of the network partition example above, even if the client's number of PINGs has not reached its maximum count when the partition is resolved, the server will respond with an error to the next PING going through because it will detect that this specific client instance has been closed due to timeouts.

When that point is reached, the connection and its subscriptions are no longer valid and need to be recreated by the user.

The client-to-server PINGs are by default set to pretty aggressive values and should likely be increased in a normal setup. That is, with the default values, the client would not tolerate a server being down/not responding for only 15 seconds or so. Users should adjust the Pings() option, deciding how long they are willing to have their application not able to communicate with a server without "knowing", versus declaring the connection lost too soon (and having to recreate the state: connection and all subscriptions).

Basic Usage

import stan "github.com/nats-io/stan.go"

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))
}, stan.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))
}, stan.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))
}, stan.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))
}, stan.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))
}, stan.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
Queue Groups

All subscriptions with the same queue name (regardless of the connection they originate from) will form a queue group. Each message will be delivered to only one subscriber per queue group, using queuing semantics. You can have as many queue groups as you wish.

Normal subscribers will continue to work as expected.

Creating a Queue Group

A queue group is automatically created when the first queue subscriber is created. If the group already exists, the member is added to the group.

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

// Create a queue subscriber on "foo" for group "bar"
qsub1, _ := sc.QueueSubscribe("foo", "bar", qcb)

// Add a second member
qsub2, _ := sc.QueueSubscribe("foo", "bar", qcb)

// Notice that you can have a regular subscriber on that subject
sub, _ := sc.Subscribe("foo", cb)

// A message on "foo" will be received by sub and qsub1 or qsub2.
Start Position

Note that once a queue group is formed, a member's start position is ignored when added to the group. It will start receive messages from the last position in the group.

Suppose the channel foo exists and there are 500 messages stored, the group bar is already created, there are two members and the last message sequence sent is 100. A new member is added. Note its start position:

sc.QueueSubscribe("foo", "bar", qcb, stan.StartAtSequence(200))

This will not produce an error, but the start position will be ignored. Assuming this member would be the one receiving the next message, it would receive message sequence 101.

Leaving the Group

There are two ways of leaving the group: closing the subscriber's connection or calling Unsubscribe:

// Have qsub leave the queue group
qsub.Unsubscribe()

If the leaving member had un-acknowledged messages, those messages are reassigned to the remaining members.

Closing a Queue Group

There is no special API for that. Once all members have left (either calling Unsubscribe, or their connections are closed), the group is removed from the server.

The next call to QueueSubscribe with the same group name will create a brand new group, that is, the start position will take effect and delivery will start from there.

Durable Queue Groups

As described above, for non durable queue subscribers, when the last member leaves the group, that group is removed. A durable queue group allows you to have all members leave but still maintain state. When a member re-joins, it starts at the last position in that group.

Creating a Durable Queue Group

A durable queue group is created in a similar manner as that of a standard queue group, except the DurableName option must be used to specify durability.

sc.QueueSubscribe("foo", "bar", qcb, stan.DurableName("dur"))

A group called dur:bar (the concatenation of durable name and group name) is created in the server. This means two things:

  • The character : is not allowed for a queue subscriber's durable name.
  • Durable and non-durable queue groups with the same name can coexist.
// Non durable queue subscriber on group "bar"
qsub, _ := sc.QueueSubscribe("foo", "bar", qcb)

// Durable queue subscriber on group "bar"
durQsub, _ := sc.QueueSubscribe("foo", "bar", qcb, stan.DurableName("mydurablegroup"))

// The same message produced on "foo" would be received by both queue subscribers.
Start Position

The rules for non-durable queue subscribers apply to durable subscribers.

Leaving the Group

As for non-durable queue subscribers, if a member's connection is closed, or if Unsubscribe its called, the member leaves the group. Any unacknowledged message is transferred to remaining members. See Closing the Group for important difference with non-durable queue subscribers.

Closing the Group

The last member calling Unsubscribe will close (that is destroy) the group. So if you want to maintain durability of the group, you should not be calling Unsubscribe.

So unlike for non-durable queue subscribers, it is possible to maintain a queue group with no member in the server. When a new member re-joins the durable queue group, it will resume from where the group left of, actually first receiving all unacknowledged messages that may have been left when the last member previously left.

Wildcard Subscriptions

NATS Streaming subscriptions do not support wildcards.

Advanced Usage

Connection configuration such as TLS, etc..

If you want more advanced configuration of the underlying NATS Connection, you will need to create a NATS connection and pass that connection to the stan.Connect() call with the stan.NatsConn() option.

// Create a NATS connection that you can configure the way you want
nc, err = nats.Connect("tls://localhost:4443", nats.ClientCert("mycerts/client-cert.pem", "mycerts/client-key.pem"))
if (err != nil)
 ...

// Then pass it to the stan.Connect() call.
sc, err = stan.Connect("test-cluster", "me", stan.NatsConn(nc))
if (err != nil)
 ...

// Note that you will be responsible for closing the NATS Connection after the streaming
// connection has been closed.
Connection Status

The fact that the NATS Streaming server and clients are not directly connected poses a challenge when it comes to know if a client is still valid. When a client disconnects, the streaming server is not notified, hence the importance of calling Close(). The server sends heartbeats to the client's private inbox and if it misses a certain number of responses, it will consider the client's connection lost and remove it from its state.

Before version 0.4.0, the client library was not sending PINGs to the streaming server to detect connection failure. This was problematic especially if an application was never sending data (had only subscriptions for instance). Picture the case where a client connects to a NATS Server which has a route to a NATS Streaming server (either connecting to a standalone NATS Server or the server it embeds). If the connection between the streaming server and the client's NATS Server is broken, the client's NATS connection would still be ok, yet, no communication with the streaming server is possible. This is why relying on Conn.NatsConn() to check the status is not helpful.

Starting version 0.4.0 of this library and server 0.10.0, the client library will now send PINGs at regular intervals (default is 5 seconds) and will close the streaming connection after a certain number of PINGs have been sent without any response (default is 3). When that happens, a callback - if one is registered - will be invoked to notify the user that the connection is permanently lost, and the reason for the failure.

Here is how you would specify your own PING values and the callback:


    // Send PINGs every 10 seconds, and fail after 5 PINGs without any response.
    sc, err := stan.Connect(clusterName, clientName,
        stan.Pings(10, 5),
        stan.SetConnectionLostHandler(func(_ stan.Conn, reason error) {
            log.Fatalf("Connection lost, reason: %v", reason)
        }))

Note that the only way to be notified is to set the callback. If the callback is not set, PINGs are still sent and the connection will be closed if needed, but the application won't know if it has only subscriptions.

When the connection is lost, your application would have to re-create it and all subscriptions if any.

When no NATS connection is provided to the Connect() call, the library creates its own NATS connection and will now set the reconnect attempts to "infinite", which was not the case before. It should therefore be possible for the library to always reconnect, but this does not mean that the streaming connection will not be closed, even if you set a very high threshold for the PINGs max out value. Keep in mind that while the client is disconnected, the server is sending heartbeats to the clients too, and when not getting any response, it will remove that client from its state. When the communication is restored, the PINGs sent to the server will allow to detect this condition and report to the client that the connection is now closed.

Also, while a client is "disconnected" from the server, another application with connectivity to the streaming server may connect and uses the same client ID. The server, when detecting the duplicate client ID, will try to contact the first client to know if it should reject the connect request of the second client. Since the communication between the server and the first client is broken, the server will not get a response and therefore will replace the first client with the second one.

Prior to client 0.4.0 and server 0.10.0, if the communication between the first client and server were to be restored, and the application would send messages, the server would accept those because the published messages client ID would be valid, although the client is not. With client at 0.4.0+ and server 0.10.0+, additional information is sent with each message to allow the server to reject messages from a client that has been replaced by another client.

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)
        }
    }

    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 outstanding 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

Unless otherwise noted, the NATS source files are distributed under the Apache Version 2.0 license found in the LICENSE file.

Documentation

Overview

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

Index

Constants

View Source
const (
	// DefaultNatsURL is the default URL the client connects to
	DefaultNatsURL = "nats://127.0.0.1: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
	// DefaultPingInterval is the default interval (in seconds) at which a connection sends a PING to the server
	DefaultPingInterval = 5
	// DefaultPingMaxOut is the number of PINGs without a response before the connection is considered lost.
	DefaultPingMaxOut = 88
)
View Source
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
)
View Source
const Version = "0.10.2"

Version is the NATS Streaming Go Client version

Variables

View Source
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")
)

Errors

View Source
var DefaultOptions = getDefaultOptions()

DEPRECATED: Use getDefaultOptions() instead in the library, neither in clients. DefaultOptions is not safe for use by multiple clients. For details see https://github.com/nats-io/nats.go/issues/308. DefaultOptions are the NATS Streaming client's default options

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

DefaultSubscriptionOptions are the default subscriptions' options

Functions

This section is empty.

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 the GUID and any error state. No error means the message was successfully received by NATS Streaming.

type Conn

type Conn interface {
	// Publish will publish to the cluster and wait for an ACK.
	Publish(subject string, data []byte) error

	// PublishAsync will publish to the cluster and asynchronously process
	// the ACK or error state. It will return the GUID for the message being sent.
	PublishAsync(subject string, data []byte, ah AckHandler) (string, error)

	// Subscribe will perform a subscription with the given options to the cluster.
	//
	// If no option is specified, DefaultSubscriptionOptions are used. The default start
	// position is to receive new messages only (messages published after the subscription is
	// registered in the cluster).
	Subscribe(subject string, cb MsgHandler, opts ...SubscriptionOption) (Subscription, error)

	// QueueSubscribe will perform a queue subscription with the given options to the cluster.
	//
	// If no option is specified, DefaultSubscriptionOptions are used. The default start
	// position is to receive new messages only (messages published after the subscription is
	// registered in the cluster).
	QueueSubscribe(subject, qgroup string, cb MsgHandler, opts ...SubscriptionOption) (Subscription, error)

	// Close a connection to the cluster.
	//
	// If there are active subscriptions at the time of the close, they are implicitly closed
	// (not unsubscribed) by the cluster. This means that durable subscriptions are maintained.
	//
	// The wait on asynchronous publish calls are canceled and ErrConnectionClosed will be
	// reported to the registered AckHandler. It is possible that the cluster received and
	// persisted these messages.
	//
	// If a NATS connection is provided as an option to the Connect() call, the NATS
	// connection is NOT closed when this call is invoked. This connection needs to be
	// managed by the application.
	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. The connection is safe to use in multiple Go routines concurrently.

func Connect

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

Connect will form a connection to the NATS Streaming subsystem. Note that clientID can contain only alphanumeric and `-` or `_` characters. The default options are those returned by GetDefaultOptions and each option specified in a parameter here overrides those defaults.

type ConnectionLostHandler added in v0.4.0

type ConnectionLostHandler func(Conn, error)

ConnectionLostHandler is used to be notified if the Streaming connection is closed due to unexpected errors.

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 AllowCloseRetry added in v0.10.2

func AllowCloseRetry(allow bool) Option

AllowCloseRetry is an Option that allows a failed connection close to be retried. See option AllowCloseRetry for more information.

func ConnectWait

func ConnectWait(t time.Duration) Option

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

func MaxPubAcksInflight added in v0.2.0

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 streaming connection object. When such option is set, closing the streaming connection does not close the provided NATS connection.

DEPRECATED: Users should use NatsOptions instead to configure the underlying NATS Connection created by the Streaming connection.

func NatsOptions added in v0.10.0

func NatsOptions(opts ...nats.Option) Option

NatsOptions is an Option to provide the NATS options that will be used to create the underlying NATS connection to be used by a streaming connection object.

func NatsURL added in v0.2.0

func NatsURL(u string) Option

NatsURL is an Option to set the URL the client should connect to. The url can contain username/password semantics. e.g. nats://derek:pass@localhost:4222 Comma separated arrays are also supported, e.g. urlA, urlB.

func Pings added in v0.4.0

func Pings(interval, maxOut int) Option

Pings is an Option to set the ping interval and max out values. The interval needs to be at least 1 and represents the number of seconds. The maxOut needs to be at least 2, since the count of sent PINGs increase whenever a PING is sent and reset to 0 when a response is received. Setting to 1 would cause the library to close the connection right away.

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.

func SetConnectionLostHandler added in v0.4.0

func SetConnectionLostHandler(handler ConnectionLostHandler) Option

SetConnectionLostHandler is an Option to set the connection lost handler. This callback will be invoked should the client permanently lose contact with the server (or another client replaces it while being disconnected). The callback will not be invoked on normal Conn.Close().

type Options

type Options struct {
	// NatsURL is an URL (or comma separated list of URLs) to a node or nodes
	// in the cluster.
	NatsURL string

	// NatsConn is a user provided low-level NATS connection that the streaming
	// connection will use to communicate with the cluster. When set, closing
	// the NATS streaming connection does NOT close this NATS connection.
	// It is the responsibility of the application to manage the lifetime of
	// the supplied NATS connection.
	//
	// DEPRECATED: Users should provide NATS options through NatsOptions()
	// instead to configure the underlying NATS connection.
	NatsConn *nats.Conn

	// NatsOptions is an array of NATS options to configure the NATS connection
	// that will be created and owned by the library. Note that some options
	// may be overridden by the library.
	NatsOptions []nats.Option

	// ConnectTimeout is the timeout for the initial Connect(). This value is also
	// used for some of the internal request/replies with the cluster.
	ConnectTimeout time.Duration

	// AckTimeout is how long to wait when a message is published for an ACK from
	// the cluster. If the library does not receive an ACK after this timeout,
	// the Publish() call (or the AckHandler) will return ErrTimeout.
	AckTimeout time.Duration

	// DiscoverPrefix is the prefix connect requests are sent to for this cluster.
	// The default is "_STAN.discover".
	DiscoverPrefix string

	// MaxPubAcksInflight specifies how many messages can be published without
	// getting ACKs back from the cluster before the Publish() or PublishAsync()
	// calls block.
	MaxPubAcksInflight int

	// DEPRECATED: Please use PingInterval instead
	PingIterval int

	// PingInterval is the interval at which client sends PINGs to the server
	// to detect the loss of a connection.
	PingInterval int

	// PingMaxOut specifies the maximum number of PINGs without a corresponding
	// PONG before declaring the connection permanently lost.
	PingMaxOut int

	// ConnectionLostCB specifies the handler to be invoked when the connection
	// is permanently lost.
	ConnectionLostCB ConnectionLostHandler

	// AllowCloseRetry specifies that a failed connection Close() can be retried.
	//
	// By default, after the first call to Close(), the underlying NATS connection
	// is closed (when owned by the library), regardless if the library gets a
	// response from the server or not, and calling Close() again is a no-op.
	// With AllowCloseRetry set to true, if the library fails to get a response
	// from the close protocol, calling Close() again is possible and the library
	// will try to resend the protocol. It means that the underlying NATS connection
	// won't be closed until the library successfully gets a response from the server.
	// This behavior can have side effects in that the underlying NATS connection
	// may stay open (or reconnect) when otherwise it would have been closed after
	// calling Close(). So AllowCloseRetry is disabled by default to maintain
	// expected default behavior in regard with the underlying NATS connection state.
	AllowCloseRetry bool
}

Options can be used to a create a customized connection.

func GetDefaultOptions deprecated added in v0.5.0

func GetDefaultOptions() Options

GetDefaultOptions returns default configuration options for the client.

Deprecated: do not use this from outside the library. This is intended to be used within the library; its visibility to clients is a design wart which we can't remove without breaking API compatibility. Instead, the Connect() function accepts Option parameters, each singular, which are the exposed interface.

type Subscription

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

	// ClearMaxPending resets the maximums seen so far.
	ClearMaxPending() error

	// Delivered returns the number of delivered messages for the internal low-level NATS subscription.
	Delivered() (int64, error)

	// Dropped returns the number of known dropped messages for the internal low-level NATS subscription.
	// This will correspond to messages dropped by violations of PendingLimits. If the server declares
	// the connection a SlowConsumer, this number may not be valid.
	Dropped() (int, error)

	// IsValid returns a boolean indicating whether the internal low-level NATS subscription is still active.
	// This will return false if the subscription has already been closed.
	IsValid() bool

	// MaxPending returns the maximum number of queued messages and queued bytes seen so far for the internal
	// low-level NATS subscription.
	MaxPending() (int, int, error)

	// Pending returns the number of queued messages and queued bytes in the client for the internal
	// low-level NATS subscription.
	Pending() (int, int, error)

	// PendingLimits returns the current limits for the internal low-level NATS subscription. If no error is
	// returned, a negative value indicates that the given metric is not limited.
	PendingLimits() (int, int, error)

	// SetPendingLimits sets the limits for pending msgs and bytes for the internal low-level NATS Subscription.
	// Zero is not allowed. Any negative value means that the given metric is not limited.
	SetPendingLimits(msgLimit, bytesLimit int) error
}

Subscription represents a subscription within the NATS Streaming cluster. Subscriptions will be rate matched and follow at-least once delivery semantics. The subscription is safe to use in multiple Go routines concurrently.

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 subscriber.

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

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.

Source Files

Directories

Path Synopsis
examples

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL