stan: github.com/nats-io/stan Index | Files | Directories

package stan

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

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

Package Files

stan.go sub.go

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
    // 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 = 3
)
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.5.0"

Version is the NATS Streaming Go Client version

Variables

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

var DefaultOptions = GetDefaultOptions()

DEPRECATED: Use GetDefaultOptions() instead. 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

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

DefaultSubscriptionOptions are the default subscriptions' options

type AckHandler Uses

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 Uses

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.

func Connect Uses

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.

type ConnectionLostHandler Uses

type ConnectionLostHandler func(Conn, error)

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

type Msg Uses

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 Uses

func (msg *Msg) Ack() error

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

type MsgHandler Uses

type MsgHandler func(msg *Msg)

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

type Option Uses

type Option func(*Options) error

Option is a function on the options for a connection.

func ConnectWait Uses

func ConnectWait(t time.Duration) Option

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

func MaxPubAcksInflight Uses

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 Uses

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.

func NatsURL Uses

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 Uses

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 Uses

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 Uses

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 Uses

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.
    NatsConn *nats.Conn

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

Options can be used to a create a customized connection.

func GetDefaultOptions Uses

func GetDefaultOptions() Options

GetDefaultOptions returns default configuration options for the client.

type Subscription Uses

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.

type SubscriptionOption Uses

type SubscriptionOption func(*SubscriptionOptions) error

SubscriptionOption is a function on the options for a subscription.

func AckWait Uses

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 Uses

func DeliverAllAvailable() SubscriptionOption

DeliverAllAvailable will deliver all messages available.

func DurableName Uses

func DurableName(name string) SubscriptionOption

DurableName sets the DurableName for the subscriber.

func MaxInflight Uses

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 Uses

func SetManualAckMode() SubscriptionOption

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

func StartAt Uses

func StartAt(sp pb.StartPosition) SubscriptionOption

StartAt sets the desired start position for the message stream.

func StartAtSequence Uses

func StartAtSequence(seq uint64) SubscriptionOption

StartAtSequence sets the desired start sequence position and state.

func StartAtTime Uses

func StartAtTime(start time.Time) SubscriptionOption

StartAtTime sets the desired start time position and state.

func StartAtTimeDelta Uses

func StartAtTimeDelta(ago time.Duration) SubscriptionOption

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

func StartWithLastReceived Uses

func StartWithLastReceived() SubscriptionOption

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

type SubscriptionOptions Uses

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.

Directories

PathSynopsis
pbPackage pb is a generated protocol buffer package.

Package stan imports 7 packages (graph). Updated 2019-11-15. Refresh now. Tools for package owners.