aukafka

Documentation

Index

• Constants
• func NewSha256ScramClient() sarama.SCRAMClient
• func NewSha512ScramClient() sarama.SCRAMClient
• func ParseTopicConfigs(jsonString string) (map[string]TopicConfig, error)
• type Config
  • func NewConfig() *Config
  • func (c *Config) ConfigItems() []auconfigapi.ConfigItem
  • func (c *Config) Obtain(getter func(key string) string)
  • func (c *Config) TopicConfigs() map[string]TopicConfig
• type Consumer
  • func CreateConsumer[E any](ctx context.Context, topicConfig TopicConfig, ...) (*Consumer[E], error)
  • func (c *Consumer[E]) Cleanup(sarama.ConsumerGroupSession) error
  • func (c *Consumer[E]) Close(ctx context.Context)
  • func (c *Consumer[E]) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (c *Consumer[E]) Setup(sarama.ConsumerGroupSession) error
• type DefaultConfigImpl
  • func NewDefaultConfig() *DefaultConfigImpl
  • func (c *DefaultConfigImpl) ConfigItems() []auconfigapi.ConfigItem
  • func (c *DefaultConfigImpl) ObtainValues(getter func(string) string) error
  • func (c *DefaultConfigImpl) TopicConfigs(key string) (TopicConfig, bool)
• type SyncProducer
  • func CreateSyncProducer[V any](_ context.Context, topicConfig TopicConfig, configPreset *sarama.Config) (*SyncProducer[V], error)
  • func (p *SyncProducer[E]) Close(ctx context.Context)
  • func (p *SyncProducer[V]) Produce(_ context.Context, key *string, value *V) error
• type TopicConfig
• 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)

Constants

const (
	DefaultKeyKafkaTopicsConfig = "KAFKA_TOPICS_CONFIG"
)

Functions

func NewSha256ScramClient

func NewSha256ScramClient() sarama.SCRAMClient

func NewSha512ScramClient

func NewSha512ScramClient() sarama.SCRAMClient

func ParseTopicConfigs

func ParseTopicConfigs(jsonString string) (map[string]TopicConfig, error)

Types

type Config

type Config struct {
	// contains filtered or unexported fields
}

func NewConfig

func NewConfig() *Config

func (*Config) ConfigItems

func (c *Config) ConfigItems() []auconfigapi.ConfigItem

func (*Config) Obtain

func (c *Config) Obtain(getter func(key string) string)

func (*Config) TopicConfigs

func (c *Config) TopicConfigs() map[string]TopicConfig

type Consumer

type Consumer[E any] struct {
	// contains filtered or unexported fields
}

func CreateConsumer

func CreateConsumer[E any](
	ctx context.Context,
	topicConfig TopicConfig,
	receiveCallback func(context.Context, *string, *E, time.Time) error,
	configPreset *sarama.Config,
) (*Consumer[E], error)

func (*Consumer[E]) Cleanup

func (c *Consumer[E]) Cleanup(sarama.ConsumerGroupSession) error

Cleanup is run at the end of a session, once all ConsumeClaim goroutines have exited

func (*Consumer[E]) Close

func (c *Consumer[E]) Close(ctx context.Context)

func (*Consumer[E]) ConsumeClaim

func (c *Consumer[E]) ConsumeClaim(
	session sarama.ConsumerGroupSession,
	claim sarama.ConsumerGroupClaim,
) error

ConsumeClaim must start a consumer loop of ConsumerGroupClaim's Messages(). Once the Messages() channel is closed, the Handler must finish its processing loop and exit.

func (*Consumer[E]) Setup

func (c *Consumer[E]) Setup(sarama.ConsumerGroupSession) error

Setup is run at the beginning of a new session, before ConsumeClaim

type DefaultConfigImpl

type DefaultConfigImpl struct {
	// contains filtered or unexported fields
}

func NewDefaultConfig

func NewDefaultConfig() *DefaultConfigImpl

func (*DefaultConfigImpl) ConfigItems

func (c *DefaultConfigImpl) ConfigItems() []auconfigapi.ConfigItem

func (*DefaultConfigImpl) ObtainValues

func (c *DefaultConfigImpl) ObtainValues(getter func(string) string) error

func (*DefaultConfigImpl) TopicConfigs

func (c *DefaultConfigImpl) TopicConfigs(key string) (TopicConfig, bool)

type SyncProducer

type SyncProducer[V any] struct {
	// contains filtered or unexported fields
}

func CreateSyncProducer

func CreateSyncProducer[V any](
	_ context.Context,
	topicConfig TopicConfig,
	configPreset *sarama.Config,
) (*SyncProducer[V], error)

func (*SyncProducer[E]) Close

func (p *SyncProducer[E]) Close(ctx context.Context)

func (*SyncProducer[V]) Produce

func (p *SyncProducer[V]) Produce(
	_ context.Context,
	key *string,
	value *V,
) error

type TopicConfig

type TopicConfig struct {
	Topic         string
	Brokers       []string
	Username      string
	Password      string
	ConsumerGroup *string
	AuthType      sarama.SASLMechanism
}

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)