server

package
v0.25.6 Latest Latest
Warning

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

 Go to latest Published: Nov 17, 2023 License: Apache-2.0 Imports: 41 Imported by: 51

Documentation

Index

Constants

View Source 
const (
	RootPath       = "/streaming"
	ServerPath     = RootPath + "/serverz"
	StorePath      = RootPath + "/storez"
	ClientsPath    = RootPath + "/clientsz"
	ChannelsPath   = RootPath + "/channelsz"
	IsFTActivePath = RootPath + "/isFTActive"
)

Routes for the monitoring pages

View Source 
const (
	// VERSION is the current version for the NATS Streaming server.
	VERSION = "0.25.6"

	DefaultClusterID      = "test-cluster"
	DefaultDiscoverPrefix = "_STAN.discover"
	DefaultPubPrefix      = "_STAN.pub"
	DefaultSubPrefix      = "_STAN.sub"
	DefaultSubClosePrefix = "_STAN.subclose"
	DefaultUnSubPrefix    = "_STAN.unsub"
	DefaultClosePrefix    = "_STAN.close"

	DefaultStoreType = stores.TypeMemory

	// DefaultHeartBeatInterval is the interval at which server sends heartbeat to a client
	DefaultHeartBeatInterval = 30 * time.Second
	// DefaultClientHBTimeout is how long server waits for a heartbeat response
	DefaultClientHBTimeout = 10 * time.Second
	// DefaultMaxFailedHeartBeats is the number of failed heartbeats before server closes
	// the client connection (total= (heartbeat interval + heartbeat timeout) * (fail count + 1)
	DefaultMaxFailedHeartBeats = int((5 * time.Minute) / DefaultHeartBeatInterval)

	// DefaultIOBatchSize is the maximum number of messages to accumulate before flushing a store.
	DefaultIOBatchSize = 1024

	// DefaultIOSleepTime is the duration (in micro-seconds) the server waits for more messages
	// before starting processing. Set to 0 (or negative) to disable the wait.
	DefaultIOSleepTime = int64(0)

	// DefaultLogCacheSize is the number of Raft log entries to cache in memory
	// to reduce disk IO.
	DefaultLogCacheSize = 512

	// DefaultLogSnapshots is the number of Raft log snapshots to retain.
	DefaultLogSnapshots = 2

	// DefaultTrailingLogs is the number of log entries to leave after a
	// snapshot and compaction.
	DefaultTrailingLogs = 10240
)

Server defaults.

Variables

View Source 
var (
	ErrInvalidSubject     = errors.New("stan: invalid subject")
	ErrInvalidStart       = errors.New("stan: invalid start position")
	ErrInvalidSub         = errors.New("stan: invalid subscription")
	ErrInvalidClient      = errors.New("stan: clientID already registered")
	ErrMissingClient      = errors.New("stan: clientID missing")
	ErrInvalidClientID    = errors.New("stan: invalid clientID: only alphanumeric and `-` or `_` characters allowed")
	ErrInvalidAckWait     = errors.New("stan: invalid ack wait time, should be >= 1s")
	ErrInvalidMaxInflight = errors.New("stan: invalid MaxInflight, should be >= 1")
	ErrInvalidConnReq     = errors.New("stan: invalid connection request")
	ErrInvalidPubReq      = errors.New("stan: invalid publish request")
	ErrInvalidSubReq      = errors.New("stan: invalid subscription request")
	ErrInvalidUnsubReq    = errors.New("stan: invalid unsubscribe request")
	ErrInvalidCloseReq    = errors.New("stan: invalid close request")
	ErrDupDurable         = errors.New("stan: duplicate durable registration")
	ErrInvalidDurName     = errors.New("stan: durable name of a durable queue subscriber can't contain the character ':'")
	ErrUnknownClient      = errors.New("stan: unknown clientID")
	ErrUnknownChannel     = errors.New("stan: unknown channel")
	ErrNoChannel          = errors.New("stan: no configured channel")
	ErrClusteredRestart   = errors.New("stan: cannot restart server in clustered mode if it was not previously clustered")
	ErrChanDelInProgress  = errors.New("stan: channel is being deleted")
)

Errors.

View Source 
var DefaultNatsServerOptions = server.Options{
	Host:   "127.0.0.1",
	Port:   4222,
	NoLog:  true,
	NoSigs: true,
}

DefaultNatsServerOptions are default options for the NATS server

Functions

func NewNATSOptions added in v0.9.0 

func NewNATSOptions() *server.Options

