cfg

Documentation

Index

• type Config
  • func (e *Config) Init(handler func(srv *grpc.Server))
  • func (e *Config) OnChange()
• type Memory
  • func (e *Memory) GetCache() (storage.AdapterCache, error)
  • func (e *Memory) GetLocker() (storage.AdapterLocker, error)
  • func (e *Memory) GetQueue() (storage.AdapterQueue, error)
  • func (Memory) String() string
• type NSQ
  • func (e *NSQ) GetCache() (storage.AdapterCache, error)
  • func (e *NSQ) GetLocker() (storage.AdapterLocker, error)
  • func (e *NSQ) GetQueue() (storage.AdapterQueue, error)
  • func (NSQ) String() string
• type NSQOptions
  • func (e NSQOptions) GetNSQOptions() (*nsq.Config, error)
• type Provider
  • func (e Provider) Init(cache, queue, locker string)
• type Redis
  • func (e *Redis) GetCache() (storage.AdapterCache, error)
  • func (e *Redis) GetLocker() (storage.AdapterLocker, error)
  • func (e *Redis) GetQueue() (storage.AdapterQueue, error)
  • func (e Redis) GetRedisOptions() (*redis.Options, error)
  • func (Redis) String() string

Types

type Config

type Config struct {
	Logger   config.Logger  `yaml:"logger" json:"logger"`
	Server   config.GRPC    `yaml:"server" json:"server"`
	Health   *config.Listen `yaml:"health" json:"health"`
	Metrics  *config.Listen `yaml:"metrics" json:"metrics"`
	Cache    string         `yaml:"cache" json:"cache"`
	Queue    string         `yaml:"queue" json:"queue"`
	Locker   string         `yaml:"locker" json:"locker"`
	Provider Provider       `yaml:"provider" json:"provider"`
}

Config config

var Cfg Config

func (*Config) Init

func (e *Config) Init(handler func(srv *grpc.Server))

Init init

func (*Config) OnChange

func (e *Config) OnChange()

type Memory

type Memory struct {
	PoolSize uint `yaml:"poolSize" json:"poolSize"`
}

func (*Memory) GetCache

func (e *Memory) GetCache() (storage.AdapterCache, error)

func (*Memory) GetLocker

func (e *Memory) GetLocker() (storage.AdapterLocker, error)

func (*Memory) GetQueue

func (e *Memory) GetQueue() (storage.AdapterQueue, error)

func (Memory) String

func (Memory) String() string

String string

type NSQ

type NSQ struct {
	NSQOptions
	ChannelPrefix string
}

func (*NSQ) GetCache

func (e *NSQ) GetCache() (storage.AdapterCache, error)

func (*NSQ) GetLocker

func (e *NSQ) GetLocker() (storage.AdapterLocker, error)

func (*NSQ) GetQueue

func (e *NSQ) GetQueue() (storage.AdapterQueue, error)

func (NSQ) String

func (NSQ) String() string

type NSQOptions

