queue

Documentation

Index

• Variables
• type DispatchableQueue
• type Handler
• type Queue
  • func New(pCh rabbitmq.PubConn, sCh rabbitmq.SubConn, reg *Registry) (q *Queue)
  • func NewNamed(name string, pCh rabbitmq.PubConn, sCh rabbitmq.SubConn, reg *Registry) (q *Queue)
  • func (q *Queue) DispatchTask(ctx context.Context, t Task) error
  • func (q *Queue) Process(ctx context.Context, data []byte, headers map[string]any, _, _ string) (err error)
  • func (q *Queue) Work(ctx context.Context, errorGroup *errgroup.Group)
• type Registry
  • func NewRegistry() *Registry
  • func (r *Registry) Add(t Task, h Handler)
• type Task

Variables

var (
	ErrUnknownTask = errors.New("unknown task")
	ErrMissingTask = errors.New("missing task type")
)

var AppName = "app"

Types

type DispatchableQueue

type DispatchableQueue interface {
	DispatchTask(ctx context.Context, t Task) error
}

type Handler

type Handler func(context.Context, Task) error

type Queue

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

func New

func New(pCh rabbitmq.PubConn, sCh rabbitmq.SubConn, reg *Registry) (q *Queue)

func NewNamed

func NewNamed(name string, pCh rabbitmq.PubConn, sCh rabbitmq.SubConn, reg *Registry) (q *Queue)

func (*Queue) DispatchTask

func (q *Queue) DispatchTask(ctx context.Context, t Task) error

func (*Queue) Process

func (q *Queue) Process(ctx context.Context, data []byte, headers map[string]any, _, _ string) (err error)

func (*Queue) Work

func (q *Queue) Work(ctx context.Context, errorGroup *errgroup.Group)

type Registry

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

func NewRegistry

func NewRegistry() *Registry

func (*Registry) Add

func (r *Registry) Add(t Task, h Handler)

type Task

type Task interface {
	GetExpiration() time.Duration
	Dump(*jsoniter.Stream)
	Load(*jsoniter.Iterator)
	Reset()
}