NewNATSOptions returns a new instance of (NATS) Options. This is needed if one wants to configure specific NATS options before starting a NATS Streaming Server (with RunServerWithOpts()).

func ProcessConfigFile added in v0.3.0 

func ProcessConfigFile(configFile string, opts *Options) error

ProcessConfigFile parses the configuration file `configFile` and updates the given Streaming options `opts`.

Types

type Channelsz added in v0.5.0 

type Channelsz struct {
	ClusterID string      `json:"cluster_id"`
	ServerID  string      `json:"server_id"`
	Now       time.Time   `json:"now"`
	Offset    int         `json:"offset"`
	Limit     int         `json:"limit"`
	Count     int         `json:"count"`
	Total     int         `json:"total"`
	Names     []string    `json:"names,omitempty"`
	Channels  []*Channelz `json:"channels,omitempty"`
}

Channelsz lists the name of all NATS Streaming Channelsz

type Channelz added in v0.5.0 

type Channelz struct {
	Name          string           `json:"name"`
	Msgs          int              `json:"msgs"`
	Bytes         uint64           `json:"bytes"`
	FirstSeq      uint64           `json:"first_seq"`
	LastSeq       uint64           `json:"last_seq"`
	SubsCount     int              `json:"subs_count"`
	Subscriptions []*Subscriptionz `json:"subscriptions,omitempty"`
}

Channelz describes a NATS Streaming Channel

type Clientsz added in v0.5.0 

type Clientsz struct {
	ClusterID string     `json:"cluster_id"`
	ServerID  string     `json:"server_id"`
	Now       time.Time  `json:"now"`
	Offset    int        `json:"offset"`
	Limit     int        `json:"limit"`
	Count     int        `json:"count"`
	Total     int        `json:"total"`
	Clients   []*Clientz `json:"clients"`
}

Clientsz lists the client connections

type Clientz added in v0.5.0 

type Clientz struct {
	ID            string                      `json:"id"`
	HBInbox       string                      `json:"hb_inbox"`
	SubsCount     int                         `json:"subs_count"`
	Subscriptions map[string][]*Subscriptionz `json:"subscriptions,omitempty"`
}

Clientz describes a NATS Streaming Client connection

type ClusteringOptions added in v0.9.0 

type ClusteringOptions struct {
	Clustered    bool     // Run the server in a clustered configuration.
	NodeID       string   // ID of the node within the cluster.
	Bootstrap    bool     // Bootstrap the cluster as a seed node if there is no existing state.
	Peers        []string // List of cluster peer node IDs to bootstrap cluster state.
	RaftLogPath  string   // Path to Raft log store directory.
	LogCacheSize int      // Number of Raft log entries to cache in memory to reduce disk IO.
	LogSnapshots int      // Number of Raft log snapshots to retain.
	TrailingLogs int64    // Number of logs left after a snapshot.
	Sync         bool     // Do a file sync after every write to the Raft log and message store.
	RaftLogging  bool     // Enable logging of Raft library (disabled by default since really verbose).

	// Enable creation of dedicated NATS connections to communicate with other
	// nodes. Normally, the server has a single NATS connection and subscribes
	// to a subject where other nodes can submit requests to "connect" to it.
	// When a remote connects, a new subscription on an inbox is created on
	// both sides and they use their single "raft" NATS connection to communicate.
	// If node "A" connects to both "B" and "C" it will have two subscriptions
	// and two "outbox" subjects (on per remote node) to which send data to.
	//
	// With this option enabled, NATS connection(s) will be created per remote
	// node. This should help with performance and reduce contention.
	// The RAFT transport is pooling connections, so there may be more than
	// one connection per remote node.
	NodesConnections bool

	// If this is enabled, the leader of the cluster will listen to add/remove
	// requests on NATS subject "_STAN.raft.<cluster ID>.node.[add|remove]".
	// Admin can/should limit permissions to send to this subject to prevent
	// a user to inadvertently change the cluster configuration.
	AllowAddRemoveNode bool

	// When a node processes a snapshot (either on startup or if falling behind) and its is
	// not in phase with the message store's state, it is required to reconcile its state
	// with the current leader. If it is unable, the node will fail to start or exit.
	// If all nodes are starting and there is no way to have a leader at this point,
	// then if this boolean is set to true, then the node will attempt to reconcile but
	// if it can't it will still proceed.
	ProceedOnRestoreFailure bool

	// These will be set to some sane defaults. Change only if experiencing raft issues.
	RaftHeartbeatTimeout time.Duration
	RaftElectionTimeout  time.Duration
	RaftLeaseTimeout     time.Duration
	RaftCommitTimeout    time.Duration

	// These options influence the RAFT store implementation which uses bolt DB.
	//
	// Sync freelist to disk. This reduces the database write performance, but
	// speed up recovery since there is no need for a full database re-sync.
	BoltFreeListSync bool

	// BoltFreeListMap sets the backend freelist type to use a map instead of
	// the default array type.
	// The "array" type (the default) is simple but suffers dramatic performance
	// degradation if database is large and framentation in freelist is common.
	// The "hashmap which is faster in almost all circumstances but doesn't guarantee
	// that it offers the smallest page id available. In normal case it is safe.
	BoltFreeListMap bool
}

