consumer

Documentation

Index

• Constants
• func DefaultIsRetry(msg interface{}, failTimes int) bool
• type Consumer
• type Context
  • func GenerateCTXByByte(ctx context.Context, b []byte, msgType message.MsgType) (Context, error)
  • func (ctx *Context) BindJson(obj any) error
  • func (ctx *Context) Deadline() (deadline time.Time, ok bool)
  • func (ctx *Context) Done() <-chan struct{}
  • func (ctx *Context) Err() error
  • func (ctx *Context) Reset()
  • func (ctx *Context) Value(key any) any
• type Handler
• type IsRetry
  • func IsRetryByFailTimes(ft int) IsRetry
• type RealHandler

Constants

const DefaultRetryTimes = 3

Functions

func DefaultIsRetry

func DefaultIsRetry(msg interface{}, failTimes int) bool

Types

type Consumer

type Consumer interface {
	AddHandler(handlers ...Handler)
	Run()
	Close()
}

type Context

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

func GenerateCTXByByte

func GenerateCTXByByte(ctx context.Context, b []byte, msgType message.MsgType) (Context, error)

func (*Context) BindJson

func (ctx *Context) BindJson(obj any) error

func (*Context) Deadline

func (ctx *Context) Deadline() (deadline time.Time, ok bool)

func (*Context) Done

func (ctx *Context) Done() <-chan struct{}

func (*Context) Err

func (ctx *Context) Err() error

func (*Context) Reset

func (ctx *Context) Reset()

func (*Context) Value

func (ctx *Context) Value(key any) any

type Handler

type Handler struct {
	Topic      string
	Key        string
	Handle     func(ctx Context) error
	PoolSize   int
	Callback   func(msg any, err error)
	RetryTimes int
	IsRetry    IsRetry
	MsgType    message.MsgType
}

type IsRetry

type IsRetry func(msg any, failTimes int) bool

func IsRetryByFailTimes

func IsRetryByFailTimes(ft int) IsRetry

type RealHandler

type RealHandler struct {
	Handle   func(msg any) error
	Callback func(msg any, err error)
	IsRetry  IsRetry
	GP       *gp.Pool
}