message

Documentation

Index

• type Acker
• type Handler
• type InputMessage
  • func (msg InputMessage) Commit(ctx context.Context)
• type Message
• type Middle
• type OutputMessage
• type RecordMiddle
  • func (middle RecordMiddle) Warp(handler Handler) Handler
• type RetryMiddle
  • func (middle RetryMiddle) Warp(handler Handler) Handler
• type Router
  • func NewRouter(option RouterOption, subs ...Subscriber) (*Router, error)
  • func (router *Router) AddHandler(topic string, handler Handler) error
  • func (router *Router) Start(ctx context.Context) error
• type RouterOption
• type Subscriber
• type TimeOutMiddle
  • func (middle TimeOutMiddle) Warp(handler Handler) Handler

Types

type Acker

type Acker interface {
	Commit(ctx context.Context)
}

type Handler

type Handler interface {
	HandleMessage(ctx context.Context, topic string, msg Message) error
	Name() string
}

type InputMessage

type InputMessage struct {
	Topic string
	Message
	Acker Acker
}

func (InputMessage) Commit

func (msg InputMessage) Commit(ctx context.Context)

type Message

type Message struct {
	Key     string
	PayLoad []byte
	Head    map[string]string
}

type Middle

type Middle interface {
	Warp(handler Handler) Handler
}

type OutputMessage

type OutputMessage struct {
	OffsetToken string
	Topic       string
	Message
}

type RecordMiddle

type RecordMiddle struct {
	Records record.Factory
}

func (RecordMiddle) Warp

func (middle RecordMiddle) Warp(handler Handler) Handler

type RetryMiddle

type RetryMiddle struct {
	CheckRetry    func(err error) bool
	MaxRetryCount int
}

func (RetryMiddle) Warp

func (middle RetryMiddle) Warp(handler Handler) Handler

type Router

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

func NewRouter

func NewRouter(option RouterOption, subs ...Subscriber) (*Router, error)

func (*Router) AddHandler

func (router *Router) AddHandler(topic string, handler Handler) error

func (*Router) Start

func (router *Router) Start(ctx context.Context) error

type RouterOption

type RouterOption struct {
	Middles []Middle
}

type Subscriber

type Subscriber interface {
	Topic() string
	Read(ctx context.Context) (InputMessage, error)
	Close() error
}

type TimeOutMiddle

type TimeOutMiddle struct {
	TimeOut time.Duration
}

func (TimeOutMiddle) Warp

func (middle TimeOutMiddle) Warp(handler Handler) Handler