ClusteringOptions contains STAN Server options related to clustering.

type Options 

type Options struct {
	ID                 string
	DiscoverPrefix     string
	StoreType          string
	FilestoreDir       string
	FileStoreOpts      stores.FileStoreOptions
	SQLStoreOpts       stores.SQLStoreOptions
	stores.StoreLimits               // Store limits (MaxChannels, etc..)
	EnableLogging      bool          // Enables logging
	CustomLogger       logger.Logger // Server will start with the provided logger
	Trace              bool          // Verbose trace
	Debug              bool          // Debug trace
	HandleSignals      bool          // Should the server setup a signal handler (for Ctrl+C, etc...)
	Secure             bool          // Create a TLS enabled connection
	ClientCert         string        // Client Certificate for TLS
	ClientKey          string        // Client Key for TLS
	ClientCA           string        // Client CAs for TLS
	TLSSkipVerify      bool          // Skips the server's certificate chain and host name verification (Insecure!)
	TLSServerName      string        // Used to verify the hostname returned in the server certificate
	IOBatchSize        int           // Maximum number of messages collected from clients before starting their processing.
	IOSleepTime        int64         // Duration (in micro-seconds) the server waits for more message to fill up a batch.
	NATSServerURL      string        // URL for external NATS Server to connect to. If empty, NATS Server is embedded.
	NATSCredentials    string        // Credentials file for connecting to external NATS Server.
	Username           string        // Username to use if not provided from command line.
	Password           string        // Password to use if not provided from command line.
	Token              string        // Authentication token to use if not provided from command line.
	NKeySeedFile       string        // File name containing NKey private key.
	ClientHBInterval   time.Duration // Interval at which server sends heartbeat to a client.
	ClientHBTimeout    time.Duration // How long server waits for a heartbeat response.
	ClientHBFailCount  int           // Number of failed heartbeats before server closes client connection.
	FTGroupName        string        // Name of the FT Group. A group can be 2 or more servers with a single active server and all sharing the same datastore.
	Partitioning       bool          // Specify if server only accepts messages/subscriptions on channels defined in StoreLimits.
	SyslogName         string        // Optional name for the syslog (usueful on Windows when running several servers as a service)
	Encrypt            bool          // Specify if server should encrypt messages payload when storing them
	EncryptionCipher   string        // Cipher used for encryption. Supported are "AES" and "CHACHA". If none is specified, defaults to AES on platforms with Intel processors, CHACHA otherwise.
	EncryptionKey      []byte        // Encryption key. The environment NATS_STREAMING_ENCRYPTION_KEY takes precedence and is the preferred way to provide the key.
	Clustering         ClusteringOptions
	NATSClientOpts     []nats.Option
	ReplaceDurable     bool // If true, the subscription request for a durable subscription will replace the current durable instead of failing with duplicate durable error.
}

Options for NATS Streaming Server

func ConfigureOptions added in v0.6.0 

func ConfigureOptions(fs *flag.FlagSet, args []string, printVersion, printHelp, printTLSHelp func()) (*Options, *natsd.Options, error)

ConfigureOptions accepts a flag set and augment it with NATS Streaming Server specific flags. It then invokes the corresponding function from NATS Server. On success, Streaming and NATS options structures are returned configured based on the selected flags and/or configuration files. The command line options take precedence to the ones in the configuration files.

func GetDefaultOptions 

func GetDefaultOptions() (o *Options)

GetDefaultOptions returns default options for the NATS Streaming Server

func (*Options) Clone added in v0.6.0 

func (o *Options) Clone() *Options

Clone returns a deep copy of the Options object.

type Serverz added in v0.5.0 

