natsio

Documentation

Index

• type AuthOptions
• type Client
  • func (c *Client) Close()
  • func (c *Client) InitConn(stopCh <-chan struct{}) error
  • func (c *Client) Publish(msg *Msg) error
  • func (c *Client) QueueSubscribe(subj string, queue string, handler nats.MsgHandler) error
  • func (c *Client) Request(msg *Msg, timeoutHandler TimeoutHandler) ([]byte, error)
  • func (c *Client) RequestAsync(msg *Msg, handler ReplyHandler, timeoutHandler TimeoutHandler) error
  • func (c *Client) RequestWithContext(ctx context.Context, msg *Msg) ([]byte, error)
  • func (c *Client) RunServer(stopCh <-chan struct{}) error
  • func (c *Client) SetClosedHandler(handler nats.ConnHandler)
  • func (c *Client) SetDisconnectErrHandler(handler nats.ConnErrHandler)
  • func (c *Client) SetErrorHandler(handler nats.ErrHandler)
  • func (c *Client) SetReconnectHandler(handler nats.ConnHandler)
  • func (c *Client) Subscribe(subj string, handler nats.MsgHandler) error
• type ClientOptions
• type ClusterOptions
• type Interface
  • func NewNats(opts *NatsOptions) Interface
• type Msg
• type NatsOptions
  • func NewOptions() *NatsOptions
  • func (s *NatsOptions) AddFlags(fs *pflag.FlagSet)
  • func (s *NatsOptions) CombineServerAddress() string
  • func (s *NatsOptions) CombineUsernameAndPassword() string
  • func (s *NatsOptions) GetConnectionString() string
  • func (s *NatsOptions) Validate() []error
• type ReplyHandler
• type ServerOptions
• type TimeoutHandler

Types

type AuthOptions

type AuthOptions struct {
	UserName string `yaml:"username" json:"user_name"`
	Password string `yaml:"password" json:"password"`
}

type Client

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

func (*Client) Close

func (c *Client) Close()

func (*Client) InitConn

func (c *Client) InitConn(stopCh <-chan struct{}) error

func (*Client) Publish

func (c *Client) Publish(msg *Msg) error

func (*Client) QueueSubscribe

func (c *Client) QueueSubscribe(subj string, queue string, handler nats.MsgHandler) error

func (*Client) Request

func (c *Client) Request(msg *Msg, timeoutHandler TimeoutHandler) ([]byte, error)

func (*Client) RequestAsync

func (c *Client) RequestAsync(msg *Msg, handler ReplyHandler, timeoutHandler TimeoutHandler) error

func (*Client) RequestWithContext

func (c *Client) RequestWithContext(ctx context.Context, msg *Msg) ([]byte, error)

func (*Client) RunServer

func (c *Client) RunServer(stopCh <-chan struct{}) error

func (*Client) SetClosedHandler

func (c *Client) SetClosedHandler(handler nats.ConnHandler)

func (*Client) SetDisconnectErrHandler

func (c *Client) SetDisconnectErrHandler(handler nats.ConnErrHandler)

func (*Client) SetErrorHandler

func (c *Client) SetErrorHandler(handler nats.ErrHandler)

func (*Client) SetReconnectHandler

func (c *Client) SetReconnectHandler(handler nats.ConnHandler)

func (*Client) Subscribe

func (c *Client) Subscribe(subj string, handler nats.MsgHandler) error

type ClientOptions

