Gue is Golang queue on top of PostgreSQL that uses transaction-level locks.
Originally this project used to be a fork of bgentry/que-go but because of some backward-compatibility breaking changes and original library author not being very responsive for PRs I turned fork into standalone project. Version 2 breaks internal backward-compatibility with the original project - DB table and all the internal logic (queries, algorithms) is completely rewritten.
The name Gue is yet another silly word transformation: Queue -> Que, Go + Que -> Gue.
go get -u github.com/vgarvardt/gue/v5Additionally, you need to apply DB migration.
package main
import (
"context"
"encoding/json"
"fmt"
"log"
"os"
"time"
"github.com/jackc/pgx/v5/pgxpool"
"golang.org/x/sync/errgroup"
"github.com/vgarvardt/gue/v5"
"github.com/vgarvardt/gue/v5/adapter/pgxv5"
)
const (
printerQueue = "name_printer"
jobTypePrinter = "PrintName"
)
type printNameArgs struct {
Name string
}
func main() {
printName := func(ctx context.Context, j *gue.Job) error {
var args printNameArgs
if err := json.Unmarshal(j.Args, &args); err != nil {
return err
}
fmt.Printf("Hello %s!\n", args.Name)
return nil
}
pgxCfg, err := pgxpool.ParseConfig(os.Getenv("DATABASE_URL"))
if err != nil {
log.Fatal(err)
}
pgxPool, err := pgxpool.NewWithConfig(context.Background(), pgxCfg)
if err != nil {
log.Fatal(err)
}
defer pgxPool.Close()
poolAdapter := pgxv5.NewConnPool(pgxPool)
gc, err := gue.NewClient(poolAdapter)
if err != nil {
log.Fatal(err)
}
wm := gue.WorkMap{
jobTypePrinter: printName,
}
finishedJobsLog := func(ctx context.Context, j *gue.Job, err error) {
if err != nil {
return
}
j.Tx().Exec(
ctx,
"INSERT INTO finished_jobs_log (queue, type, run_at) VALUES ($1, $2, now())",
j.Queue,
j.Type,
)
}
// create a pool w/ 2 workers
workers, err := gue.NewWorkerPool(gc, wm, 2, gue.WithPoolQueue(printerQueue), gue.WithPoolHooksJobDone(finishedJobsLog))
if err != nil {
log.Fatal(err)
}
ctx, shutdown := context.WithCancel(context.Background())
// work jobs in goroutine
g, gctx := errgroup.WithContext(ctx)
g.Go(func() error {
err := workers.Run(gctx)
if err != nil {
// In a real-world applications, use a better way to shut down
// application on unrecoverable error. E.g. fx.Shutdowner from
// go.uber.org/fx module.
log.Fatal(err)
}
return err
})
args, err := json.Marshal(printNameArgs{Name: "vgarvardt"})
if err != nil {
log.Fatal(err)
}
j := &gue.Job{
Type: jobTypePrinter,
Queue: printerQueue,
Args: args,
}
if err := gc.Enqueue(context.Background(), j); err != nil {
log.Fatal(err)
}
j = &gue.Job{
Type: jobTypePrinter,
Queue: printerQueue,
RunAt: time.Now().UTC().Add(30 * time.Second), // delay 30 seconds
Args: args,
}
if err := gc.Enqueue(context.Background(), j); err != nil {
log.Fatal(err)
}
time.Sleep(30 * time.Second) // wait for while
// send shutdown signal to worker
shutdown()
if err := g.Wait(); err != nil {
log.Fatal(err)
}
}Package supports several PostgreSQL drivers using adapter interface internally. Currently, adapters for the following drivers have been implemented:
package main
import (
"log"
"os"
"github.com/jackc/pgx/v5/pgxpool"
"github.com/vgarvardt/gue/v5"
"github.com/vgarvardt/gue/v5/adapter/pgxv5"
)
func main() {
pgxCfg, err := pgxpool.ParseConfig(os.Getenv("DATABASE_URL"))
if err != nil {
log.Fatal(err)
}
pgxPool, err := pgxpool.NewConfig(context.Background(), pgxCfg)
if err != nil {
log.Fatal(err)
}
defer pgxPool.Close()
poolAdapter := pgxv5.NewConnPool(pgxPool)
gc, err := gue.NewClient(poolAdapter)
...
}package main
import (
"context"
"log"
"os"
"github.com/jackc/pgx/v4/pgxpool"
"github.com/vgarvardt/gue/v5"
"github.com/vgarvardt/gue/v5/adapter/pgxv4"
)
func main() {
pgxCfg, err := pgxpool.ParseConfig(os.Getenv("DATABASE_URL"))
if err != nil {
log.Fatal(err)
}
pgxPool, err := pgxpool.ConnectConfig(context.Background(), pgxCfg)
if err != nil {
log.Fatal(err)
}
defer pgxPool.Close()
poolAdapter := pgxv4.NewConnPool(pgxPool)
gc, err := gue.NewClient(poolAdapter)
...
}package main
import (
"database/sql"
"log"
"os"
_ "github.com/lib/pq" // register postgres driver
"github.com/vgarvardt/gue/v5"
"github.com/vgarvardt/gue/v5/adapter/libpq"
)
func main() {
db, err := sql.Open("postgres", os.Getenv("DATABASE_URL"))
if err != nil {
log.Fatal(err)
}
defer db.Close()
poolAdapter := libpq.NewConnPool(db)
gc, err := gue.NewClient(poolAdapter)
...
}Package supports several logging libraries using adapter interface internally. Currently, adapters for the following drivers have been implemented:
- NoOp (
adapter.NoOpLogger) - default adapter that does nothing, so it is basically/dev/nulllogger - Stdlib
log- adapter that usesloglogger for logs output. Instantiate it withadapter.NewStdLogger(...). - Uber
zap- adapter that usesgo.uber.org/zaplogger for logs output. Instantiate it withadapter/zap.New(...). - Olivier Poitrey's
zerolog- adapter that usesgithub.com/rs/zerologlogger for logs output. Instantiate it withadapter/zerolog.New(...). - (Future) Stdlib
slog- adapter that usesgolang.org/x/exp/sloglogger for logs output. Instantiate it withadapter/slog.New(...). Adapter is expected to be included into stdlib, once it is there - dependency will be changed fromgolang.org/x/exp/slogtolog/slogor whenever it will land.