kafka

Documentation

Index

• func New(configuration librepo.Configuration, customConfig config.CustomConfiguration, ...) repository.Kafka
• type Impl
  • func (r *Impl) Connect(ctx context.Context) error
  • func (r *Impl) Disconnect(ctx context.Context) error
  • func (r *Impl) IsKafka() bool
  • func (r *Impl) Level() kgo.LogLevel
  • func (r *Impl) Log(level kgo.LogLevel, msg string, keyvals ...interface{})
  • func (r *Impl) Send(ctx context.Context, event repository.UpdateEvent) error
  • func (r *Impl) Setup() error
  • func (r *Impl) StartReceiveLoop(ctx context.Context) error
  • func (r *Impl) SubscribeIncoming(ctx context.Context, callback repository.ReceiverCallback) error
  • func (r *Impl) Teardown()

Functions

func New

func New(
	configuration librepo.Configuration,
	customConfig config.CustomConfiguration,
	logging librepo.Logging,
	hostIP repository.HostIP,
) repository.Kafka

Types

type Impl

type Impl struct {
	Configuration       librepo.Configuration
	CustomConfiguration config.CustomConfiguration
	Logging             librepo.Logging
	HostIP              repository.HostIP

	Callback      repository.ReceiverCallback
	KafkaProducer *kgo.Client
	KafkaConsumer *kgo.Client
	KafkaTopic    string
}

func (*Impl) Connect

func (r *Impl) Connect(ctx context.Context) error

func (*Impl) Disconnect

func (r *Impl) Disconnect(ctx context.Context) error

func (*Impl) IsKafka

func (r *Impl) IsKafka() bool

func (*Impl) Level

func (r *Impl) Level() kgo.LogLevel

func (*Impl) Log

func (r *Impl) Log(level kgo.LogLevel, msg string, keyvals ...interface{})

func (*Impl) Send

func (r *Impl) Send(ctx context.Context, event repository.UpdateEvent) error

func (*Impl) Setup

func (r *Impl) Setup() error

func (*Impl) StartReceiveLoop

func (r *Impl) StartReceiveLoop(ctx context.Context) error

func (*Impl) SubscribeIncoming

func (r *Impl) SubscribeIncoming(ctx context.Context, callback repository.ReceiverCallback) error

func (*Impl) Teardown

func (r *Impl) Teardown()