kafka

Documentation

Index

• Constants
• type Itemer
• type Producer
  • func NewKafkaProducer(ctx context.Context) (*Producer, error)
  • func (p *Producer) Close()
  • func (p *Producer) CreateProducersPool(chanItem <-chan Itemer) (<-chan Result, <-chan struct{})
• type ProducerProvider
• type Result

Constants

const (
	// TopicShopItems all items will be sent to this topic
	TopicShopItems = "shop_items"
	// TopicShopItemsBidding will recieve only items with bidding set and greater than zero
	TopicShopItemsBidding = "shop_items_bidding"
	// KafkaAddressCtxKey context key for kafka address
	KafkaAddressCtxKey = "addressKafka"
	// MaxProducersCtxKey context key for max numbers of producers
	MaxProducersCtxKey = "kafkaMaxProducers"
)

Types

type Itemer

type Itemer interface {
	GetContext() string
	GetID() string
	Marshal() ([]byte, error)
	Topics() []string
}

Itemer defines interface for processed entities

type Producer

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

Producer for kafka topics

func NewKafkaProducer

func NewKafkaProducer(ctx context.Context) (*Producer, error)

NewKafkaProducer returned configured kafka producer

func (*Producer) Close

func (p *Producer) Close()

Close wrapper for producer provider

func (*Producer) CreateProducersPool

func (p *Producer) CreateProducersPool(chanItem <-chan Itemer) (<-chan Result, <-chan struct{})

CreateProducersPool creates pool of goroutines which will handle populating items to kafka

type ProducerProvider

type ProducerProvider interface {
	Produce(*kafka.Message, chan kafka.Event) error
	Close()
}

ProducerProvider for kafka topics

type Result

type Result struct {
	ItemContext string
	ItemID      string
	Err         error
}

Result indicates message processing status on success - err will be nil on error - err will contain corresponding error