PGPQ

Priority queues with Postgres, implemented in Go.

Example:

import (
	"context"
	"fmt"
	"os"

	"github.com/bsm/pgpq"
	"github.com/google/uuid"
)

func main() {
	ctx := context.Background()

	// connection URL:
	// - use `sslmode=verify-ca` for production
	// - use `default_query_exec_mode=simple_protocol` in combination with proxies
	url := "postgres://localhost/pgpq_test?sslmode=disable"

	if v := os.Getenv("DATABASE_URL"); v != "" {
		url = v
	}

	// connect to postgres
	client, err := pgpq.Connect(ctx, url)
	if err != nil {
		panic(err)
	}
	defer client.Close()

	// truncate the queue, for testing only
	if err := client.Truncate(ctx); err != nil {
		panic(err)
	}

	// push some tasks into the queue
	if err := client.Push(ctx, &pgpq.Task{
		Priority:	3,
		Payload:	[]byte(`{"foo":1}`),
	}); err != nil {
		panic(err)
	}
	if err := client.Push(ctx, &pgpq.Task{
		ID:		uuid.MustParse("28667ce4-1999-4af4-9ff2-1757b3844048"),	// custom UUID
		Priority:	4,
		Payload:	[]byte(`{"bar":2}`),
	}); err != nil {
		panic(err)
	}
	if err := client.Push(ctx, &pgpq.Task{
		Payload: []byte(`{"baz":3}`),
	}); err != nil {
		panic(err)
	}
	if err := client.Push(ctx, &pgpq.Task{
		Payload:	[]byte(`{"baz":4}`),
		NotBefore:	time.Now().Add(time.Minute),	// delay this task for 1m
	}); err != nil {
		panic(err)
	}

	// shift the task with the highest priority from the queue
	claim, err := client.Shift(ctx)
	if err != nil {
		panic(err)
	}
	defer claim.Release(ctx)

	// print ID and payload
	fmt.Println(claim.ID.String())
	fmt.Println(string(claim.Payload))

	// mark task done and remove from the queue
	if err := claim.Done(ctx); err != nil {
		panic(err)
	}

}