queue

Documentation

Index

• type Consumer
  • func NewKafkaConsumer(cfg KafkaConfig) (Consumer, error)
• type KafkaConfig
• type Producer
  • func NewKafkaProducer(cfg KafkaConfig) (Producer, error)
• type Queue
  • func NewQueue(c Consumer, p Producer) *Queue

Types

type Consumer

type Consumer interface {
	StartConsumer(func(v interface{}) error) error
	StopConsumer()
}

Consumer is an abstraction over the queue impl

func NewKafkaConsumer

func NewKafkaConsumer(cfg KafkaConfig) (Consumer, error)

NewKafkaConsumer creates a new kafka consumer

type KafkaConfig

type KafkaConfig struct {
	EventsTopic   string `envconfig:"KAFKA_EVENTS_TOPIC"`
	CommandsTopic string `envconfig:"KAFKA_COMMANDS_TOPIC"`
	Brokers       string `envconfig:"KAFKA_BROKERS"`
	GroupID       string `envconfig:"KAFKA_GROUPID"`
	Username      string `envconfig:"KAFKA_USERNAME"`
	Password      string `envconfig:"KAFKA_PASSWORD"`
}

type Producer

type Producer interface {
	SendCommand(*model.CommandParams) (*model.Command, error)
}

Producer is an abstraction over the queue impl

func NewKafkaProducer

func NewKafkaProducer(cfg KafkaConfig) (Producer, error)

NewKafkaProducer creates a new kafka producer

type Queue

type Queue struct {
	Producer Producer
	Consumer Consumer
}

func NewQueue

func NewQueue(c Consumer, p Producer) *Queue