egress

Documentation

Index

• Variables
• type Controller
  • func New(config ControllerConfig) (*Controller, error)
  • func (c *Controller) Start() error
  • func (c *Controller) Stop()
• type ControllerConfig
• type TopicConfig

Variables

var DefaultTopicConfig = TopicConfig{
	MaxSegmentMessages: 1000000,
	MaxSegmentSize:     100 * 1024 * 1024,
	MaxSegmentAge:      5 * time.Minute,
	WorkerChanSize:     10000,
	WorkerStopTimeout:  10 * time.Second,
}

DefaultTopicConfig is the default egress topic configuration.

Types

type Controller

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

Controller represents the egress controller

func New

func New(config ControllerConfig) (*Controller, error)

New returns a new egress controller instance

func (*Controller) Start

func (c *Controller) Start() error

Start will start the controller

func (*Controller) Stop

func (c *Controller) Stop()

Stop will stop the controller

type ControllerConfig

type ControllerConfig struct {
	// Consumer is used to consume topic messages.
	Consumer core.Factory `required:"true"`

	// SegmentStore is used to write segments.
	SegmentStore core.Factory `required:"true"`

	// Unique name that identifies the local region/data center/cloud.
	//
	// Field value is required.
	LocalRegion string `required:"true"`

	// Source Kafka topic names that will be consumed and written to segment store.
	//
	// Will use DefaultTopicConfig if topic config was not set.
	//
	// Field value is required.
	Topics map[string]*TopicConfig `required:"true"`
}

ControllerConfig represents the egress controller configuration

type TopicConfig

type TopicConfig struct {
	// Maximum number of messages written to a segment.
	MaxSegmentMessages int `min:"2"`

	// Maximum byte size of messages contained in a segment.
	//
	// The value is the raw Kafka message size before compression
	// and other possible encoding performed by the formatter.
	MaxSegmentSize uint64 `min:"1"`

	// Maximum duration from segment creation to completion.
	//
	// A higher value results in fewer and larger segments being created
	// at the expense of increased message delivery latency.
	MaxSegmentAge time.Duration `min:"1ms"`

	// Size of egress worker buffered channel.
	WorkerChanSize int `min:"1"`

	// Egress worker shutdown grace period.
	WorkerStopTimeout time.Duration `min:"1ms"`

	// Retrier instance used for Consumer operations
	ConsumerRetrier core.Retrier

	// Retrier instance used for SegmentStore operations
	SegmentStoreRetrier core.Retrier
}

TopicConfig represents the egress topic configuration.