kafka

Documentation

Index

• func InitTopic(bootstrapUrl string, topics ...string) (err error)
• func InitTopicWithPartitionNumber(bootstrapUrl string, partitionNumber int, topics ...string) (err error)
• func NewConsumer(ctx context.Context, config configuration.Config, topic string, ...) error
• func NewConsumerWithFullMessage(ctx context.Context, config configuration.Config, topic string, ...) error
• type KeySeparationBalancer
  • func (this *KeySeparationBalancer) Balance(msg kafka.Message, partitions ...int) (partition int)
• type Producer
  • func NewProducer(ctx context.Context, config configuration.Config, topic string) (*Producer, error)
  • func NewProducerWithBalancer(ctx context.Context, config configuration.Config, topic string, ...) (*Producer, error)
  • func NewProducerWithKeySeparationBalancer(ctx context.Context, config configuration.Config, topic string) (*Producer, error)
  • func (this *Producer) Produce(key string, message []byte) error

Functions

func InitTopic

func InitTopic(bootstrapUrl string, topics ...string) (err error)

func InitTopicWithPartitionNumber

func InitTopicWithPartitionNumber(bootstrapUrl string, partitionNumber int, topics ...string) (err error)

func NewConsumer

func NewConsumer(ctx context.Context, config configuration.Config, topic string, listener func(delivery []byte) error) error

func NewConsumerWithFullMessage

func NewConsumerWithFullMessage(ctx context.Context, config configuration.Config, topic string, listener func(delivery kafka.Message) error) error

Types

type KeySeparationBalancer

type KeySeparationBalancer struct {
	SubBalancer kafka.Balancer
	Seperator   string
}

func (*KeySeparationBalancer) Balance

func (this *KeySeparationBalancer) Balance(msg kafka.Message, partitions ...int) (partition int)

type Producer

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

func NewProducer

func NewProducer(ctx context.Context, config configuration.Config, topic string) (*Producer, error)

func NewProducerWithBalancer

func NewProducerWithBalancer(ctx context.Context, config configuration.Config, topic string, balancer kafka.Balancer) (*Producer, error)

func NewProducerWithKeySeparationBalancer

func NewProducerWithKeySeparationBalancer(ctx context.Context, config configuration.Config, topic string) (*Producer, error)

func (*Producer) Produce

func (this *Producer) Produce(key string, message []byte) error