conf

Documentation

Index

• Constants
• func LoadConfigFromFile(configFile string, configStruct interface{}, strict bool) error
• func LoadConfigFromMap(m map[string]interface{}, configStruct interface{}, strict bool) error
• func LoadConfigFromString(configString string, configStruct interface{}, strict bool) error
• func ValidateDirPath(path string) (string, error)
• func ValidateFilePath(path string) (string, error)
• type ConnectorConfig
• type HTTPConfig
• type JetStreamConfig
• type NATSConfig
• type NATSKafkaBridgeConfig
  • func DefaultBridgeConfig() NATSKafkaBridgeConfig
• type NATSStreamingConfig
• type SASL
• type TLSConf
  • func (tlsConf *TLSConf) MakeTLSConfig() (*tls.Config, error)

Constants

const (
	// Hash can be used in the connector config to set the hash balancing method
	Hash = "hash"
	// LeastBytes can be used in the connector config to set the least bytes received balancing method
	LeastBytes = "leastbytes"
)

const (
	// FixedKey in the keytype causes the key to always be the value in keyvalue
	FixedKey = "fixed"
	// SubjectKey in the keytype causes the key to be the incoming subject or channel
	SubjectKey = "subject"
	// ReplyToKey in the keytype causes the key to be the incoming reply-to subject or durable name
	ReplyToKey = "reply"
	// SubjectRegex in the keytype causes the key to use the keyvalue as a regex
	// on the incoming subject or channel and take the first match group as the key
	// for example foo\.(.*) for the subject/channel foo.bar would use "bar" as the key
	SubjectRegex = "subjectre"
	// ReplyRegex in the keytype causes the key to use the keyvalue as a regex
	// on the incoming reply-to and take the first match group as the key
	// for example foo\.(.*) for the subject/durable foo.bar would use "bar" as the key
	ReplyRegex = "replyre"
)

const (
	// KafkaToNATS specifies a connector from Kafka to NATS
	KafkaToNATS = "KafkaToNATS"
	// KafkaToStan specifies a connector from Kafka to NATS streaming
	KafkaToStan = "KafkaToStan"
	// KafkaToJetStream specifies a connector from Kafka to JetStream
	KafkaToJetStream = "KafkaToJetStream"

	// NATSToKafka specifies a connector from NATS to Kafka
	NATSToKafka = "NATSToKafka"
	// STANToKafka specifies a connector from NATS streaming to Kafka
	STANToKafka = "STANToKafka"
	// JetStreamToKafka specifies a connector from JetStream to Kafka
	JetStreamToKafka = "JetStreamToKafka"
)

Functions

func LoadConfigFromFile

func LoadConfigFromFile(configFile string, configStruct interface{}, strict bool) error

LoadConfigFromFile - given a struct, load a config from a file and fill in the struct If strict is true, all of the fields in the config struct must be in the file otherwise, the fields in the config struct will act as defaults if the file doesn't contain them Strict will also force an error if the struct contains any fields which are not settable with reflection

func LoadConfigFromMap

func LoadConfigFromMap(m map[string]interface{}, configStruct interface{}, strict bool) error

LoadConfigFromMap load a config struct from a map, this is useful if the type of a config isn't known at load time.

func LoadConfigFromString

func LoadConfigFromString(configString string, configStruct interface{}, strict bool) error

LoadConfigFromString - like LoadConfigFromFile but uses a string

func ValidateDirPath

func ValidateDirPath(path string) (string, error)

ValidateDirPath checks that the provided path exists and is a dir

func ValidateFilePath

func ValidateFilePath(path string) (string, error)

ValidateFilePath checks that the provided path exists and is not a dir

Types

type ConnectorConfig

