config

Documentation

Index

• Constants
• func GetOrDefault[V any](config *Config, canonicalProperty string, defaultValue V) V
• func Unmarshall(content []byte, config *Config, toml bool) error
• type AwsConnectionConfig
• type AwsKinesisConfig
• type AwsKinesisStreamConfig
• type AwsSqsConfig
• type AwsSqsQueueConfig
• type Config
• type DispatcherConfig
• type EncodingConfig
• type EventFilterConfig
• type FileStorageConfig
• type HttpAuthenticationConfig
• type HttpAuthenticationType
• type HttpBasicAuthenticationConfig
• type HttpConfig
• type HttpHeaderAuthenticationConfig
• type IncludedTablesConfig
• type InitialSnapshotMode
• type InternalConfig
• type KafkaConfig
• type KafkaSaslConfig
• type LoggerConfig
• type LoggerConsoleConfig
• type LoggerFileConfig
• type LoggerOutputConfig
• type NamingStrategyType
• type NatsAuthorizationType
• type NatsConfig
• type NatsCredentialsConfig
• type NatsJWTConfig
• type NatsUserInfoConfig
• type PostgreSQLConfig
• type PostgresqlEventsConfig
• type PublicationConfig
• type RedisConfig
• type RedisRetryBackoffConfig
• type RedisRetryConfig
• type RedisTimeoutConfig
• type ReplicationSlotConfig
• type RuntimeStatsConfig
• type SinkConfig
• type SinkType
• type SnapshotConfig
• type SnapshotterConfig
• type StateStorageConfig
• type StateStorageType
• type StatsConfig
• type SubLoggerConfig
• type TLSConfig
• type TimescaleDBConfig
• type TimescaleEventsConfig
• type TopicConfig
• type TopicNamingStrategyConfig
• type TransactionConfig
• type TransactionWindowConfig

Constants