type NSQOptions struct {
	DialTimeout time.Duration `opt:"dial_timeout" default:"1s"`

	// Deadlines for network reads and writes
	ReadTimeout  time.Duration `opt:"read_timeout" min:"100ms" max:"5m" default:"60s"`
	WriteTimeout time.Duration `opt:"write_timeout" min:"100ms" max:"5m" default:"1s"`

	// Addresses is the local address to use when dialing an nsqd.
	Addresses []string `opt:"addresses"`

	// Duration between polling lookupd for new producers, and fractional jitter to add to
	// the lookupd pool loop. this helps evenly distribute requests even if multiple consumers
	// restart at the same time
	//
	// NOTE: when not using nsqlookupd, LookupdPollInterval represents the duration of time between
	// reconnection attempts
	LookupdPollInterval time.Duration `opt:"lookupd_poll_interval" min:"10ms" max:"5m" default:"60s"`
	LookupdPollJitter   float64       `opt:"lookupd_poll_jitter" min:"0" max:"1" default:"0.3"`

	// Maximum duration when REQueueing (for doubling of deferred requeue)
	MaxRequeueDelay     time.Duration `opt:"max_requeue_delay" min:"0" max:"60m" default:"15m"`
	DefaultRequeueDelay time.Duration `opt:"default_requeue_delay" min:"0" max:"60m" default:"90s"`

	// Maximum amount of time to backoff when processing fails 0 == no backoff
	MaxBackoffDuration time.Duration `opt:"max_backoff_duration" min:"0" max:"60m" default:"2m"`
	// Unit of time for calculating consumer backoff
	BackoffMultiplier time.Duration `opt:"backoff_multiplier" min:"0" max:"60m" default:"1s"`

	// Maximum number of times this consumer will attempt to process a message before giving up
	MaxAttempts uint16 `opt:"max_attempts" min:"0" max:"65535" default:"5"`

	// Duration to wait for a message from an nsqd when in a state where RDY
	// counts are re-distributed (e.g. max_in_flight < num_producers)
	LowRdyIdleTimeout time.Duration `opt:"low_rdy_idle_timeout" min:"1s" max:"5m" default:"10s"`
	// Duration to wait until redistributing RDY for an nsqd regardless of LowRdyIdleTimeout
	LowRdyTimeout time.Duration `opt:"low_rdy_timeout" min:"1s" max:"5m" default:"30s"`
	// Duration between redistributing max-in-flight to connections
	RDYRedistributeInterval time.Duration `opt:"rdy_redistribute_interval" min:"1ms" max:"5s" default:"5s"`

	// Identifiers sent to nsqd representing this client
	// UserAgent is in the spirit of HTTP (default: "<client_library_name>/<version>")
	ClientID  string `opt:"client_id"` // (defaults: short hostname)
	Hostname  string `opt:"hostname"`
	UserAgent string `opt:"user_agent"`

	// Duration of time between heartbeats. This must be less than ReadTimeout
	HeartbeatInterval time.Duration `opt:"heartbeat_interval" default:"30s"`
	// Integer percentage to sample the channel (requires nsqd 0.2.25+)
	SampleRate int32 `opt:"sample_rate" min:"0" max:"99"`

	Tls *config.Tls `yaml:"tls" json:"tls"`

	// Compression Settings
	Deflate      bool `opt:"deflate"`
	DeflateLevel int  `opt:"deflate_level" min:"1" max:"9" default:"6"`
	Snappy       bool `opt:"snappy"`

	// Size of the buffer (in bytes) used by nsqd for buffering writes to this connection
	OutputBufferSize int64 `opt:"output_buffer_size" default:"16384"`
	// Timeout used by nsqd before flushing buffered writes (set to 0 to disable).
	//
	// WARNING: configuring clients with an extremely low
	// (< 25ms) output_buffer_timeout has a significant effect
	// on nsqd CPU usage (particularly with > 50 clients connected).
	OutputBufferTimeout time.Duration `opt:"output_buffer_timeout" default:"250ms"`

	// Maximum number of messages to allow in flight (concurrency knob)
	MaxInFlight int `opt:"max_in_flight" min:"0" default:"1"`

	// The server-side message timeout for messages delivered to this client
	MsgTimeout time.Duration `opt:"msg_timeout" min:"0"`

	// secret for nsqd authentication (requires nsqd 0.2.29+)
	AuthSecret string `opt:"auth_secret"`
}

func (NSQOptions) GetNSQOptions

func (e NSQOptions) GetNSQOptions() (*nsq.Config, error)

type Provider

type Provider struct {
	Memory *Memory `yaml:"memory" json:"memory"`
	Redis  *Redis  `yaml:"redis" json:"redis"`
	NSQ    *NSQ    `yaml:"nsq" json:"nsq"`
}

Provider provider

func (Provider) Init

func (e Provider) Init(cache, queue, locker string)

type Redis

type Redis struct {
	Network    string                      `yaml:"network" json:"network"`
	Addr       string                      `yaml:"addr" json:"addr"`
	Username   string                      `yaml:"username" json:"username"`
	Password   string                      `yaml:"password" json:"password"`
	DB         int                         `yaml:"db" json:"db"`
	PoolSize   int                         `yaml:"pool_size" json:"pool_size"`
	Tls        *config.Tls                 `yaml:"tls" json:"tls"`
	MaxRetries int                         `yaml:"max_retries" json:"max_retries"`
	Producer   *redisqueue.ProducerOptions `yaml:"producer" json:"producer"`
	Consumer   *redisqueue.ConsumerOptions `yaml:"consumer" json:"consumer"`
}

func (*Redis) GetCache

func (e *Redis) GetCache() (storage.AdapterCache, error)

func (*Redis) GetLocker

func (e *Redis) GetLocker() (storage.AdapterLocker, error)

func (*Redis) GetQueue

func (e *Redis) GetQueue() (storage.AdapterQueue, error)

func (Redis) GetRedisOptions

func (e Redis) GetRedisOptions() (*redis.Options, error)

func (Redis) String

func (Redis) String() string