config

Documentation

Index

• type BufferConfiguration
  • func (c *BufferConfiguration) NewOptions(iOpts instrument.Options) buffer.Options
• type ConnectionConfiguration
  • func (c *ConnectionConfiguration) NewOptions(iOpts instrument.Options) writer.ConnectionOptions
• type ProducerConfiguration
  • func (c *ProducerConfiguration) NewProducer(cs client.Client, iOpts instrument.Options, rwOpts xio.Options) (producer.Producer, error)
• type StaticMessageRetryConfiguration
• type WriterConfiguration
  • func (c *WriterConfiguration) NewOptions(cs client.Client, iOpts instrument.Options, rwOptions xio.Options) (writer.Options, error)

Types

type BufferConfiguration

type BufferConfiguration struct {
	OnFullStrategy        *buffer.OnFullStrategy `yaml:"onFullStrategy"`
	MaxBufferSize         *int                   `yaml:"maxBufferSize"`
	MaxMessageSize        *int                   `yaml:"maxMessageSize"`
	CloseCheckInterval    *time.Duration         `yaml:"closeCheckInterval"`
	DropOldestInterval    *time.Duration         `yaml:"dropOldestInterval"`
	ScanBatchSize         *int                   `yaml:"scanBatchSize"`
	AllowedSpilloverRatio *float64               `yaml:"allowedSpilloverRatio"`
	CleanupRetry          *retry.Configuration   `yaml:"cleanupRetry"`
}

BufferConfiguration configs the buffer.

func (*BufferConfiguration) NewOptions

func (c *BufferConfiguration) NewOptions(iOpts instrument.Options) buffer.Options

NewOptions creates new buffer options.

type ConnectionConfiguration

type ConnectionConfiguration struct {
	NumConnections  *int                 `yaml:"numConnections"`
	DialTimeout     *time.Duration       `yaml:"dialTimeout"`
	WriteTimeout    *time.Duration       `yaml:"writeTimeout"`
	KeepAlivePeriod *time.Duration       `yaml:"keepAlivePeriod"`
	ResetDelay      *time.Duration       `yaml:"resetDelay"`
	Retry           *retry.Configuration `yaml:"retry"`
	FlushInterval   *time.Duration       `yaml:"flushInterval"`
	WriteBufferSize *int                 `yaml:"writeBufferSize"`
	ReadBufferSize  *int                 `yaml:"readBufferSize"`
	// ContextDialer specifies a custom dialer to use when creating TCP connections to the consumer.
	// See writer.ConnectionOptions.ContextDialer for details.
	ContextDialer writer.ContextDialerFn `yaml:"-"` // not serializable
}

ConnectionConfiguration configs the connection options.

func (*ConnectionConfiguration) NewOptions

func (c *ConnectionConfiguration) NewOptions(iOpts instrument.Options) writer.ConnectionOptions

NewOptions creates connection options.

type ProducerConfiguration

type ProducerConfiguration struct {
	Buffer BufferConfiguration `yaml:"buffer"`
	Writer WriterConfiguration `yaml:"writer"`
}

ProducerConfiguration configs the producer.

func (*ProducerConfiguration) NewProducer

func (c *ProducerConfiguration) NewProducer(
	cs client.Client,
	iOpts instrument.Options,
	rwOpts xio.Options,
) (producer.Producer, error)

NewProducer creates new producer.

type StaticMessageRetryConfiguration

type StaticMessageRetryConfiguration struct {
	Backoff []time.Duration `yaml:"backoff"`
}

StaticMessageRetryConfiguration configs the static message retry policy. When messageRetry config exists, messageRetry will override the static config.

type WriterConfiguration

type WriterConfiguration struct {
	TopicName                         string                         `yaml:"topicName" validate:"nonzero"`
	TopicServiceOverride              kv.OverrideConfiguration       `yaml:"topicServiceOverride"`
	TopicWatchInitTimeout             *time.Duration                 `yaml:"topicWatchInitTimeout"`
	PlacementOptions                  placement.Configuration        `yaml:"placement"`
	PlacementServiceOverride          services.OverrideConfiguration `yaml:"placementServiceOverride"`
	PlacementWatchInitTimeout         *time.Duration                 `yaml:"placementWatchInitTimeout"`
	MessagePool                       *pool.ObjectPoolConfiguration  `yaml:"messagePool"`
	MessageQueueNewWritesScanInterval *time.Duration                 `yaml:"messageQueueNewWritesScanInterval"`
	MessageQueueFullScanInterval      *time.Duration                 `yaml:"messageQueueFullScanInterval"`
	MessageQueueScanBatchSize         *int                           `yaml:"messageQueueScanBatchSize"`
	InitialAckMapSize                 *int                           `yaml:"initialAckMapSize"`
	CloseCheckInterval                *time.Duration                 `yaml:"closeCheckInterval"`
	AckErrorRetry                     *retry.Configuration           `yaml:"ackErrorRetry"`
	Encoder                           *proto.Configuration           `yaml:"encoder"`
	Decoder                           *proto.Configuration           `yaml:"decoder"`
	Connection                        *ConnectionConfiguration       `yaml:"connection"`

	// StaticMessageRetry configs a static message retry policy.
	StaticMessageRetry *StaticMessageRetryConfiguration `yaml:"staticMessageRetry"`
	// MessageRetry configs a algorithmic retry policy.
	// Only one of the retry configuration should be used.
	MessageRetry *retry.Configuration `yaml:"messageRetry"`

	// IgnoreCutoffCutover allows producing writes ignoring cutoff/cutover timestamp.
	// Must be in sync with AggregatorConfiguration.WritesIgnoreCutoffCutover.
	IgnoreCutoffCutover bool `yaml:"ignoreCutoffCutover"`
	// WithoutConsumerScope drops the consumer tag from the metrics. For large m3msg deployments the consumer tag can
	// add a lot of cardinality to the metrics.
	WithoutConsumerScope bool `yaml:"withoutConsumerScope"`
}

WriterConfiguration configs the writer options.

func (*WriterConfiguration) NewOptions

func (c *WriterConfiguration) NewOptions(
	cs client.Client,
	iOpts instrument.Options,
	rwOptions xio.Options,
) (writer.Options, error)

NewOptions creates writer options.