memory

README

Overview

Package memory is a memory based implementation of the queue

Installation

go get -u -v github.com/gopi-frame/queue/driver/memory

Import

import "github.com/gopi-frame/queue/driver/memory"

Usage

package main

import (
    "github.com/gopi-frame/queue"
    "github.com/gopi-frame/queue/driver/memory"
)

type CustomJob struct {
    queue.Job
}

func (c *CustomJob) Handle() error {
    // do something
    return nil
}

func (c *CustomJob) Failed(err error) {
    // handle failed job
}

func main() {
    q := memory.NewQueue("test")
    q.Enqueue(new(CustomJob))
    q.Enqueue(new(CustomJob))
    q.Enqueue(new(CustomJob))
    fmt.Println("count:", q.Count()) // Output: count: 3
    for {
        if job, ok := q.Dequeue(); ok {
            if err := job.Handle(); err != nil {
                if job.GetQueueable().GetAttempts < job.GetMaxAttempts() {
                    q.Release(job)
                } else {
                    job.Failed(err)
                }
            } else {
                q.Ack(job)
            }
        } else {
            time.Sleep(time.Millisecond * 100)
        }
    }
}

Documentation

Index

• func Open(options map[string]any) (queuecontract.Queue, error)
• type Comparator
  • func (Comparator) Compare(a, b *Job) int
• type Driver
  • func (d Driver) Open(options map[string]any) (queuecontract.Queue, error)
• type Job
  • func NewJob(job queue.Job, queue string) *Job
  • func (d *Job) GetAttempts() int
  • func (d *Job) GetID() string
  • func (d *Job) GetPayload() queue.Job
  • func (d *Job) GetQueue() string
• type Queue
  • func NewQueue(name string) *Queue
  • func (q *Queue) Ack(_ queuecontract.Job)
  • func (q *Queue) Count() int64
  • func (q *Queue) Dequeue() (queuecontract.Job, bool)
  • func (q *Queue) Empty() bool
  • func (q *Queue) Enqueue(job queuecontract.Job) (queuecontract.Job, bool)
  • func (q *Queue) Name() string
  • func (q *Queue) Release(job queuecontract.Job)
  • func (q *Queue) Remove(job queuecontract.Job)

Functions

func Open

func Open(options map[string]any) (queuecontract.Queue, error)

Open is a convenience function that calls Driver.Open.

Types

type Comparator

type Comparator struct{}

func (Comparator) Compare

func (Comparator) Compare(a, b *Job) int

type Driver

type Driver struct {
}

Driver provides memory queue driver

func (Driver) Open

func (d Driver) Open(options map[string]any) (queuecontract.Queue, error)

Open opens memory queue, options must contain "name"

type Job

type Job struct {
	ID          uuid.UUID
	Queue       string
	Payload     queue.Job
	Attempts    int
	AvailableAt time.Time
}

Job is a memory queue job

func NewJob

func NewJob(job queue.Job, queue string) *Job

NewJob creates a new memory queue job

func (*Job) GetAttempts

func (d *Job) GetAttempts() int

GetAttempts returns the job attempts

func (*Job) GetID

func (d *Job) GetID() string

GetID returns the job ID

func (*Job) GetPayload

func (d *Job) GetPayload() queue.Job

GetPayload returns the job payload

func (*Job) GetQueue

func (d *Job) GetQueue() string

GetQueue returns the job queue

type Queue

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

Queue is a memory queue

func NewQueue

func NewQueue(name string) *Queue

NewQueue creates a new memory queue

func (*Queue) Ack

func (q *Queue) Ack(_ queuecontract.Job)

Ack acknowledges a job

func (*Queue) Count

func (q *Queue) Count() int64

Count returns the number of jobs in the queue

func (*Queue) Dequeue

func (q *Queue) Dequeue() (queuecontract.Job, bool)

Dequeue removes a job from the queue

func (*Queue) Empty

func (q *Queue) Empty() bool

Empty returns true if the queue is empty

func (*Queue) Enqueue

func (q *Queue) Enqueue(job queuecontract.Job) (queuecontract.Job, bool)

Enqueue adds a job to the queue

func (*Queue) Name

func (q *Queue) Name() string

Name returns the queue name

func (*Queue) Release

func (q *Queue) Release(job queuecontract.Job)

Release releases a job and adds it back to the queue for the next attempt.

func (*Queue) Remove

func (q *Queue) Remove(job queuecontract.Job)

Remove removes a job from the queue