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/v2
Additionally, you need to apply DB migration.
package main
import (
"context"
"encoding/json"
"fmt"
"log"
"os"
"time"
"github.com/jackc/pgx/v4/pgxpool"
"golang.org/x/sync/errgroup"
"github.com/vgarvardt/gue/v2"
"github.com/vgarvardt/gue/v2/adapter/pgxv4"
)
type printNameArgs struct {
Name string
}
func main() {
printName := func(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.ConnectConfig(context.Background(), pgxCfg)
if err != nil {
log.Fatal(err)
}
defer pgxPool.Close()
poolAdapter := pgxv4.NewConnPool(pgxPool)
gc := gue.NewClient(poolAdapter)
wm := gue.WorkMap{
"PrintName": printName,
}
// create a pool w/ 2 workers
workers := gue.NewWorkerPool(gc, wm, 2, gue.WithPoolQueue("name_printer"))
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: "PrintName",
Args: args,
}
if err := gc.Enqueue(context.Background(), j); err != nil {
log.Fatal(err)
}
j := &gue.Job{
Type: "PrintName",
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 (
"context"
"log"
"os"
"github.com/jackc/pgx/v4/pgxpool"
"github.com/vgarvardt/gue/v2"
"github.com/vgarvardt/gue/v2/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 := gue.NewClient(poolAdapter)
...
}
package main
import (
"log"
"os"
"github.com/jackc/pgx"
"github.com/vgarvardt/gue/v2"
"github.com/vgarvardt/gue/v2/adapter/pgxv3"
)
func main() {
pgxCfg, err := pgx.ParseURI(os.Getenv("DATABASE_URL"))
if err != nil {
log.Fatal(err)
}
pgxPool, err := pgx.NewConnPool(pgx.ConnPoolConfig{
ConnConfig: pgxCfg,
})
if err != nil {
log.Fatal(err)
}
defer pgxPool.Close()
poolAdapter := pgxv3.NewConnPool(pgxPool)
gc := gue.NewClient(poolAdapter)
...
}
package main
import (
"database/sql"
"log"
"os"
_ "github.com/lib/pq" // register postgres driver
"github.com/vgarvardt/gue/v2"
"github.com/vgarvardt/gue/v2/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 := gue.NewClient(poolAdapter)
...
}
package main
import (
"log"
"os"
"github.com/go-pg/pg/v10"
"github.com/vgarvardt/gue/v2"
"github.com/vgarvardt/gue/v2/adapter/gopgv10"
)
func main() {
opts, err := pg.ParseURL(os.Getenv("DATABASE_URL"))
if err != nil {
log.Fatal(err)
}
db := pg.Connect(opts)
defer db.Close()
poolAdapter := gopgv10.NewConnPool(db)
gc := 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/null
logger - Stdlib
log
- adapter that useslog
logger for logs output. Instantiate it withadapter.NewStdLogger(...)
. - Uber
zap
- adapter that usesgo.uber.org/zap
logger for logs output. Instantiate it withadapter.zap.New(...)
.
Linter and tests are running for every Pull Request, but it is possible to run linter
and tests locally using docker
and make
.
Run linter: make link
. This command runs liner in docker container with the project
source code mounted.
Run tests: make test
. This command runs project dependencies in docker containers
if they are not started yet and runs go tests with coverage.