broker

Documentation

Index

• func StartConsuming(ctx context.Context, cfg config.Kafka, consumeFunction ConsumeFunction, ...) error
• type ConsumeFunction
• type Consumer
• type Message
• type MessageType
• type Producer
  • func NewProducer(ctx context.Context, cfg *config.Kafka, logger *zap.Logger) (Producer, error)

Functions

func StartConsuming

func StartConsuming(
	ctx context.Context,
	cfg config.Kafka,
	consumeFunction ConsumeFunction,
	logger *zap.Logger,
) error

Types

type ConsumeFunction

type ConsumeFunction func(ctx context.Context, message *sarama.ConsumerMessage) error

type Consumer

type Consumer interface {
	sarama.ConsumerGroupHandler
}

type Message

type Message struct {
	Type  MessageType
	Value map[string]interface{}
}

type MessageType

type MessageType uint16

const (
	Create MessageType = iota
	Update
	Remove
)

type Producer

type Producer interface {
	CreateBasket(basket *model.Basket)
	UpdateBasket(basket *model.Basket)
	RemoveBasket(basketID uint64)
}

func NewProducer

func NewProducer(
	ctx context.Context,
	cfg *config.Kafka,
	logger *zap.Logger,
) (
	Producer,
	error,
)