kafka

package

Go to main page

Versions in this module

v1

v1.9.1

Feb 13, 2024

v1.9.0

Jan 9, 2024

v1.8.1

Aug 31, 2023

Changes in this version

type KafkaHandler

+ Reset func() error

type KafkaSensor

+ func (s *KafkaSensor) Reset() error

type KafkaTriggerConnection

+ func (c *KafkaTriggerConnection) Reset()

type KafkaTriggerHandler

+ Reset func()

v1.8.0

May 8, 2023

Changes in this version

+ type Checkpoint struct

+ Init bool

+ Logger *zap.SugaredLogger

+ Offsets map[string]int64

+ func (c *Checkpoint) Metadata() string

+ func (c *Checkpoint) Set(key string, offset int64)

+ func (c *Checkpoint) Skip(key string, offset int64) bool

+ type Checkpoints map[string]map[int32]*Checkpoint

+ type KafkaHandler struct

+ GroupName string

+ Handlers map[string]func(*sarama.ConsumerMessage) ([]*sarama.ProducerMessage, int64, func())

+ Logger *zap.SugaredLogger

+ OffsetManager sarama.OffsetManager

+ Producer sarama.AsyncProducer

+ TriggerTopic string

+ func (h *KafkaHandler) Cleanup(session sarama.ConsumerGroupSession) error

+ func (h *KafkaHandler) Close() error

+ func (h *KafkaHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

+ func (h *KafkaHandler) Setup(session sarama.ConsumerGroupSession) error

+ type KafkaSensor struct

+ func NewKafkaSensor(kafkaConfig *eventbusv1alpha1.KafkaBus, sensor *sensorv1alpha1.Sensor, ...) *KafkaSensor

+ func (s *KafkaSensor) Action(msg *sarama.ConsumerMessage) ([]*sarama.ProducerMessage, int64, func())

+ func (s *KafkaSensor) Close() error

+ func (s *KafkaSensor) Connect(ctx context.Context, triggerName string, depExpression string, ...) (eventbuscommon.TriggerConnection, error)

+ func (s *KafkaSensor) Disconnect()

+ func (s *KafkaSensor) Event(msg *sarama.ConsumerMessage) ([]*sarama.ProducerMessage, int64, func())

+ func (s *KafkaSensor) Initialize() error

+ func (s *KafkaSensor) IsClosed() bool

+ func (s *KafkaSensor) Listen(ctx context.Context)

+ func (s *KafkaSensor) Trigger(msg *sarama.ConsumerMessage) ([]*sarama.ProducerMessage, int64, func())

+ type KafkaTransaction struct

+ GroupName string

+ Logger *zap.SugaredLogger

+ Partition int32

+ Producer sarama.AsyncProducer

+ ResetMetadata string

+ ResetOffset int64

+ Topic string

+ func (t *KafkaTransaction) Commit(session sarama.ConsumerGroupSession, messages []*sarama.ProducerMessage, ...) error

+ type KafkaTriggerConnection struct

+ func (c *KafkaTriggerConnection) Action(events []*cloudevents.Event) func()

+ func (c *KafkaTriggerConnection) Close() error

+ func (c *KafkaTriggerConnection) DependsOn(event *cloudevents.Event) (string, bool)

+ func (c *KafkaTriggerConnection) Filter(depName string, event *cloudevents.Event) bool

+ func (c *KafkaTriggerConnection) IsClosed() bool

+ func (c *KafkaTriggerConnection) Name() string

+ func (c *KafkaTriggerConnection) Offset(partition int32, offset int64) int64

+ func (c *KafkaTriggerConnection) OneAndDone() bool

+ func (c *KafkaTriggerConnection) Ready() bool

+ func (c *KafkaTriggerConnection) String() string

+ func (c *KafkaTriggerConnection) Subscribe(ctx context.Context, closeCh <-chan struct{}, ...) error

+ func (c *KafkaTriggerConnection) Transform(depName string, event *cloudevents.Event) (*cloudevents.Event, error)

+ func (c *KafkaTriggerConnection) Update(event *cloudevents.Event, partition int32, offset int64, timestamp time.Time) ([]*cloudevents.Event, error)

+ type KafkaTriggerHandler interface

+ Action func([]*cloudevents.Event) func()

+ DependsOn func(*cloudevents.Event) (string, bool)

+ Filter func(string, *cloudevents.Event) bool

+ Name func() string

+ Offset func(int32, int64) int64

+ OneAndDone func() bool

+ Ready func() bool

+ Transform func(string, *cloudevents.Event) (*cloudevents.Event, error)

+ Update func(event *cloudevents.Event, partition int32, offset int64, timestamp time.Time) ([]*cloudevents.Event, error)

+ type Parameters map[string]bool

+ func (p Parameters) Get(name string) (interface{}, error)

+ type Topics struct

+ func (t *Topics) List() []string

+ type TriggerWithDepName struct

+ type Triggers map[string]KafkaTriggerHandler

+ func (t Triggers) List(event *cloudevents.Event) []*TriggerWithDepName

+ func (t Triggers) Ready() bool

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL