kafka_input

Documentation

Index

• Variables
• type Config
• type KafkaInput
  • func (k *KafkaInput) Close() error
  • func (k *KafkaInput) SetEventProcessors(ps map[string]map[string]interface{}, logger *log.Logger, ...)
  • func (k *KafkaInput) SetLogger(logger *log.Logger)
  • func (k *KafkaInput) SetName(name string)
  • func (k *KafkaInput) SetOutputs(outs map[string]outputs.Output)
  • func (k *KafkaInput) Start(ctx context.Context, name string, cfg map[string]interface{}, ...) error
• type XDGSCRAMClient
  • func (x *XDGSCRAMClient) Begin(userName, password, authzID string) (err error)
  • func (x *XDGSCRAMClient) Done() bool
  • func (x *XDGSCRAMClient) Step(challenge string) (response string, err error)

Variables

var SHA256 scram.HashGeneratorFcn = func() hash.Hash { return sha256.New() }

var SHA512 scram.HashGeneratorFcn = func() hash.Hash { return sha512.New() }

Types

type Config

type Config struct {
	Name              string           `mapstructure:"name,omitempty"`
	Address           string           `mapstructure:"address,omitempty"`
	Topics            string           `mapstructure:"topics,omitempty"`
	SASL              *types.SASL      `mapstructure:"sasl,omitempty"`
	TLS               *types.TLSConfig `mapstructure:"tls,omitempty"`
	GroupID           string           `mapstructure:"group-id,omitempty"`
	SessionTimeout    time.Duration    `mapstructure:"session-timeout,omitempty"`
	HeartbeatInterval time.Duration    `mapstructure:"heartbeat-interval,omitempty"`
	RecoveryWaitTime  time.Duration    `mapstructure:"recovery-wait-time,omitempty"`
	Version           string           `mapstructure:"version,omitempty"`
	Format            string           `mapstructure:"format,omitempty"`
	Debug             bool             `mapstructure:"debug,omitempty"`
	NumWorkers        int              `mapstructure:"num-workers,omitempty"`
	Outputs           []string         `mapstructure:"outputs,omitempty"`
	EventProcessors   []string         `mapstructure:"event-processors,omitempty"`
	// contains filtered or unexported fields
}

Config //

type KafkaInput

type KafkaInput struct {
	Cfg *Config
	// contains filtered or unexported fields
}

KafkaInput //

func (*KafkaInput) Close

func (k *KafkaInput) Close() error

func (*KafkaInput) SetEventProcessors

func (k *KafkaInput) SetEventProcessors(ps map[string]map[string]interface{}, logger *log.Logger, tcs map[string]*types.TargetConfig)

func (*KafkaInput) SetLogger

func (k *KafkaInput) SetLogger(logger *log.Logger)

func (*KafkaInput) SetName

func (k *KafkaInput) SetName(name string)

func (*KafkaInput) SetOutputs

func (k *KafkaInput) SetOutputs(outs map[string]outputs.Output)

func (*KafkaInput) Start

func (k *KafkaInput) Start(ctx context.Context, name string, cfg map[string]interface{}, opts ...inputs.Option) error

type XDGSCRAMClient

type XDGSCRAMClient struct {
	*scram.Client
	*scram.ClientConversation
	scram.HashGeneratorFcn
}

func (*XDGSCRAMClient) Begin

func (x *XDGSCRAMClient) Begin(userName, password, authzID string) (err error)

func (*XDGSCRAMClient) Done

func (x *XDGSCRAMClient) Done() bool

func (*XDGSCRAMClient) Step

func (x *XDGSCRAMClient) Step(challenge string) (response string, err error)