config

Documentation

Overview

Package config contains the configuration structs and validators used in the project.

Index

• func ConsumerStructLevelValidator(structLevel validator.StructLevel)
• func Validate(cfg *Config) error
• type Config
  • func Default() Config
  • func Read(filename string) (Config, error)
• type Consumer
• type JetstreamConsumerConfig
  • func DefaultJetstream() JetstreamConsumerConfig
• type JetstreamConsumerSettings
• type KafkaConsumerConfig
  • func DefaultKafka() KafkaConsumerConfig
• type KafkaConsumerSettings
• type KerberosConfig
• type PlainSASLConfig
• type PrometheusConfig
• type PubsubConsumerConfig
  • func DefaultPubsub() PubsubConsumerConfig
• type PubsubConsumerSettings
• type PulsarConsumerConfig
  • func DefaultPulsar() PulsarConsumerConfig
• type PulsarConsumerSettings
• type RunOptions
• type ServiceBusConsumerConfig
  • func DefaultServiceBus() ServiceBusConsumerConfig
• type ServiceBusConsumerSettings
• type TLSConfig

Functions

func ConsumerStructLevelValidator

func ConsumerStructLevelValidator(structLevel validator.StructLevel)

ConsumerStructLevelValidator is a custom validator which validates broker structure.

func Validate

func Validate(cfg *Config) error

Validate validates CentralConsumer struct.

Types

type Config

type Config struct {
	Consumer   Consumer `toml:"consumer"`
	RunOptions RunOptions
}

func Default

func Default() Config

Default returns a Config object with default values.

func Read

func Read(filename string) (Config, error)

Read returns a Config structure for the given path to a .toml file.

type Consumer

type Consumer struct {
	Type       string                   `toml:"type" val:"oneof=kafka pubsub servicebus jetstream pulsar"`
	Kafka      KafkaConsumerConfig      `toml:"kafka"`
	Pubsub     PubsubConsumerConfig     `toml:"pubsub"`
	ServiceBus ServiceBusConsumerConfig `toml:"servicebus"`
	Jetstream  JetstreamConsumerConfig  `toml:"jetstream"`
	Pulsar     PulsarConsumerConfig     `toml:"pulsar"`
}

type JetstreamConsumerConfig

type JetstreamConsumerConfig struct {
	URL          string                    `toml:"url"`
	Subject      string                    `toml:"subject"`
	ConsumerName string                    `toml:"consumer_name"`
	Settings     JetstreamConsumerSettings `toml:"settings"`
}

func DefaultJetstream

func DefaultJetstream() JetstreamConsumerConfig

type JetstreamConsumerSettings

type JetstreamConsumerSettings struct {
	BatchSize int `toml:"batch_size"`
}

type KafkaConsumerConfig

type KafkaConsumerConfig struct {
	Address    string                `toml:"address"`
	TLSConfig  TLSConfig             `toml:"tls_config"`
	Topic      string                `toml:"topic"`
	GroupID    string                `toml:"group_id"`
	Kerberos   KerberosConfig        `toml:"kerberos"`
	PlainSASL  PlainSASLConfig       `toml:"plain_sasl"`
	Prometheus PrometheusConfig      `toml:"prometheus"`
	Settings   KafkaConsumerSettings `toml:"settings"`
}

func DefaultKafka

func DefaultKafka() KafkaConsumerConfig

type KafkaConsumerSettings

type KafkaConsumerSettings struct {
	MinBytes             int           `toml:"min_bytes"`
	MaxWait              time.Duration `toml:"max_wait"`
	MaxBytes             int           `toml:"max_bytes"`
	MaxConcurrentFetches int           `toml:"max_concurrent_fetches"`
	MaxPollRecords       int           `toml:"max_poll_records"`
	Transactional        bool          `toml:"transactional"`
}

type KerberosConfig

