kafka

Documentation

Overview

Package kafka provides a kafka broker using sarama cluster

Index

• type Broker
  • func New(opts ...Option) *Broker
  • func (k *Broker) Close(ctx context.Context) error
  • func (k *Broker) Open(context.Context) error
  • func (k *Broker) Publish(ctx context.Context, topic string, msg *broker.Message, ...) error
  • func (k *Broker) String() string
  • func (k *Broker) Subscribe(ctx context.Context, topic string, handler broker.Handler, ...) (broker.Subscriber, error)
• type JSONCodec
  • func (m JSONCodec) Marshal(v interface{}) ([]byte, error)
  • func (m JSONCodec) Unmarshal(data []byte, v interface{}) error
• type Option
  • func Address(addrs []string) Option
  • func AsyncPublish() Option
  • func Codec(c codec) Option
  • func Config(conf *sarama.Config) Option
  • func Logger(l logger) Option
  • func OnAsyncPublishFailure(f func(*PublishError)) Option
  • func OnAsyncPublishSuccess(f func(*broker.Message)) Option
• type PublishError

Types

type Broker

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

func New

func New(opts ...Option) *Broker

func (*Broker) Close

func (k *Broker) Close(ctx context.Context) error

func (*Broker) Open

func (k *Broker) Open(context.Context) error

func (*Broker) Publish

func (k *Broker) Publish(ctx context.Context, topic string, msg *broker.Message, opts ...broker.PublishOption) error

func (*Broker) String

func (k *Broker) String() string

func (*Broker) Subscribe

func (k *Broker) Subscribe(ctx context.Context, topic string, handler broker.Handler, opts ...broker.SubscribeOption) (broker.Subscriber, error)

type JSONCodec

type JSONCodec struct{}

func (JSONCodec) Marshal

func (m JSONCodec) Marshal(v interface{}) ([]byte, error)

func (JSONCodec) Unmarshal

func (m JSONCodec) Unmarshal(data []byte, v interface{}) error

type Option

type Option func(*Broker)

func Address

func Address(addrs []string) Option

func AsyncPublish

func AsyncPublish() Option

func Codec

func Codec(c codec) Option

func Config

func Config(conf *sarama.Config) Option

func Logger

func Logger(l logger) Option

func OnAsyncPublishFailure

func OnAsyncPublishFailure(f func(*PublishError)) Option

func OnAsyncPublishSuccess

func OnAsyncPublishSuccess(f func(*broker.Message)) Option

type PublishError

type PublishError struct {
	Error   error
	Message *broker.Message
}