const (
	PropertyPostgresqlConnection              = "postgresql.connection"
	PropertyPostgresqlPassword                = "postgresql.password"
	PropertyPostgresqlPublicationName         = "postgresql.publication.name"
	PropertyPostgresqlPublicationCreate       = "postgresql.publication.create"
	PropertyPostgresqlPublicationAutoDrop     = "postgresql.publication.autodrop"
	PropertyPostgresqlSnapshotInitialMode     = "postgresql.snapshot.initial"
	PropertyPostgresqlSnapshotBatchsize       = "postgresql.snapshot.batchsize"
	PropertyPostgresqlReplicationSlotName     = "postgresql.replicationslot.name"
	PropertyPostgresqlReplicationSlotCreate   = "postgresql.replicationslot.create"
	PropertyPostgresqlReplicationSlotAutoDrop = "postgresql.replicationslot.autodrop"
	PropertyPostgresqlTxwindowEnabled         = "postgresql.transaction.window.enabled"
	PropertyPostgresqlTxwindowTimeout         = "postgresql.transaction.window.timeout"
	PropertyPostgresqlTxwindowMaxsize         = "postgresql.transaction.window.maxsize"

	PropertySink          = "sink.type"
	PropertySinkTombstone = "sink.tombstone"

	PropertyStatsEnabled        = "stats.enabled"
	PropertyStatsPort           = "stats.port"
	PropertyRuntimeStatsEnabled = "stats.runtime.enabled"

	PropertyStateStorageType     = "statestorage.type"
	PropertyFileStateStoragePath = "statestorage.file.path"

	PropertyDispatcherInitialQueueCapacity = "internal.dispatcher.initialqueuecapacity"
	PropertySnapshotterParallelism         = "internal.snapshotter.parallelism"
	PropertyEncodingCustomReflection       = "internal.encoding.customreflection"

	PropertyHypertableEventsRead          = "timescaledb.events.read"
	PropertyHypertableEventsInsert        = "timescaledb.events.insert"
	PropertyHypertableEventsUpdate        = "timescaledb.events.update"
	PropertyHypertableEventsDelete        = "timescaledb.events.delete"
	PropertyHypertableEventsTruncate      = "timescaledb.events.truncate"
	PropertyHypertableEventsCompression   = "timescaledb.events.compression"
	PropertyHypertableEventsDecompression = "timescaledb.events.decompression"
	PropertyHypertableEventsMessage       = "timescaledb.events.message" // FIXME: deprecated

	PropertyPostgresqlEventsRead     = "postgresql.events.read"
	PropertyPostgresqlEventsInsert   = "postgresql.events.insert"
	PropertyPostgresqlEventsUpdate   = "postgresql.events.update"
	PropertyPostgresqlEventsDelete   = "postgresql.events.delete"
	PropertyPostgresqlEventsTruncate = "postgresql.events.truncate"
	PropertyPostgresqlEventsMessage  = "postgresql.events.message"

	PropertyNamingStrategy = "topic.namingstrategy.type"

	PropertyKafkaBrokers       = "sink.kafka.brokers"
	PropertyKafkaSaslEnabled   = "sink.kafka.sasl.enabled"
	PropertyKafkaSaslUser      = "sink.kafka.sasl.user"
	PropertyKafkaSaslPassword  = "sink.kafka.sasl.password"
	PropertyKafkaSaslMechanism = "sink.kafka.sasl.mechanism"
	PropertyKafkaTlsEnabled    = "sink.kafka.tls.enabled"
	PropertyKafkaTlsSkipVerify = "sink.kafka.tls.skipverify"
	PropertyKafkaTlsClientAuth = "sink.kafka.tls.clientauth"

	PropertyNatsAddress                = "sink.nats.address"
	PropertyNatsAuthorization          = "sink.nats.authorization"
	PropertyNatsUserinfoUsername       = "sink.nats.userinfo.username"
	PropertyNatsUserinfoPassword       = "sink.nats.userinfo.password"
	PropertyNatsCredentialsCertificate = "sink.nats.credentials.certificate"
	PropertyNatsCredentialsSeeds       = "sink.nats.credentials.seeds"
	PropertyNatsJwt                    = "sink.nats.jwt.jwt"
	PropertyNatsJwtSeed                = "sink.nats.jwt.seed"

	PropertyRedisNetwork           = "sink.redis.network"
	PropertyRedisAddress           = "sink.redis.address"
	PropertyRedisPassword          = "sink.redis.password"
	PropertyRedisDatabase          = "sink.redis.database"
	PropertyRedisPoolsize          = "sink.redis.poolsize"
	PropertyRedisRetriesMax        = "sink.redis.retries.maxattempts"
	PropertyRedisRetriesBackoffMin = "sink.redis.retries.backoff.min"
	PropertyRedisRetriesBackoffMax = "sink.redis.retries.backoff.max"
	PropertyRedisTimeoutDial       = "sink.redis.timeouts.dial"
	PropertyRedisTimeoutRead       = "sink.redis.timeouts.read"
	PropertyRedisTimeoutWrite      = "sink.redis.timeouts.write"
	PropertyRedisTimeoutPool       = "sink.redis.timeouts.pool"
	PropertyRedisTimeoutIdle       = "sink.redis.timeouts.idle"
	PropertyRedisTlsSkipVerify     = "sink.redis.tls.skipverify"
	PropertyRedisTlsClientAuth     = "sink.redis.tls.clientauth"

	PropertyKinesisStreamName         = "sink.kinesis.stream.name"
	PropertyKinesisStreamCreate       = "sink.kinesis.stream.create"
	PropertyKinesisStreamShardCount   = "sink.kinesis.stream.shardcount"
	PropertyKinesisStreamMode         = "sink.kinesis.stream.mode"
	PropertyKinesisRegion             = "sink.kinesis.aws.region"
	PropertyKinesisAwsEndpoint        = "sink.kinesis.aws.endpoint"
	PropertyKinesisAwsAccessKeyId     = "sink.kinesis.aws.accesskeyid"
	PropertyKinesisAwsSecretAccessKey = "sink.kinesis.aws.secretaccesskey"
	PropertyKinesisAwsSessionToken    = "sink.kinesis.aws.sessiontoken"

	PropertySqsQueueUrl           = "sink.sqs.queue.url"
	PropertySqsAwsRegion          = "sink.sqs.aws.region"
	PropertySqsAwsEndpoint        = "sink.sqs.aws.endpoint"
	PropertySqsAwsAccessKeyId     = "sink.sqs.aws.accesskeyid"
	PropertySqsAwsSecretAccessKey = "sink.sqs.aws.secretaccesskey"
	PropertySqsAwsSessionToken    = "sink.sqs.aws.sessiontoken"

	PropertyHttpUrl                             = "sink.http.url"
	PropertyHttpAuthenticationType              = "sink.http.authentication.type"
	PropertyHttpBasicAuthenticationUsername     = "sink.http.authentication.basic.username"
	PropertyHttpBasicAuthenticationPassword     = "sink.http.authentication.basic.password"
	PropertyHttpHeaderAuthenticationHeaderName  = "sink.http.authentication.header.name"
	PropertyHttpHeaderAuthenticationHeaderValue = "sink.http.authentication.header.value"
	PropertyHttpTlsSkipVerify                   = "sink.http.tls.skipverify"
	PropertyHttpTlsClientAuth                   = "sink.http.tls.clientauth"
)