type Serverz struct {
	ClusterID     string    `json:"cluster_id"`
	ServerID      string    `json:"server_id"`
	Version       string    `json:"version"`
	GoVersion     string    `json:"go"`
	State         string    `json:"state"`
	Role          string    `json:"role,omitempty"`
	NodeID        string    `json:"node_id,omitempty"`
	Now           time.Time `json:"now"`
	Start         time.Time `json:"start_time"`
	Uptime        string    `json:"uptime"`
	Clients       int       `json:"clients"`
	Subscriptions int       `json:"subscriptions"`
	Channels      int       `json:"channels"`
	TotalMsgs     int       `json:"total_msgs"`
	TotalBytes    uint64    `json:"total_bytes"`
	InMsgs        int64     `json:"in_msgs"`
	InBytes       int64     `json:"in_bytes"`
	OutMsgs       int64     `json:"out_msgs"`
	OutBytes      int64     `json:"out_bytes"`
	OpenFDs       int       `json:"open_fds,omitempty"`
	MaxFDs        int       `json:"max_fds,omitempty"`
}

Serverz describes the NATS Streaming Server

type StanServer 

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

StanServer structure represents the NATS Streaming Server

func Run added in v0.11.0 

func Run(sOpts *Options, nOpts *natsd.Options) (*StanServer, error)

Run starts the NATS Streaming server. This wrapper function allows Windows to add a hook for running NATS Streaming as a service.

func RunServer 

func RunServer(ID string) (*StanServer, error)

RunServer will startup an embedded NATS Streaming Server and a nats-server to support it.

func RunServerWithOpts 

func RunServerWithOpts(stanOpts *Options, natsOpts *server.Options) (newServer *StanServer, returnedError error)

RunServerWithOpts allows you to run a NATS Streaming Server with full control on the Streaming and NATS Server configuration.

func (*StanServer) ClientURL added in v0.19.0 

func (s *StanServer) ClientURL() string

ClientURL returns the basic URL string representation suitable for a client to use to connect

func (*StanServer) ClusterID 

func (s *StanServer) ClusterID() string

ClusterID returns the NATS Streaming Server's ID.

func (*StanServer) HandleChannelsz added in v0.22.0 

func (s *StanServer) HandleChannelsz(w http.ResponseWriter, r *http.Request)

func (*StanServer) HandleClientsz added in v0.22.0 

func (s *StanServer) HandleClientsz(w http.ResponseWriter, r *http.Request)

func (*StanServer) HandleIsFTActivez added in v0.22.0 

func (s *StanServer) HandleIsFTActivez(w http.ResponseWriter, r *http.Request)

func (*StanServer) HandleRootz added in v0.22.0 

func (s *StanServer) HandleRootz(w http.ResponseWriter, r *http.Request)

func (*StanServer) HandleServerz added in v0.22.0 

func (s *StanServer) HandleServerz(w http.ResponseWriter, r *http.Request)

func (*StanServer) HandleStorez added in v0.22.0 

func (s *StanServer) HandleStorez(w http.ResponseWriter, r *http.Request)

func (*StanServer) LastError added in v0.4.0 

func (s *StanServer) LastError() error

LastError returns the last fatal error the server experienced.

func (*StanServer) Shutdown 

func (s *StanServer) Shutdown()

Shutdown will close our NATS connection and shutdown any embedded NATS server.

func (*StanServer) State added in v0.4.0 

func (s *StanServer) State() State

State returns the state of this server.

type State added in v0.4.0 

type State int8

State represents the possible server states 

const (
	Standalone State = iota
	FTActive
	FTStandby
	Failed
	Shutdown
	Clustered
)

Possible server states

func (State) String added in v0.4.0 

func (state State) String() string

type Storez added in v0.5.0 

type Storez struct {
	ClusterID  string             `json:"cluster_id"`
	ServerID   string             `json:"server_id"`
	Now        time.Time          `json:"now"`
	Type       string             `json:"type"`
	Limits     stores.StoreLimits `json:"limits"`
	TotalMsgs  int                `json:"total_msgs"`
	TotalBytes uint64             `json:"total_bytes"`
}

Storez describes the NATS Streaming Store

type Subscriptionz added in v0.5.0 

type Subscriptionz struct {
	ClientID     string `json:"client_id"`
	Inbox        string `json:"inbox"`
	AckInbox     string `json:"ack_inbox"`
	DurableName  string `json:"durable_name,omitempty"`
	QueueName    string `json:"queue_name,omitempty"`
	IsDurable    bool   `json:"is_durable"`
	IsOffline    bool   `json:"is_offline"`
	MaxInflight  int    `json:"max_inflight"`
	AckWait      int    `json:"ack_wait"`
	LastSent     uint64 `json:"last_sent"`
	PendingCount int    `json:"pending_count"`
	IsStalled    bool   `json:"is_stalled"`
}

Subscriptionz describes a NATS Streaming Subscription

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.