type ClientOptions struct {
	ServerAddress     []string `yaml:"serverAddress" json:"serverAddress"`
	SubjectSuffix     string   `yaml:"subjectSuffix" json:"subjectSuffix"`
	QueueGroupName    string   `yaml:"queueGroupName" json:"queueGroupName"`
	NodeReportSubject string   `yaml:"nodeReportSubject" json:"nodeReportSubject"`
	TimeOutSeconds    int      `yaml:"timeOutSeconds" json:"timeOutSeconds"`
	// ReconnectWait sets the time to backoff after attempting a reconnect
	// to a server that we were already connected to previously.
	ReconnectInterval time.Duration `yaml:"reconnectInterval" json:"reconnectInterval"`
	// MaxReconnect sets the number of reconnect attempts that will be
	// tried before giving up. If negative, then it will never give up
	// trying to reconnect.
	MaxReconnect int `yaml:"maxReconnect" json:"maxReconnect"`
	// PingInterval is the period at which the client will be sending ping
	// commands to the server, disabled if 0 or negative.
	PingInterval time.Duration `yaml:"pingInterval" json:"pingInterval"`
	// MaxPingsOut is the maximum number of pending ping commands that can
	// be awaiting a response before raising an ErrStaleConnection error.
	MaxPingsOut int    `yaml:"maxPingsOut" json:"maxPingsOut"`
	TLSCaPath   string `yaml:"tlsCaPath" json:"tlsCaPath"`
	TLSCertPath string `yaml:"tlsCertPath" json:"tlsCertPath"`
	TLSKeyPath  string `yaml:"tlsKeyPath" json:"tlsKeyPath"`
}

type ClusterOptions

type ClusterOptions struct {
	Host       string `yaml:"host" json:"host"`
	Port       int    `yaml:"port" json:"port"`
	LeaderHost string `yaml:"leaderHost" json:"leader_host"`
}

type Interface

type Interface interface {
	SetDisconnectErrHandler(handler nats.ConnErrHandler)
	SetReconnectHandler(handler nats.ConnHandler)
	SetErrorHandler(handler nats.ErrHandler)
	SetClosedHandler(handler nats.ConnHandler)
	RunServer(stopCh <-chan struct{}) error
	InitConn(stopCh <-chan struct{}) error
	Publish(msg *Msg) error
	Subscribe(subj string, handler nats.MsgHandler) error
	QueueSubscribe(subj string, queue string, handler nats.MsgHandler) error
	Request(msg *Msg, timeoutHandler TimeoutHandler) ([]byte, error)
	RequestWithContext(ctx context.Context, msg *Msg) ([]byte, error)
	RequestAsync(msg *Msg, handler ReplyHandler, timeoutHandler TimeoutHandler) error
	Close()
}

func NewNats

func NewNats(opts *NatsOptions) Interface

type Msg

type Msg struct {
	Subject string
	From    string
	To      string
	Step    string
	Timeout time.Duration
	Data    []byte
}

type NatsOptions

type NatsOptions struct {
	External bool          `yaml:"external" json:"external" mapstructure:"external"`
	Client   ClientOptions `yaml:"client" json:"client" mapstructure:"client"`
	Server   ServerOptions `yaml:"server" json:"server" mapstructure:"server"`
	Auth     AuthOptions   `yaml:"auth" json:"auth" mapstructure:"auth"`
}

func NewOptions

func NewOptions() *NatsOptions

func (*NatsOptions) AddFlags

func (s *NatsOptions) AddFlags(fs *pflag.FlagSet)

func (*NatsOptions) CombineServerAddress

func (s *NatsOptions) CombineServerAddress() string

func (*NatsOptions) CombineUsernameAndPassword

func (s *NatsOptions) CombineUsernameAndPassword() string

func (*NatsOptions) GetConnectionString

func (s *NatsOptions) GetConnectionString() string

func (*NatsOptions) Validate

func (s *NatsOptions) Validate() []error

type ReplyHandler

type ReplyHandler func(msg *nats.Msg) error

type ServerOptions

type ServerOptions struct {
	Host        string         `yaml:"host" json:"host"`
	Port        int            `yaml:"port" json:"port"`
	Cluster     ClusterOptions `yaml:"cluster" json:"cluster" mapstructure:"cluster"`
	TLSCaPath   string         `yaml:"tlsCaPath" json:"tlsCaPath"`
	TLSCertPath string         `yaml:"tlsCertPath" json:"tlsCertPath"`
	TLSKeyPath  string         `yaml:"tlsKeyPath" json:"tlsKeyPath"`
}

type TimeoutHandler

type TimeoutHandler func(msg *Msg) error