Functions

func GetOrDefault

func GetOrDefault[V any](
	config *Config, canonicalProperty string, defaultValue V,
) V

func Unmarshall

func Unmarshall(
	content []byte, config *Config, toml bool,
) error

Types

type AwsConnectionConfig

type AwsConnectionConfig struct {
	Region          *string `toml:"region" yaml:"region"`
	Endpoint        string  `toml:"endpoint" yaml:"endpoint"`
	AccessKeyId     string  `toml:"accesskeyid" yaml:"accessKeyId"`
	SecretAccessKey string  `toml:"secretaccesskey" yaml:"secretAccessKey"`
	SessionToken    string  `toml:"sessiontoken" yaml:"sessionToken"`
}

type AwsKinesisConfig

type AwsKinesisConfig struct {
	Stream AwsKinesisStreamConfig `toml:"stream" yaml:"stream"`
	Aws    AwsConnectionConfig    `toml:"aws" yaml:"aws"`
}

type AwsKinesisStreamConfig

type AwsKinesisStreamConfig struct {
	Name       *string `toml:"name" yaml:"name"`
	Create     *bool   `toml:"create" yaml:"create"`
	ShardCount *int64  `toml:"shardcount" yaml:"shardCount"`
	Mode       *string `toml:"mode" yaml:"mode"`
}

type AwsSqsConfig

type AwsSqsConfig struct {
	Queue AwsSqsQueueConfig   `toml:"queue" yaml:"queue"`
	Aws   AwsConnectionConfig `toml:"aws" yaml:"aws"`
}

type AwsSqsQueueConfig

type AwsSqsQueueConfig struct {
	Url *string `toml:"url" yaml:"url"`
}

type Config

type Config struct {
	PostgreSQL   PostgreSQLConfig   `toml:"postgresql" yaml:"postgresql"`
	Sink         SinkConfig         `toml:"sink" yaml:"sink"`
	Topic        TopicConfig        `toml:"topic" yaml:"topic"`
	TimescaleDB  TimescaleDBConfig  `toml:"timescaledb" yaml:"timescaledb"`
	Logging      LoggerConfig       `toml:"logging" yaml:"logging"`
	StateStorage StateStorageConfig `toml:"statestorage" yaml:"stateStorage"`
	Internal     InternalConfig     `toml:"internal" yaml:"internal"`
	Plugins      []string           `toml:"plugins" yaml:"plugins"`
	Stats        StatsConfig        `toml:"stats" yaml:"stats"`
}

type DispatcherConfig

type DispatcherConfig struct {
	InitialQueueCapacity uint `toml:"initialqueuecapacity" yaml:"initialQueueCapacity"`
}

type EncodingConfig

type EncodingConfig struct {
	CustomReflection *bool `toml:"customreflection" yaml:"customReflection"`
}

type EventFilterConfig

type EventFilterConfig struct {
	Tables       *IncludedTablesConfig `toml:"tables" yaml:"tables"`
	DefaultValue *bool                 `toml:"default" yaml:"default"`
	Condition    string                `toml:"condition" yaml:"condition"`
}

type FileStorageConfig

type FileStorageConfig struct {
	Path string `toml:"path" yaml:"path"`
}

type HttpAuthenticationConfig

