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/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(...).
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.
React
Typescript
Django
Laravel
Facebook
Microsoft
Google
Alibaba
D3
Tencent