nats

README

NATS Module for Helium

Module provides you with the following things:

• *nats.Conn represents a bare connection to a nats-server. It can send and receive []byte payloads
• stan.Conn represents a connection to the NATS Streaming subsystem. It can Publish and Subscribe to messages within the NATS Streaming cluster.

Configuration:

• yaml example

nats:
  url: nats://<host>:<port>
  cluster_id: string
  client_id: string
  servers: [...server slice...]
  no_randomize: bool
  name: string
  verbose: bool
  pedantic: bool
  secure: bool
  allow_reconnect: bool
  max_reconnect: int
  reconnect_wait: duration
  timeout: duration
  flusher_timeout: duration
  ping_interval: duration
  max_pings_out: int
  reconnect_buf_size: int
  sub_chan_len: int
  user: string
  password: string
  token: string
  stan:
    connect_wait: duration
    pub_ack_wait: duration
    max_pub_acks_inflight: int
    ping_max_out: int
    ping_interval: int

• env example

NATS_URL=nats://<host>:<port>
NATS_CLUSTER_ID=string
NATS_CLIENT_ID=string
NATS_SERVERS=[...server slice...]
NATS_NO_RANDOMIZE=bool
NATS_NAME=string
NATS_VERBOSE=bool
NATS_PEDANTIC=bool
NATS_SECURE=bool
NATS_ALLOW_RECONNECT=bool
NATS_MAX_RECONNECT=int
NATS_RECONNECT_WAIT=duration
NATS_TIMEOUT=duration
NATS_FLUSHER_TIMEOUT=duration
NATS_PING_INTERVAL=duration
NATS_MAX_PINGS_OUT=int
NATS_RECONNECT_BUF_SIZE=int
NATS_SUB_CHAN_LEN=int
NATS_USER=string
NATS_PASSWORD=string
NATS_TOKEN=string
NATS_STAN_CONNECT_WAIT=duration
NATS_STAN_PUB_ACK_WAIT=duration
NATS_STAN_MAX_PUB_ACKS_INFLIGHT=int
NATS_STAN_PING_MAX_OUT=int
NATS_STAN_PING_INTERVAL=int

Documentation

Index

• Constants
• Variables
• func NewStreamer(opts *StreamerConfig) (stan.Conn, error)
• type Client
  • func NewConnection(opts *Config) (bus *Client, err error)
• type Config
  • func NewDefaultConfig(v *viper.Viper) (*Config, error)
• type Error
  • func (e Error) Error() string
• type StreamerConfig
  • func NewDefaultStreamerConfig(p StreamerParams) (*StreamerConfig, error)
• type StreamerParams

Constants

const (
	// ErrEmptyConfig when given empty options
	ErrEmptyConfig = Error("nats empty config")
	// ErrEmptyStreamerConfig when given empty options
	ErrEmptyStreamerConfig = Error("nats-streamer empty config")
	// ErrEmptyConnection when empty nats.Conn
	ErrEmptyConnection = Error("nats connection empty")
	// ErrClusterIDEmpty when empty clusterID
	ErrClusterIDEmpty = Error("nats.cluster_id cannot be empty")
	// ErrClientIDEmpty when empty clientID
	ErrClientIDEmpty = Error("nats.client_id cannot be empty")
)

Variables

var (
	// Module is default Nats client
	Module = module.Module{
		{Constructor: NewDefaultConfig},
		{Constructor: NewConnection},
		{Constructor: NewDefaultStreamerConfig},
		{Constructor: NewStreamer},
	}
)

Functions

func NewStreamer

func NewStreamer(opts *StreamerConfig) (stan.Conn, error)

NewStreamer is nats-streamer client

Types

type Client

type Client = nats.Conn

Client alias

func NewConnection

func NewConnection(opts *Config) (bus *Client, err error)

NewConnection of nats client

type Config

type Config = nats.Options

Config alias

func NewDefaultConfig

func NewDefaultConfig(v *viper.Viper) (*Config, error)

NewDefaultConfig default settings for connection

type Error

type Error string

Error is constant error

func (Error) Error

func (e Error) Error() string

Error returns error message string

type StreamerConfig

type StreamerConfig struct {
	ClientID  string
	ClusterID string
	Options   []stan.Option
}

StreamerConfig for NSS client

func NewDefaultStreamerConfig

func NewDefaultStreamerConfig(p StreamerParams) (*StreamerConfig, error)

NewDefaultStreamerConfig default settings for streaming connection

type StreamerParams

type StreamerParams struct {
	dig.In

	Bus              *Client
	Viper            *viper.Viper
	OnConnectionLost stan.ConnectionLostHandler `optional:"true"`
}

StreamerParams param to initialize streamer client instance