type HttpAuthenticationConfig struct {
	Type   HttpAuthenticationType         `toml:"type" yaml:"type"`
	Basic  HttpBasicAuthenticationConfig  `toml:"basic" yaml:"basic"`
	Header HttpHeaderAuthenticationConfig `toml:"header" yaml:"header"`
}

type HttpAuthenticationType

type HttpAuthenticationType string

const (
	NoneAuthentication   HttpAuthenticationType = "none"
	BasicAuthentication  HttpAuthenticationType = "basic"
	HeaderAuthentication HttpAuthenticationType = "header"
)

type HttpBasicAuthenticationConfig

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

type HttpConfig

type HttpConfig struct {
	Url            string                   `toml:"url" yaml:"url"`
	Authentication HttpAuthenticationConfig `toml:"authentication" yaml:"authentication"`
	TLS            TLSConfig                `toml:"tls" yaml:"tls"`
}

type HttpHeaderAuthenticationConfig

type HttpHeaderAuthenticationConfig struct {
	Name  string `toml:"name" yaml:"name"`
	Value string `toml:"value" yaml:"value"`
}

type IncludedTablesConfig

type IncludedTablesConfig struct {
	Excludes []string `toml:"excludes" yaml:"excludes"`
	Includes []string `toml:"includes" yaml:"includes"`
}

type InitialSnapshotMode

type InitialSnapshotMode string

const (
	Always      InitialSnapshotMode = "always"
	Never       InitialSnapshotMode = "never"
	InitialOnly InitialSnapshotMode = "initial_only"
)

type InternalConfig

type InternalConfig struct {
	Dispatcher  DispatcherConfig  `toml:"dispatcher" yaml:"dispatcher"`
	Snapshotter SnapshotterConfig `toml:"snapshotter" yaml:"snapshotter"`
	Encoding    EncodingConfig    `toml:"encoding" yaml:"encoding"`
}

type KafkaConfig

type KafkaConfig struct {
	Brokers    []string        `toml:"brokers" yaml:"brokers"`
	Idempotent *bool           `toml:"idempotent" yaml:"idempotent"`
	Sasl       KafkaSaslConfig `toml:"sasl" yaml:"sasl"`
	TLS        TLSConfig       `toml:"tls" yaml:"tls"`
}

type KafkaSaslConfig

type KafkaSaslConfig struct {
	Enabled   *bool                `toml:"enabled" yaml:"enabled"`
	User      string               `toml:"user" yaml:"user"`
	Password  string               `toml:"password" yaml:"password"`
	Mechanism sarama.SASLMechanism `toml:"mechanism" yaml:"mechanism"`
}

type LoggerConfig

type LoggerConfig struct {
	Level   string                     `toml:"level" yaml:"level"`
	Outputs LoggerOutputConfig         `toml:"outputs" yaml:"outputs"`
	Loggers map[string]SubLoggerConfig `toml:"loggers" yaml:"loggers"`
}

type LoggerConsoleConfig

type LoggerConsoleConfig struct {
	Enabled *bool `toml:"enabled" yaml:"enabled"`
}

type LoggerFileConfig

type LoggerFileConfig struct {
	Enabled     *bool   `toml:"enabled" yaml:"enabled"`
	Path        string  `toml:"path" yaml:"path"`
	Rotate      *bool   `toml:"rotate" yaml:"rotate"`
	MaxSize     *string `toml:"maxsize" yaml:"maxSize"`
	MaxDuration *int    `toml:"maxduration" yaml:"maxDuration"`
	Compress    bool    `toml:"compress" yaml:"compress"`
}

type LoggerOutputConfig

type LoggerOutputConfig struct {
	Console LoggerConsoleConfig `toml:"console" yaml:"console"`
	File    LoggerFileConfig    `toml:"file" yaml:"file"`
}

type NamingStrategyType

type NamingStrategyType string

const (
	Debezium NamingStrategyType = "debezium"
)

type NatsAuthorizationType

type NatsAuthorizationType string

const (
	UserInfo    NatsAuthorizationType = "userinfo"
	Credentials NatsAuthorizationType = "credentials"
	Jwt         NatsAuthorizationType = "jwt"
)

type NatsConfig

