kafka

Documentation

Index

• func IsTopicValid(topic string) bool
• func New(logger *zap.Logger, mqCfg messageQueue.Config, routerUrl string) (messageQueue.MessageQueue, error)
• type Factory
  • func (factory *Factory) Create(logger *zap.Logger, mqCfg messageQueue.Config, routerUrl string) (messageQueue.MessageQueue, error)
• type Kafka
  • func (kafka Kafka) Subscribe(trigger *fv1.MessageQueueTrigger) (messageQueue.Subscription, error)
  • func (kafka Kafka) Unsubscribe(subscription messageQueue.Subscription) error
• type MqtConsumer
• type MqtConsumerGroupHandler
  • func NewMqtConsumerGroupHandler(version sarama.KafkaVersion, logger *zap.Logger, ...) MqtConsumerGroupHandler
  • func (ch MqtConsumerGroupHandler) Cleanup(session sarama.ConsumerGroupSession) error
  • func (ch MqtConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (ch MqtConsumerGroupHandler) Setup(session sarama.ConsumerGroupSession) error

Functions

func IsTopicValid

func IsTopicValid(topic string) bool

The validation is based on Kafka's internal implementation: https://github.com/apache/kafka/blob/cde6d18983b5d58199f8857d8d61d7efcbe6e54a/clients/src/main/java/org/apache/kafka/common/internals/Topic.java#L36-L47

func New

func New(logger *zap.Logger, mqCfg messageQueue.Config, routerUrl string) (messageQueue.MessageQueue, error)

Types

type Factory

type Factory struct{}

func (*Factory) Create

func (factory *Factory) Create(logger *zap.Logger, mqCfg messageQueue.Config, routerUrl string) (messageQueue.MessageQueue, error)

type Kafka

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

func (Kafka) Subscribe

func (kafka Kafka) Subscribe(trigger *fv1.MessageQueueTrigger) (messageQueue.Subscription, error)

func (Kafka) Unsubscribe

func (kafka Kafka) Unsubscribe(subscription messageQueue.Subscription) error

type MqtConsumer

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

type MqtConsumerGroupHandler

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

func NewMqtConsumerGroupHandler

func NewMqtConsumerGroupHandler(version sarama.KafkaVersion,
	logger *zap.Logger,
	trigger *fv1.MessageQueueTrigger,
	producer sarama.SyncProducer,
	routerUrl string) MqtConsumerGroupHandler

func (MqtConsumerGroupHandler) Cleanup

func (ch MqtConsumerGroupHandler) Cleanup(session sarama.ConsumerGroupSession) error

Cleanup implemented to satisfy the sarama.ConsumerGroupHandler interface

func (MqtConsumerGroupHandler) ConsumeClaim

func (ch MqtConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

ConsumeClaims implemented to satisfy the sarama.ConsumerGroupHandler interface

func (MqtConsumerGroupHandler) Setup

func (ch MqtConsumerGroupHandler) Setup(session sarama.ConsumerGroupSession) error

Setup implemented to satisfy the sarama.ConsumerGroupHandler interface