kafka

Documentation

Index

• type ClusterAdmin
• type Consumer
• type Decoder
• type Service
  • func NewService(admin ClusterAdmin, consumer Consumer, protoDecoder Decoder) *Service
  • func (s *Service) GetTopics() ([]*models.Topic, error)
  • func (s *Service) Subscribe(ctx context.Context, topic string, key string) (<-chan *models.Record, error)

Types

type ClusterAdmin

type ClusterAdmin interface {
	ListTopics() (map[string]sarama.TopicDetail, error)
}

ClusterAdmin is the kafka cluster admin client

type Consumer

type Consumer interface {
	ConsumePartition(topic string, partition int32, offset int64) (sarama.PartitionConsumer, error)
	Partitions(topic string) ([]int32, error)
}

Consumer is the kafka consumer client

type Decoder

type Decoder interface {
	Decode(ctx context.Context, id uint32, message []byte) (string, error)
}

Decoder can decode a message based on the schema id

type Service

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

Service is the kafka service

func NewService

func NewService(admin ClusterAdmin, consumer Consumer, protoDecoder Decoder) *Service

NewService creates a new kafka service

func (*Service) GetTopics

func (s *Service) GetTopics() ([]*models.Topic, error)

GetTopics returns all the topics in kafka

func (*Service) Subscribe

func (s *Service) Subscribe(ctx context.Context, topic string, key string) (<-chan *models.Record, error)

Subscribe subscribes to a kafka topic by key