dgqueue

README

D(EFERRED) G(ENERIC) Queue

It was a test task i finished in 2 hours in 2017 year, i polished code a little, created example with contexts and added 100% unit tests coverage in 2023.

What does it do?

With this package we can make deferred queue of tasks to be executed, like execute this in 3 minutes, execute that in 15 seconds from now and so on. Then, we can consume this tasks by concurrent goroutines and they (tasks) will be provided to consumers in proper order, like first task will be that to be executed in 15 seconds from now.

It is worth notice, that any element of queue should be of the same type, and, thanks to God and generics, typecasting is fast. If you need more agile and less performant approach, you can see typecasting based alternative - https://github.com/vodolaz095/dqueue with possibility to make every tasks' payload in queue to have different type.

Basic usage

Make queue handler:

handler := dgqueue.New[string]() // import "github.com/vodolaz095/dgqueue"

something := "task" // payload can be a string only

// Create tasks to be executed in future
handler.ExecuteAt(something, time.Now().Add(time.Minute))
handler.ExecuteAfter(something, time.Minute)

// Extract task ready to be executed
task, ready := handler.Get()
if ready { // task is ready
    fmt.Printf("Task %s is ready to be executed at %s",
		task.Payload, // no need to typecast! 
		task.ExecuteAt.Format(time.Kitchen),
	)
} else {
	fmt.Println("No tasks are ready to be executed")
}
// Count tasks left
tasksInQueue := handler.Len()

// Extract all tasks, so, we can, for example, save all delivery queue  before closing application
tasks:= handler.Dump()

// Prune queue:
handler.Prune()

Concurrent consumers example

See full example at example.go

handler := dgqueue.New[string]() // import "github.com/vodolaz095/dgqueue"

// Publish tasks
something := "task" // payload can be a string only
handler.ExecuteAt(something, time.Now().Add(time.Minute))
handler.ExecuteAfter(something, time.Minute)

// make global context to be canceled when application is stopping
wg := sync.WaitGroup{}
mainCtx, mainCancel := context.WithTimeout(context.Background(), 3*time.Second)
defer mainCancel()

// Start concurrent consumers
wg := sync.WaitGroup{}
for j := 0; j < 10; j++ {
    wg.Add(1)
    go func(workerNumber int, initialCtx context.Context) {
        ctx, cancel := context.WithCancel(initialCtx)
        defer cancel()
        ticker := time.NewTicker(time.Millisecond)
        for {
            select {
            case t := <-ticker.C:
                task, ready := handler.Get()
                if ready { // task is ready
                    err := ProcessTask(task)
                    if err != nil { // aka, requeue message to be delivered in 1 minute
                      handler.ExecuteAfter(something, time.Minute)
                    }
                }
                break
            case <-ctx.Done():
                fmt.Printf("Closing worker %v, there are %v tasks in queue\n", workerNumber, handler.Len())
                wg.Done()
                ticker.Stop()
                return
            }
        }
    }(j, mainCtx)
}
wg.Wait()

// See tasks left, so they can be restored somehow when application is restarted
tasks := handler.Dump()

Documentation

Overview

Package dqueue implements thread safe deferred queue for tasks to be executed in future after some delay

Index

• type Handler
  • func New[T comparable]() Handler[T]
  • func (h *Handler[T]) Dump() (ret []Task[T])
  • func (h *Handler[T]) ExecuteAfter(payload T, after time.Duration) (ok bool)
  • func (h *Handler[T]) ExecuteAt(payload T, when time.Time) (ok bool)
  • func (h *Handler[T]) Get() (task Task[T], ready bool)
  • func (h *Handler[T]) Len() int
  • func (h *Handler[T]) Prune()
• type Task

Types

type Handler

type Handler[T comparable] struct {
	// contains filtered or unexported fields
}

Handler is deferred queue handler used to store and retrieve (in time) tasks to be executed

func New

func New[T comparable]() Handler[T]

New creates new deferred queue Handler.

func (*Handler[T]) Dump

func (h *Handler[T]) Dump() (ret []Task[T])

Dump returns copy of contents of the queue in sorted manner, leaving queue intact.

func (*Handler[T]) ExecuteAfter

func (h *Handler[T]) ExecuteAfter(payload T, after time.Duration) (ok bool)

ExecuteAfter schedules task for execution on after time.Duration provided, it returns true, if task is accepted.

func (*Handler[T]) ExecuteAt

func (h *Handler[T]) ExecuteAt(payload T, when time.Time) (ok bool)

ExecuteAt schedules task for execution on time desired, it returns true, if task is accepted.

func (*Handler[T]) Get

func (h *Handler[T]) Get() (task Task[T], ready bool)

Get extracts one of tasks from queue in thread safe manner. If second argument is true, it means task is ready to be executed and is removed from queue. If there are no ready tasks in queue, first argument is zero Task struct and second one - false.

func (*Handler[T]) Len

func (h *Handler[T]) Len() int

Len is thread save function to see how much tasks are in queue.

func (*Handler[T]) Prune

func (h *Handler[T]) Prune()

Prune resets queue.

type Task

type Task[T comparable] struct {
	ExecuteAt time.Time
	Payload   T
}

Task is element describing things to be done in future