type KerberosConfig struct {
	Enabled    bool   `toml:"enabled"`
	KeytabPath string `toml:"key_tab_path"`
	ConfigPath string `toml:"config_path"`
	Realm      string `toml:"realm"`
	Service    string `toml:"service"`
	Username   string `toml:"username"`
}

type PlainSASLConfig

type PlainSASLConfig struct {
	Enabled bool   `toml:"enabled"`
	Zid     string `toml:"zid"`
	User    string `toml:"user"`
	Pass    string `toml:"pass"`
}

type PrometheusConfig

type PrometheusConfig struct {
	Enabled    bool   `toml:"enabled"`
	Namespace  string `toml:"namespace"`
	Registerer prometheus.Registerer
	Gatherer   prometheus.Gatherer
}

type PubsubConsumerConfig

type PubsubConsumerConfig struct {
	ProjectID      string                 `toml:"project_id"`
	SubscriptionID string                 `toml:"subscription_id"`
	Settings       PubsubConsumerSettings `toml:"settings"`
}

func DefaultPubsub

func DefaultPubsub() PubsubConsumerConfig

type PubsubConsumerSettings

type PubsubConsumerSettings struct {
	MaxExtension           time.Duration `toml:"max_extension"`
	MaxExtensionPeriod     time.Duration `toml:"max_extension_period"`
	MaxOutstandingMessages int           `toml:"max_outstanding_messages"`
	MaxOutstandingBytes    int           `toml:"max_outstanding_bytes"`
	NumGoroutines          int           `toml:"num_goroutines"`
}

type PulsarConsumerConfig

type PulsarConsumerConfig struct {
	ServiceURL   string                 `toml:"service_url"`
	Topic        string                 `toml:"topic"`
	Subscription string                 `toml:"subscription"`
	TLSConfig    TLSConfig              `toml:"tls_config"`
	Settings     PulsarConsumerSettings `toml:"settings"`
}

func DefaultPulsar

func DefaultPulsar() PulsarConsumerConfig

type PulsarConsumerSettings

type PulsarConsumerSettings struct {
	ConnectionTimeout          time.Duration `toml:"connection_timeout"`
	OperationTimeout           time.Duration `toml:"operation_timeout"`
	NackRedeliveryDelay        time.Duration `toml:"nack_redelivery_delay"`
	MaxConnectionsPerBroker    int           `toml:"max_connections_per_broker"`
	SubscriptionType           string        `toml:"subscription_type"`
	ReceiverQueueSize          int           `toml:"receiver_queue_size"`
	MaxReconnectToBroker       *uint         `toml:"max_reconnect_to_broker"`
	TLSTrustCertsFilePath      string        `toml:"tls_trust_certs_file_path"`
	TLSAllowInsecureConnection bool          `toml:"tls_allow_insecure_connection" `
}

type RunOptions

type RunOptions struct {
	ErrThreshold int64         `toml:"err_threshold"`
	ErrInterval  time.Duration `toml:"err_interval"`
	NumRetries   int           `toml:"num_retries"`
}

type ServiceBusConsumerConfig

type ServiceBusConsumerConfig struct {
	ConnectionString string                     `toml:"connection_string"`
	Topic            string                     `toml:"topic"`
	Subscription     string                     `toml:"subscription"`
	Settings         ServiceBusConsumerSettings `toml:"settings"`
}

func DefaultServiceBus

func DefaultServiceBus() ServiceBusConsumerConfig

type ServiceBusConsumerSettings

type ServiceBusConsumerSettings struct {
	BatchSize int `toml:"batch_size"`
}

type TLSConfig

type TLSConfig struct {
	Enabled        bool   `toml:"enabled"`
	ClientCertFile string `toml:"client_cert_file" val:"required_if=Enabled true,omitempty,file"`
	ClientKeyFile  string `toml:"client_key_file" val:"required_if=Enabled true,omitempty,file"`
	CaCertFile     string `toml:"ca_cert_file" val:"required_if=Enabled true,omitempty,file"`
}