kafka

package

Go to main page

Versions in this module

v1

v1.9.2

Jun 14, 2024

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