- 首先定义
Options
struct
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
}
- 定义
Option
类型的字段
type Option func(*Options) error
- 定义
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 NatsURL(u string) Option {
return func(o *Options) error {
o.NatsURL = u
return nil
}
}
// ConnectWait is an Option to set the timeout for establishing a connection.
func ConnectWait(t time.Duration) Option {
return func(o *Options) error {
o.ConnectTimeout = t
return nil
}
}
// PubAckWait is an Option to set the timeout for waiting for an ACK for a
// published message.
func PubAckWait(t time.Duration) Option {
return func(o *Options) error {
o.AckTimeout = t
return nil
}
}
// MaxPubAcksInflight is an Option to set the maximum number of published
// messages without outstanding ACKs from the server.
func MaxPubAcksInflight(max int) Option {
return func(o *Options) error {
o.MaxPubAcksInflight = max
return nil
}
}
// 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 NatsConn(nc *nats.Conn) Option {
return func(o *Options) error {
o.NatsConn = nc
return nil
}
}
// 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 Pings(interval, maxOut int) Option {
return func(o *Options) error {
// For tests, we may pass negative value that will be interpreted
// by the library as milliseconds. If this test boolean is set,
// do not check values.
if !testAllowMillisecInPings {
if interval < 1 || maxOut <= 2 {
return fmt.Errorf("invalid ping values: interval=%v (min>0) maxOut=%v (min=2)", interval, maxOut)
}
}
o.PingInterval = interval
o.PingMaxOut = maxOut
return nil
}
}
// 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().
func SetConnectionLostHandler(handler ConnectionLostHandler) Option {
return func(o *Options) error {
o.ConnectionLostCB = handler
return nil
}
}
- 方法接收
func Connect(stanClusterID, clientID string, options ...Option) (Conn, error) {
...
}
这样,通过options
接收的值,遍历调用,验证值得正确性。