pubsub

Documentation

Index

• func DefaultConsumerConfig(brokers []string, consumerGroup string, topics []string) kafka.ReaderConfig
• type Balancer
  • func (r Balancer) AssignGroups(members []kafka.GroupMember, topicPartitions []kafka.Partition) kafka.GroupMemberAssignments
  • func (r Balancer) ProtocolName() string
  • func (r Balancer) UserData() ([]byte, error)
• type KafkaPublisher
  • func NewKafkaPublisher(brokers []string) KafkaPublisher
• type KafkaSubscriber
  • func NewKafkaSubscriber(config kafka.ReaderConfig) KafkaSubscriber
• type OnKafkaMessage

Functions

func DefaultConsumerConfig

func DefaultConsumerConfig(brokers []string, consumerGroup string, topics []string) kafka.ReaderConfig

Types

type Balancer

type Balancer struct{}

func (Balancer) AssignGroups

func (r Balancer) AssignGroups(members []kafka.GroupMember, topicPartitions []kafka.Partition) kafka.GroupMemberAssignments

func (Balancer) ProtocolName

func (r Balancer) ProtocolName() string

func (Balancer) UserData

func (r Balancer) UserData() ([]byte, error)

type KafkaPublisher

type KafkaPublisher interface {
	Publish(ctx context.Context, messages ...kafka.Message) error
	PublishMetric(ctx context.Context, topic, measurement string, tags map[string]string, fields map[string]interface{}, tm time.Time) error
	Close() error
}

func NewKafkaPublisher

func NewKafkaPublisher(brokers []string) KafkaPublisher

type KafkaSubscriber

type KafkaSubscriber interface {
	Subscribe(ctx context.Context, onMessage OnKafkaMessage) error
	Close() error
}

func NewKafkaSubscriber

func NewKafkaSubscriber(config kafka.ReaderConfig) KafkaSubscriber

type OnKafkaMessage

type OnKafkaMessage func(message kafka.Message)