queue

package

v1.14.0 Latest Latest Go to latest Published: Nov 20, 2025 License: MIT Imports: 15 Imported by: 0

Details

Valid go.mod file
Redistributable license
Tagged version
Stable version
Learn more about best practices

Repository

gitlab.com/wolfine/libs/app

Links

Open Source Insights

Documentation ¶

Index ¶

Constants
Variables
func NotifyOrDone(ctx context.Context, notify chan<- Announce, msg Announce)
func NotifyOrDoneErr(ctx context.Context, notify chan<- Announce, msg Announce) error
type Announce
type BusAnnounce
type DispatchableQueue
type Handler
type Queue
- func New(pCh rabbitmq.PubConn, sCh rabbitmq.SubConn, reg *Registry, name ...string) (q *Queue)
type Registry
- func NewRegistry() *Registry
- func (r *Registry) Add(task Task, handler Handler)
type Task

Constants ¶

View Source

const (
	TaskContentType    = "application/vnd.msgpack"
	DefaultContentType = "application/json"
)

Variables ¶

View Source

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

Functions ¶

func NotifyOrDone ¶ added in v1.13.3

func NotifyOrDone(ctx context.Context, notify chan<- Announce, msg Announce)

func NotifyOrDoneErr ¶ added in v1.13.3

func NotifyOrDoneErr(ctx context.Context, notify chan<- Announce, msg Announce) error

Types ¶

type Announce ¶ added in v1.3.0

type Announce interface {
	Dump(*jsoniter.Stream)
}

type BusAnnounce ¶ added in v1.3.0

type BusAnnounce interface {
	Announce
	Exchange() *rabbitmq.Exchange
	Headers() map[string]any
	Route() string
}

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, name ...string) (q *Queue)

func (*Queue) DispatchTask ¶

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

func (*Queue) Process ¶

func (q *Queue) Process(ctx context.Context, data []byte, headers map[string]any, _, cType 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(task Task, handler Handler)

type Task ¶

type Task interface {
	GetExpiration() time.Duration
	Reset()
	msgp.Encodable
	msgp.Decodable
}

Source Files ¶

View all Source files

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL