kafkasarama

Documentation

Index

• Constants
• func DefaultConfig() *sarama.Config
• type AsyncSender
  • func NewAsyncSender(addr []string, conf *sarama.Config, o ...AsyncSenderOption) (*AsyncSender, error)
  • func (s *AsyncSender) Close() error
  • func (s *AsyncSender) Send(_ context.Context, channel string, h messaging.Header, route string, ...) (err error)
• type AsyncSenderOption
  • func AsyncSenderErrorHook(f ...func(err *sarama.ProducerError)) AsyncSenderOption
  • func AsyncSenderSuccessHook(f ...func(m *sarama.ProducerMessage)) AsyncSenderOption
• type Hook
• type Receiver
  • func NewReceiver(addr []string, conf *sarama.Config, opts ...ReceiverOption) *Receiver
  • func (r *Receiver) Channels() []string
  • func (r *Receiver) Close() (err error)
  • func (r *Receiver) Receive(ctx context.Context, topic string, h messaging.Handler) error
  • func (r *Receiver) Suspend(_ context.Context, topic string) (err error)
• type ReceiverOption
  • func ReceiverErrorHandler(h messaging.ErrorHandler) ReceiverOption
  • func WithGroupPostfix(prefix string) ReceiverOption
  • func WithGroupPrefix(prefix string) ReceiverOption
  • func WithGroupingByChannel() ReceiverOption
  • func WithReBalanceHook(hook ...Hook) ReceiverOption
• type Sender
  • func NewSender(addr []string, conf *sarama.Config) (*Sender, error)
  • func (s *Sender) Close() error
  • func (s *Sender) Send(_ context.Context, channel string, h messaging.Header, route string, ...) (err error)

Constants

const (
	DefaultGroupPostfix = "group"
	DefaultGroupPrefix  = "group"
)

Functions

func DefaultConfig

func DefaultConfig() *sarama.Config

Types

type AsyncSender

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

func NewAsyncSender

func NewAsyncSender(addr []string, conf *sarama.Config, o ...AsyncSenderOption) (*AsyncSender, error)

func (*AsyncSender) Close

func (s *AsyncSender) Close() error

func (*AsyncSender) Send

func (s *AsyncSender) Send(_ context.Context, channel string, h messaging.Header, route string, message []byte) (err error)

type AsyncSenderOption

type AsyncSenderOption func(*AsyncSender)

func AsyncSenderErrorHook

func AsyncSenderErrorHook(f ...func(err *sarama.ProducerError)) AsyncSenderOption

func AsyncSenderSuccessHook

func AsyncSenderSuccessHook(f ...func(m *sarama.ProducerMessage)) AsyncSenderOption

type Hook

type Hook func(sarama.ConsumerGroupSession)

type Receiver

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

func NewReceiver

func NewReceiver(addr []string, conf *sarama.Config, opts ...ReceiverOption) *Receiver

func (*Receiver) Channels

func (r *Receiver) Channels() []string

func (*Receiver) Close

func (r *Receiver) Close() (err error)

func (*Receiver) Receive

func (r *Receiver) Receive(ctx context.Context, topic string, h messaging.Handler) error

func (*Receiver) Suspend

func (r *Receiver) Suspend(_ context.Context, topic string) (err error)

type ReceiverOption

type ReceiverOption func(*Receiver)

func ReceiverErrorHandler

func ReceiverErrorHandler(h messaging.ErrorHandler) ReceiverOption

func WithGroupPostfix

func WithGroupPostfix(prefix string) ReceiverOption

func WithGroupPrefix

func WithGroupPrefix(prefix string) ReceiverOption

func WithGroupingByChannel

func WithGroupingByChannel() ReceiverOption

func WithReBalanceHook

func WithReBalanceHook(hook ...Hook) ReceiverOption

type Sender

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

func NewSender

func NewSender(addr []string, conf *sarama.Config) (*Sender, error)

func (*Sender) Close

func (s *Sender) Close() error

func (*Sender) Send

func (s *Sender) Send(_ context.Context, channel string, h messaging.Header, route string, message []byte) (err error)