stan

README

nats

Documentation

Index

• Variables
• type Client
  • func NewClient(config *Config, tracer opentracing.Tracer, opts ...Option) (*Client, error)
  • func (c *Client) Close() error
  • func (c *Client) Publish(ctx context.Context, topic string, data []byte) error
  • func (c *Client) QueueSubscribe(topic, queue string, handler Handler, opts ...SubOption) error
  • func (c *Client) Subscribe(topic string, handler Handler, opts ...SubOption) error
  • func (c *Client) Unsubscribe(topic string) error
  • func (c *Client) UnsubscribeAll() error
• type Config
• type ConnectionLostHandler
• type Handler
• type Message
• type Option
  • func ConnectWait(t time.Duration) Option
  • func MaxPubAcksInflight(max int) Option
  • func Pings(interval, maxOut int) Option
  • func PubAckWait(t time.Duration) Option
  • func SetConnectionLostHandler(handler ConnectionLostHandler) Option
• type StartPosition
• type SubOption
  • func AckWait(t time.Duration) SubOption
  • func DeliverAllAvailable() SubOption
  • func DurableName(name string) SubOption
  • func MaxInflight(m int) SubOption
  • func SetManualAckMode() SubOption
  • func StartAt(sp StartPosition) SubOption
  • func StartAtSequence(seq uint64) SubOption
  • func StartAtTime(start time.Time) SubOption
  • func StartAtTimeDelta(ago time.Duration) SubOption
  • func StartWithLastReceived() SubOption

Variables

var ErrUnsubscribe = errors.New("unsubscribe")

Types

type Client

type Client struct {
	// contains filtered or unexported fields
}

func NewClient

func NewClient(config *Config, tracer opentracing.Tracer, opts ...Option) (*Client, error)

func (*Client) Close

func (c *Client) Close() error

func (*Client) Publish

func (c *Client) Publish(ctx context.Context, topic string, data []byte) error

func (*Client) QueueSubscribe

func (c *Client) QueueSubscribe(topic, queue string, handler Handler, opts ...SubOption) error

func (*Client) Subscribe

func (c *Client) Subscribe(topic string, handler Handler, opts ...SubOption) error

func (*Client) Unsubscribe

func (c *Client) Unsubscribe(topic string) error

func (*Client) UnsubscribeAll

func (c *Client) UnsubscribeAll() error

type Config

type Config struct {
	Addr      string
	ClientID  string
	ClusterID string
}

type ConnectionLostHandler

type ConnectionLostHandler = stan.ConnectionLostHandler

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

type Handler

type Handler func(ctx context.Context, msg *Message) error

Handler to handle sub message.

type Message

type Message = stan.Msg

Stan message is stan.Msg.

type Option

type Option = stan.Option

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 Pings

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

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 StartPosition

type StartPosition = pb.StartPosition

Enum for start position type.

const (
	StartPositionNewOnly        StartPosition = 0
	StartPositionLastReceived   StartPosition = 1
	StartPositionTimeDeltaStart StartPosition = 2
	StartPositionSequenceStart  StartPosition = 3
	StartPositionFirst          StartPosition = 4
)

type SubOption

type SubOption = stan.SubscriptionOption

SubOption is a function on the options for a subscription.

func AckWait

func AckWait(t time.Duration) SubOption

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() SubOption

DeliverAllAvailable will deliver all messages available.

func DurableName

func DurableName(name string) SubOption

DurableName sets the DurableName for the subscriber.

func MaxInflight

func MaxInflight(m int) SubOption

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

func SetManualAckMode

func SetManualAckMode() SubOption

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

func StartAt

func StartAt(sp StartPosition) SubOption

StartAt sets the desired start position for the message stream.

func StartAtSequence

func StartAtSequence(seq uint64) SubOption

StartAtSequence sets the desired start sequence position and state.

func StartAtTime

func StartAtTime(start time.Time) SubOption

StartAtTime sets the desired start time position and state.

func StartAtTimeDelta

func StartAtTimeDelta(ago time.Duration) SubOption

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

func StartWithLastReceived

func StartWithLastReceived() SubOption

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