type NatsConfig struct {
	Address       string                `toml:"address" yaml:"address"`
	Authorization NatsAuthorizationType `toml:"authorization" yaml:"authorization"`
	UserInfo      NatsUserInfoConfig    `toml:"userinfo" yaml:"userInfo"`
	Credentials   NatsCredentialsConfig `toml:"credentials" yaml:"credentials"`
	JWT           NatsJWTConfig         `toml:"jwt" yaml:"jwt"`
}

type NatsCredentialsConfig

type NatsCredentialsConfig struct {
	Certificate string   `toml:"certificate" yaml:"certificate"`
	Seeds       []string `toml:"seeds" yaml:"seeds"`
}

type NatsJWTConfig

type NatsJWTConfig struct {
	JWT  string `toml:"jwt" yaml:"jwt"`
	Seed string `toml:"seed" yaml:"Seed"`
}

type NatsUserInfoConfig

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

type PostgreSQLConfig

type PostgreSQLConfig struct {
	Connection      string                 `toml:"connection" yaml:"connection"`
	Password        string                 `toml:"password" yaml:"password"`
	Publication     PublicationConfig      `toml:"publication" yaml:"publication"`
	ReplicationSlot ReplicationSlotConfig  `toml:"replicationslot" yaml:"replicationSlot"`
	Transaction     TransactionConfig      `toml:"transaction" yaml:"transaction"`
	Snapshot        SnapshotConfig         `toml:"snapshot" yaml:"snapshot"`
	Tables          IncludedTablesConfig   `toml:"tables" yaml:"tables"`
	Events          PostgresqlEventsConfig `toml:"events" yaml:"events"`
}

type PostgresqlEventsConfig

type PostgresqlEventsConfig struct {
	Read     *bool `toml:"read" yaml:"read"`
	Insert   *bool `toml:"insert" yaml:"insert"`
	Update   *bool `toml:"update" yaml:"update"`
	Delete   *bool `toml:"delete" yaml:"delete"`
	Truncate *bool `toml:"truncate" yaml:"truncate"`
	Message  *bool `toml:"message" yaml:"message"`
}

type PublicationConfig

type PublicationConfig struct {
	Name     string `toml:"name" yaml:"name"`
	Create   *bool  `toml:"create" yaml:"create"`
	AutoDrop *bool  `toml:"autodrop" yaml:"autoDrop"`
}

type RedisConfig

type RedisConfig struct {
	Network  string             `toml:"network" yaml:"network"`
	Address  string             `toml:"address" yaml:"address"`
	Password string             `toml:"password" yaml:"password"`
	Database int                `toml:"database" yaml:"database"`
	Retries  RedisRetryConfig   `toml:"retries" yaml:"retries"`
	Timeouts RedisTimeoutConfig `toml:"timeouts" yaml:"timeouts"`
	PoolSize int                `toml:"poolsize" yaml:"poolSize"`
	TLS      TLSConfig          `toml:"tls" yaml:"tls"`
}

type RedisRetryBackoffConfig

type RedisRetryBackoffConfig struct {
	Min int `toml:"min" yaml:"min"`
	Max int `toml:"max" yaml:"max"`
}

type RedisRetryConfig

type RedisRetryConfig struct {
	MaxAttempts int                     `toml:"maxattempts" yaml:"maxAttempts"`
	Backoff     RedisRetryBackoffConfig `toml:"backoff" yaml:"backoff"`
}

type RedisTimeoutConfig

type RedisTimeoutConfig struct {
	Dial  int `toml:"dial" yaml:"dial"`
	Read  int `toml:"read" yaml:"read"`
	Write int `toml:"write" yaml:"write"`
	Pool  int `toml:"pool" yaml:"pool"`
	Idle  int `toml:"idle" yaml:"idle"`
}

type ReplicationSlotConfig

type ReplicationSlotConfig struct {
	Name     string `toml:"name" yaml:"name"`
	Create   *bool  `toml:"create" yaml:"create"`
	AutoDrop *bool  `toml:"autodrop" yaml:"autoDrop"`
}

type RuntimeStatsConfig

type RuntimeStatsConfig struct {
	Enabled *bool `toml:"enabled" yaml:"enabled"`
}

type SinkConfig

