kafka

Documentation

Index

• func InitTopic(bootstrapUrl string, topics ...string) (err error)
• type Consumer
  • func NewConsumer(broker string, groupid string, topic string, debug bool, ...) (consumer *Consumer, err error)
  • func (this *Consumer) Restart()
  • func (this *Consumer) Stop()
• type Interface
  • func Init(broker string, group string, debug bool) (Interface, error)
• type Kafka
  • func (this *Kafka) Close()
  • func (this *Kafka) Consume(topic string, listener func(delivery []byte) error) (err error)
  • func (this *Kafka) EnsureTopic(bootstrapUrl string, topic string, config map[string]string) (err error)
  • func (this *Kafka) Publish(topic string, key string, payload []byte) error
• type Publisher
  • func NewPublisher(broker string, topic string, debug bool) (*Publisher, error)
  • func (this *Publisher) Publish(topic string, key string, payload []byte) (err error)

Functions

func InitTopic

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

Types

type Consumer

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

func NewConsumer

func NewConsumer(broker string, groupid string, topic string, debug bool, listener func(topic string, msg []byte) error, errorhandler func(err error, consumer *Consumer)) (consumer *Consumer, err error)

func (*Consumer) Restart

func (this *Consumer) Restart()

func (*Consumer) Stop

func (this *Consumer) Stop()

type Interface

type Interface interface {
	Consume(topic string, listener func(delivery []byte) error) (err error)
	Publish(topic string, key string, payload []byte) error
	Close()
	EnsureTopic(bootstrapUrl string, topic string, config map[string]string) (err error)
}

func Init

func Init(broker string, group string, debug bool) (Interface, error)

type Kafka

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

func (*Kafka) Close

func (this *Kafka) Close()

func (*Kafka) Consume

func (this *Kafka) Consume(topic string, listener func(delivery []byte) error) (err error)

func (*Kafka) EnsureTopic

func (this *Kafka) EnsureTopic(bootstrapUrl string, topic string, config map[string]string) (err error)

func (*Kafka) Publish

func (this *Kafka) Publish(topic string, key string, payload []byte) error

type Publisher

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

func NewPublisher

func NewPublisher(broker string, topic string, debug bool) (*Publisher, error)

func (*Publisher) Publish

func (this *Publisher) Publish(topic string, key string, payload []byte) (err error)