type ConnectorConfig struct {
	ID   string // user specified id for a connector, will be defaulted if none is provided
	Type string // Can be any of the type constants (STANToKafka, ...)

	Channel         string // Used for stan connections
	DurableName     string // Optional, used for stan and jetstream connections
	StartAtSequence int64  // Start position for stan and jetstream connection, -1 means StartWithLastReceived, 0 means DeliverAllAvailable (default)
	StartAtTime     int64  // Start time, as Unix, time takes precedence over sequence

	Subject   string // Used for nats and jetstream connections
	QueueName string // Optional, used for nats connections
	Stream    string // Uses BindStream option for JetStream to consume from sourced streams

	Brokers []string // list of brokers to use for creating a reader/writer
	Topic   string   // kafka topic
	TLS     TLSConf  // tls config for connecting to the kafka brokers
	SASL    SASL     // SASL config for connecting to the kafka brokers, specifically EventHub

	MinBytes  int64  // used by the Kafka reader (for kafka->nats connectors)
	MaxBytes  int64  // used by the Kafka reader (for kafka->nats connectors)
	Partition int64  // optional partition for the reader
	GroupID   string // optional group id for reader, exclusive with partition

	Balancer string // can be hash or leastbytes

	KeyType  string // what to do with the key, can be FixedKey, ...
	KeyValue string // extra data for handling the key based on the keytype, may be ignored

	SchemaRegistryURL string // Schema registry url for message schema validation
	SubjectName       string // Name of the subject in the schema registry for the value
	SchemaVersion     int    // Version of the value schema to use. Default is latest.
	SchemaType        string // Can be avro, json, protobuf. Default is avro.
}

ConnectorConfig configuration for a bridge connection (of any type)

type HTTPConfig

type HTTPConfig struct {
	HTTPHost  string
	HTTPPort  int
	HTTPSPort int
	TLS       TLSConf

	ReadTimeout  int // milliseconds
	WriteTimeout int // milliseconds
}

HTTPConfig is used to specify the host/port/tls for an HTTP server

type JetStreamConfig

type JetStreamConfig struct {
	PublishAsyncMaxPending int
	MaxWait                int // milliseconds
	EnableFlowControl      bool
	EnableAckSync          bool
	HeartbeatInterval      int // milliseconds
}

JetStreamConfig configuration for a JetStream connection

type NATSConfig

type NATSConfig struct {
	Servers []string

	ConnectTimeout int // milliseconds
	ReconnectWait  int // milliseconds
	MaxReconnects  int

	TLS             TLSConf
	UserCredentials string
}

NATSConfig configuration for a NATS connection

type NATSKafkaBridgeConfig

type NATSKafkaBridgeConfig struct {
	ReconnectInterval int // milliseconds
	ConnectTimeout    int // milliseconds, connect timeout for Kafka connections

	Logging    logging.Config
	NATS       NATSConfig
	STAN       NATSStreamingConfig
	JetStream  JetStreamConfig
	Monitoring HTTPConfig
	Connect    []ConnectorConfig
}

NATSKafkaBridgeConfig is the root structure for a bridge configuration file.

func DefaultBridgeConfig

func DefaultBridgeConfig() NATSKafkaBridgeConfig

DefaultBridgeConfig generates a default configuration with logging set to colors, time, debug and trace

type NATSStreamingConfig

type NATSStreamingConfig struct {
	ClusterID string
	ClientID  string

	PubAckWait         int // milliseconds
	DiscoverPrefix     string
	MaxPubAcksInflight int
	ConnectWait        int // milliseconds
}

NATSStreamingConfig configuration for a STAN connection

type SASL

type SASL struct {
	User               string
	Password           string
	InsecureSkipVerify bool
	Mechanism          string
	TLS                bool
}

SASL holds the configuration for SASL authentication with Kafka.

type TLSConf

type TLSConf struct {
	Key  string
	Cert string
	Root string
}

TLSConf holds the configuration for a TLS connection/server

func (*TLSConf) MakeTLSConfig

func (tlsConf *TLSConf) MakeTLSConfig() (*tls.Config, error)

MakeTLSConfig creates a tls.Config from a TLSConf, setting up the key pairs and certs