sources

Documentation

Index

• func NewFileSource(path, readFrom string, parsers []parsers.Parser, ...) fileSource
• func NewKafkaSource(config KafkaSourceConfig, parsers []parsers.Parser) kafkaSource
• type FileSourceConfig
• type KafkaSaslConfig
• type KafkaSourceConfig
• type Source

Functions

func NewFileSource

func NewFileSource(path, readFrom string, parsers []parsers.Parser, receivers []processors.StreamingProcessor) fileSource

func NewKafkaSource

func NewKafkaSource(config KafkaSourceConfig, parsers []parsers.Parser) kafkaSource

Types

type FileSourceConfig

type FileSourceConfig struct {
	Path     string `yaml:"path"`
	ReadFrom string `yaml:"read-from"`
}

type KafkaSaslConfig

type KafkaSaslConfig struct {
	Username string `yaml:"username"`
	Password string `yaml:"password"`
}

type KafkaSourceConfig

type KafkaSourceConfig struct {
	Brokers            string           `yaml:"brokers"`
	Tls                bool             `yaml:"tls"`
	GroupID            string           `yaml:"group-id"`
	Topic              string           `yaml:"topic"`
	MinBytes           int              `yaml:"min-bytes"`
	MaxBytes           int              `yaml:"max-bytes"`
	CommitInterval     time.Duration    `yaml:"commit-interval"`
	Timeout            time.Duration    `yaml:"timeout"`
	StartFromBeginning bool             `yaml:"start-from-beginning"`
	SaslConfig         *KafkaSaslConfig `yaml:"sasl-config"`
}

type Source

type Source interface {
	AddProcessors([]processors.StreamingProcessor)
	Run(context.Context, *sync.WaitGroup)
	Close()
}