kafka

Documentation

Index

• type Committer
• type Handler
• type Handlers
• type Header
• type Listener
  • func NewListener(reader Reader, handlers Handlers, opts ...ListenerOption) (*Listener, error)
  • func (l *Listener) Close() error
  • func (l *Listener) Listen(ctx context.Context) error
• type ListenerOption
  • func ListenerAsyncHandle(asyncHandle bool) ListenerOption
  • func ListenerErrorHandler(errHandler transport.ErrorHandler) ListenerOption
  • func ListenerErrorLogger(logger log.Logger) ListenerOption
  • func ListenerManualCommit(manualCommit bool) ListenerOption
• type Message
• type NoopErrorHandler
  • func NewNoopErrorHandler() *NoopErrorHandler
  • func (eh *NoopErrorHandler) Handle(context.Context, error)
• type Reader

Types

type Committer

type Committer interface {
	Commit(ctx context.Context) error
}

type Handler

type Handler interface {
	Handle(ctx context.Context, msg Message) error
}

type Handlers

type Handlers map[string]Handler

Handlers represents Topic -> Handler mapping

type Header

type Header interface {
	Key() string
	Value() []byte
}

type Listener

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

func NewListener

func NewListener(reader Reader, handlers Handlers, opts ...ListenerOption) (*Listener, error)

func (*Listener) Close

func (l *Listener) Close() error

func (*Listener) Listen

func (l *Listener) Listen(ctx context.Context) error

type ListenerOption

type ListenerOption func(*Listener)

func ListenerAsyncHandle

func ListenerAsyncHandle(asyncHandle bool) ListenerOption

func ListenerErrorHandler

func ListenerErrorHandler(errHandler transport.ErrorHandler) ListenerOption

func ListenerErrorLogger

func ListenerErrorLogger(logger log.Logger) ListenerOption

func ListenerManualCommit

func ListenerManualCommit(manualCommit bool) ListenerOption

type Message

type Message interface {
	Topic() string
	Partition() int
	Offset() int64
	Key() []byte
	Value() []byte
	Headers() []Header
	Timestamp() time.Time
}

type NoopErrorHandler

type NoopErrorHandler struct {
}

func NewNoopErrorHandler

func NewNoopErrorHandler() *NoopErrorHandler

func (*NoopErrorHandler) Handle

func (eh *NoopErrorHandler) Handle(context.Context, error)

type Reader

type Reader interface {
	ReadMessage(ctx context.Context) (Message, error)
	Committer
	io.Closer
}