kafka

type Checkpoint struct {
	Logger  *zap.SugaredLogger
	Init    bool
	Offsets map[string]int64
}

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

type KafkaHandler struct {
	*sync.Mutex
	Logger *zap.SugaredLogger

	// kafka details
	GroupName     string
	Producer      sarama.AsyncProducer
	OffsetManager sarama.OffsetManager
	TriggerTopic  string

	// handler functions
	// one function for each consumed topic, return messages, an
	// offset and an optional function that will in a transaction
	Handlers map[string]func(*sarama.ConsumerMessage) ([]*sarama.ProducerMessage, int64, func())

	// cleanup function
	// used to clear state when consumer group is rebalanced
	Reset func() error
	// contains filtered or unexported fields
}

type KafkaSensor struct {
	*base.Kafka
	*sync.Mutex
	// contains filtered or unexported fields
}

type KafkaTransaction struct {
	Logger *zap.SugaredLogger

	// kafka details
	Producer  sarama.AsyncProducer
	GroupName string
	Topic     string
	Partition int32

	// used to reset the offset and metadata if transaction fails
	ResetOffset   int64
	ResetMetadata string
}

type KafkaTriggerConnection struct {
	*base.KafkaConnection
	KafkaTriggerHandler
	// contains filtered or unexported fields
}

type KafkaTriggerHandler interface {
	common.TriggerConnection
	Name() string
	Ready() bool
	Reset()
	OneAndDone() bool
	DependsOn(*cloudevents.Event) (string, bool)
	Transform(string, *cloudevents.Event) (*cloudevents.Event, error)
	Filter(string, *cloudevents.Event) bool
	Update(event *cloudevents.Event, partition int32, offset int64, timestamp time.Time) ([]*cloudevents.Event, error)
	Offset(int32, int64) int64
	Action([]*cloudevents.Event) func()
}

type Parameters map[string]bool

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

type TriggerWithDepName struct {
	KafkaTriggerHandler
	// contains filtered or unexported fields
}

type Triggers map[string]KafkaTriggerHandler