type SinkConfig struct {
	Type       SinkType                     `toml:"type" yaml:"type"`
	Tombstone  *bool                        `toml:"tombstone" yaml:"tombstone"`
	Filters    map[string]EventFilterConfig `toml:"filters" yaml:"filters"`
	Nats       NatsConfig                   `toml:"nats" yaml:"nats"`
	Kafka      KafkaConfig                  `toml:"kafka" yaml:"kafka"`
	Redis      RedisConfig                  `toml:"redis" yaml:"redis"`
	AwsKinesis AwsKinesisConfig             `toml:"kinesis" yaml:"kinesis"`
	AwsSqs     AwsSqsConfig                 `toml:"sqs" yaml:"sqs"`
	Http       HttpConfig                   `toml:"http" yaml:"http"`
}

type SinkType

type SinkType string

const (
	Stdout     SinkType = "stdout"
	NATS       SinkType = "nats"
	Kafka      SinkType = "kafka"
	Redis      SinkType = "redis"
	AwsKinesis SinkType = "kinesis"
	AwsSQS     SinkType = "sqs"
	Http       SinkType = "http"
)

type SnapshotConfig

type SnapshotConfig struct {
	BatchSize uint                 `toml:"batchsize" yaml:"batchSize"`
	Initial   *InitialSnapshotMode `toml:"initial" yaml:"initial"`
}

type SnapshotterConfig

type SnapshotterConfig struct {
	Parallelism uint8 `toml:"parallelism" yaml:"parallelism"`
}

type StateStorageConfig

type StateStorageConfig struct {
	Type        StateStorageType  `toml:"type" yaml:"type"`
	FileStorage FileStorageConfig `toml:"file" yaml:"file"`
}

type StateStorageType

type StateStorageType string

const (
	NoneStorage StateStorageType = "none"
	FileStorage StateStorageType = "file"
)

type StatsConfig

type StatsConfig struct {
	Enabled *bool              `toml:"enabled" yaml:"enabled"`
	Port    *int               `toml:"port" yaml:"port"`
	Runtime RuntimeStatsConfig `toml:"runtime" yaml:"runtime"`
}

type SubLoggerConfig

type SubLoggerConfig struct {
	Level   *string            `toml:"level" yaml:"level"`
	Outputs LoggerOutputConfig `toml:"outputs" yaml:"outputs"`
}

type TLSConfig

type TLSConfig struct {
	Enabled    bool               `toml:"enabled" yaml:"enabled"`
	SkipVerify *bool              `toml:"skipverify" yaml:"skipVerify"`
	ClientAuth tls.ClientAuthType `toml:"clientauth" yaml:"clientAuth"`
}

type TimescaleDBConfig

type TimescaleDBConfig struct {
	Hypertables IncludedTablesConfig  `toml:"hypertables" yaml:"hypertables"`
	Events      TimescaleEventsConfig `toml:"events" yaml:"events"`
}

type TimescaleEventsConfig

type TimescaleEventsConfig struct {
	Read          *bool `toml:"read" yaml:"read"`
	Insert        *bool `toml:"insert" yaml:"insert"`
	Update        *bool `toml:"update" yaml:"update"`
	Delete        *bool `toml:"delete" yaml:"delete"`
	Truncate      *bool `toml:"truncate" yaml:"truncate"`
	Message       *bool `toml:"message" yaml:"message"` // deprecated
	Compression   *bool `toml:"compression" yaml:"compression"`
	Decompression *bool `toml:"decompression" yaml:"decompression"`
}

type TopicConfig

type TopicConfig struct {
	NamingStrategy TopicNamingStrategyConfig `toml:"namingstrategy" yaml:"namingStrategy"`
	Prefix         string                    `toml:"prefix" yaml:"prefix"`
}

type TopicNamingStrategyConfig

type TopicNamingStrategyConfig struct {
	Type NamingStrategyType `toml:"type" yaml:"type"`
}

type TransactionConfig

type TransactionConfig struct {
	Window TransactionWindowConfig `toml:"window" yaml:"window"`
}

type TransactionWindowConfig

type TransactionWindowConfig struct {
	Enabled *bool `toml:"enabled" yaml:"enabled"`
	Timeout int   `toml:"timeout" yaml:"timeout"`
	MaxSize uint  `toml:"maxsize" yaml:"maxSize"`
}