pubsub

Documentation

Index

• func GetSaramaConfig() *sarama.Config
• type KafkaPubSub
  • func (kf *KafkaPubSub) Cleanup(sarama.ConsumerGroupSession) error
  • func (kf *KafkaPubSub) Consume(ctx context.Context) error
  • func (kf *KafkaPubSub) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (kp *KafkaPubSub) Publish(ctx context.Context, message *models.Message) error
  • func (kf *KafkaPubSub) Setup(sarama.ConsumerGroupSession) error
  • func (kf *KafkaPubSub) Shutdown()
  • func (kf *KafkaPubSub) Subscribe(ctx context.Context, topic string, fn ...MessageHandler) error
• type MessageHandler
  • func (fn MessageHandler) GetFunctionName() string
• type PubSub
  • func NewKafkaPubSub(ctx context.Context) PubSub

Functions

func GetSaramaConfig

func GetSaramaConfig() *sarama.Config

Types

type KafkaPubSub

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

func (*KafkaPubSub) Cleanup

func (kf *KafkaPubSub) Cleanup(sarama.ConsumerGroupSession) error

Cleanup is run at the end of a session, once all ConsumeClaim goroutines have exited

func (*KafkaPubSub) Consume

func (kf *KafkaPubSub) Consume(ctx context.Context) error

func (*KafkaPubSub) ConsumeClaim

func (kf *KafkaPubSub) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

ConsumeClaim must start a consumer loop of ConsumerGroupClaim's Messages(). Once the Messages() channel is closed, the Handler must finish its processing loop and exit.

func (*KafkaPubSub) Publish

func (kp *KafkaPubSub) Publish(ctx context.Context, message *models.Message) error

func (*KafkaPubSub) Setup

func (kf *KafkaPubSub) Setup(sarama.ConsumerGroupSession) error

Setup is run at the beginning of a new session, before ConsumeClaim

func (*KafkaPubSub) Shutdown

func (kf *KafkaPubSub) Shutdown()

func (*KafkaPubSub) Subscribe

func (kf *KafkaPubSub) Subscribe(ctx context.Context, topic string, fn ...MessageHandler) error

type MessageHandler

type MessageHandler func(context.Context, *models.Message) error

func (MessageHandler) GetFunctionName

func (fn MessageHandler) GetFunctionName() string

type PubSub

type PubSub interface {
	Publish(ctx context.Context, msg *models.Message) error
	Subscribe(ctx context.Context, topic string, fn ...MessageHandler) error
	Consume(ctx context.Context) error
}

func NewKafkaPubSub

func NewKafkaPubSub(ctx context.Context) PubSub