mirror of
https://github.com/acaloiaro/neoq
synced 2026-07-21 18:29:08 +00:00
Adriano Caloiaro
GitHub
commit
acbba8d214
11 changed files with 99 additions and 70 deletions
15
README.md
15
README.md
|
|
@ -42,8 +42,9 @@ Queue Handlers are simple Go functions that accept a `Context` parameter.
|
|||
**Example**: Add a listener on the `hello_world` queue
|
||||
|
||||
```go
|
||||
nq, _ := neoq.New(neoq.ConnectionString("postgres://postgres:postgres@localhost:5432/neoq"))
|
||||
nq.Listen("hello_world", neoq.NewHandler(func(ctx context.Context) (err error) {
|
||||
nq, err := neoq.New(ctx, neoq.ConnectionString("postgres://postgres:postgres@localhost:5432/neoq"))
|
||||
handleErr(err)
|
||||
nq.Listen(ctx, "hello_world", neoq.NewHandler(func(ctx context.Context) (err error) {
|
||||
j, err := neoq.JobFromContext(ctx)
|
||||
log.Println("got job id:", j.ID, "messsage:", j.Payload["message"])
|
||||
return
|
||||
|
|
@ -55,14 +56,16 @@ nq.Listen("hello_world", neoq.NewHandler(func(ctx context.Context) (err error) {
|
|||
**Example**: Add a "Hello World" job to the `hello_world` queue
|
||||
|
||||
```go
|
||||
nq, _ := neoq.New(neoq.ConnectionString("postgres://postgres:postgres@localhost:5432/neoq"))
|
||||
jid, _ := nq.Enqueue(neoq.Job{
|
||||
nq, err := neoq.New(ctx, neoq.ConnectionString("postgres://postgres:postgres@localhost:5432/neoq"))
|
||||
handleErr(err)
|
||||
|
||||
jid, err := nq.Enqueue(ctx, neoq.Job{
|
||||
Queue: "hello_world",
|
||||
Payload: map[string]interface{}{
|
||||
"message": "hello world",
|
||||
},
|
||||
})
|
||||
|
||||
handleErr(err)
|
||||
```
|
||||
|
||||
# Example Code
|
||||
|
|
@ -72,5 +75,3 @@ Additional example integration code can be found at https://github.com/acaloiaro
|
|||
# Status
|
||||
|
||||
This project is currently in alpha. Future releases may change the API. It currently leaks some resources. It can handle unimportant workloads.
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -1,6 +1,7 @@
|
|||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log"
|
||||
"time"
|
||||
|
||||
|
|
@ -9,10 +10,14 @@ import (
|
|||
|
||||
func main() {
|
||||
const queue = "foobar"
|
||||
nq, _ := neoq.New(neoq.PgTransactionTimeoutOpt(1000))
|
||||
ctx := context.Background()
|
||||
nq, err := neoq.New(ctx, neoq.PgTransactionTimeoutOpt(1000))
|
||||
if err != nil {
|
||||
log.Fatalf("error initializing neoq: %v", err)
|
||||
}
|
||||
|
||||
// Add a job that will execute 1 hour from now
|
||||
jobID, err := nq.Enqueue(neoq.Job{
|
||||
jobID, err := nq.Enqueue(ctx, neoq.Job{
|
||||
Queue: queue,
|
||||
Payload: map[string]interface{}{
|
||||
"message": "hello, future world",
|
||||
|
|
@ -20,7 +25,7 @@ func main() {
|
|||
RunAfter: time.Now().Add(1 * time.Hour),
|
||||
})
|
||||
if err != nil {
|
||||
log.Println("error adding job", err)
|
||||
log.Fatalf("error adding job: %v", err)
|
||||
}
|
||||
|
||||
log.Println("added job:", jobID)
|
||||
|
|
|
|||
|
|
@ -10,8 +10,12 @@ import (
|
|||
func main() {
|
||||
var err error
|
||||
const queue = "foobar"
|
||||
ctx := context.Background()
|
||||
//
|
||||
nq, _ := neoq.New("postgres://postgres:postgres@127.0.0.1:5432/neoq?sslmode=disable")
|
||||
nq, err := neoq.New(ctx, neoq.ConnectionString("postgres://postgres:postgres@127.0.0.1:5432/neoq?sslmode=disable"))
|
||||
if err != nil {
|
||||
log.Fatalf("error initializing neoq: %v", err)
|
||||
}
|
||||
|
||||
// we use a done channel here to make sure that our test doesn't exit before the job finishes running
|
||||
// this is probably not a pattern you want to use in production jobs and you see it here only for testing reasons
|
||||
|
|
@ -30,13 +34,13 @@ func main() {
|
|||
// Option 2: Set options after the handler is created
|
||||
handler = handler.WithOption(neoq.HandlerConcurrencyOpt(8))
|
||||
|
||||
err = nq.Listen(queue, handler)
|
||||
err = nq.Listen(ctx, queue, handler)
|
||||
if err != nil {
|
||||
log.Println("error listening to queue", err)
|
||||
}
|
||||
|
||||
// enqueue a job
|
||||
_, err = nq.Enqueue(neoq.Job{
|
||||
_, err = nq.Enqueue(ctx, neoq.Job{
|
||||
Queue: queue,
|
||||
Payload: map[string]interface{}{
|
||||
"message": "hello, world",
|
||||
|
|
|
|||
|
|
@ -11,11 +11,15 @@ import (
|
|||
func main() {
|
||||
var err error
|
||||
const queue = "foobar"
|
||||
ctx := context.Background()
|
||||
|
||||
// by default neoq connects to a local postgres server using: [neoq.DefaultPgConnectionString]
|
||||
// connection strings can be set explicitly as follows:
|
||||
// neoq.New(neoq.ConnectionString("postgres://username:passsword@hostname/database"))
|
||||
nq, _ := neoq.New()
|
||||
nq, err := neoq.New(ctx)
|
||||
if err != nil {
|
||||
log.Fatalf("error initializing neoq: %v", err)
|
||||
}
|
||||
|
||||
// we use a done channel here to make sure that our test doesn't exit before the job finishes running
|
||||
// this is probably not a pattern you want to use in production jobs and you see it here only for testing reasons
|
||||
|
|
@ -33,12 +37,12 @@ func main() {
|
|||
// this 10ms deadline will cause our job that sleeps for 1s to fail
|
||||
handler = handler.WithOption(neoq.HandlerDeadlineOpt(10 * time.Millisecond))
|
||||
|
||||
err = nq.Listen(queue, handler)
|
||||
err = nq.Listen(ctx, queue, handler)
|
||||
if err != nil {
|
||||
log.Println("error listening to queue", err)
|
||||
}
|
||||
|
||||
_, err = nq.Enqueue(neoq.Job{
|
||||
_, err = nq.Enqueue(ctx, neoq.Job{
|
||||
Queue: queue,
|
||||
Payload: map[string]interface{}{
|
||||
"message": "hello, world",
|
||||
|
|
|
|||
|
|
@ -9,10 +9,15 @@ import (
|
|||
)
|
||||
|
||||
func main() {
|
||||
ctx := context.Background()
|
||||
// by default neoq connects to a local postgres server using: [neoq.DefaultPgConnectionString]
|
||||
// connection strings can be set explicitly as follows:
|
||||
// neoq.New(neoq.ConnectionString("postgres://username:passsword@hostname/database"))
|
||||
nq, _ := neoq.New(neoq.PgTransactionTimeoutOpt(1000))
|
||||
nq, err := neoq.New(ctx, neoq.PgTransactionTimeoutOpt(1000))
|
||||
if err != nil {
|
||||
log.Fatalf("error initializing neoq: %v", err)
|
||||
}
|
||||
|
||||
// run a job periodically
|
||||
handler := neoq.NewHandler(func(ctx context.Context) (err error) {
|
||||
log.Println("running periodic job")
|
||||
|
|
@ -22,7 +27,7 @@ func main() {
|
|||
WithOption(neoq.HandlerDeadlineOpt(time.Duration(500 * time.Millisecond))).
|
||||
WithOption(neoq.HandlerConcurrencyOpt(1))
|
||||
|
||||
nq.ListenCron("* * * * * *", handler)
|
||||
nq.ListenCron(ctx, "* * * * * *", handler)
|
||||
|
||||
time.Sleep(5 * time.Second)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -10,8 +10,12 @@ import (
|
|||
func main() {
|
||||
var err error
|
||||
const queue = "foobar"
|
||||
ctx := context.Background()
|
||||
//
|
||||
nq, _ := neoq.New(neoq.ConnectionString("postgres://postgres:postgres@127.0.0.1:5432/neoq"))
|
||||
nq, err := neoq.New(ctx, neoq.ConnectionString("postgres://postgres:postgres@127.0.0.1:5432/neoq"))
|
||||
if err != nil {
|
||||
log.Fatalf("error initializing neoq: %v", err)
|
||||
}
|
||||
|
||||
handler := neoq.NewHandler(func(ctx context.Context) (err error) {
|
||||
var j *neoq.Job
|
||||
|
|
@ -20,7 +24,7 @@ func main() {
|
|||
return
|
||||
})
|
||||
|
||||
err = nq.Listen(queue, handler)
|
||||
err = nq.Listen(ctx, queue, handler)
|
||||
if err != nil {
|
||||
log.Println("error listening to queue", err)
|
||||
}
|
||||
|
|
|
|||
6
go.mod
6
go.mod
|
|
@ -4,18 +4,18 @@ go 1.19
|
|||
|
||||
require (
|
||||
github.com/guregu/null v4.0.0+incompatible
|
||||
github.com/iancoleman/strcase v0.2.0
|
||||
github.com/jackc/pgx/v5 v5.2.0
|
||||
github.com/jsuar/go-cron-descriptor v0.1.0
|
||||
github.com/pkg/errors v0.9.1
|
||||
github.com/robfig/cron v1.2.0
|
||||
golang.org/x/exp v0.0.0-20230213192124-5e25df0256eb
|
||||
)
|
||||
|
||||
require (
|
||||
github.com/iancoleman/strcase v0.2.0 // indirect
|
||||
github.com/jackc/pgpassfile v1.0.0 // indirect
|
||||
github.com/jackc/pgservicefile v0.0.0-20200714003250-2b9c44734f2b // indirect
|
||||
github.com/jackc/puddle/v2 v2.1.2 // indirect
|
||||
github.com/jsuar/go-cron-descriptor v0.1.0 // indirect
|
||||
github.com/robfig/cron v1.2.0 // indirect
|
||||
go.uber.org/atomic v1.10.0 // indirect
|
||||
go.uber.org/multierr v1.5.0 // indirect
|
||||
go.uber.org/zap v1.15.0 // indirect
|
||||
|
|
|
|||
7
go.sum
7
go.sum
|
|
@ -1,3 +1,4 @@
|
|||
github.com/BurntSushi/toml v0.3.1 h1:WXkYYl6Yr3qBf1K79EBnL4mak0OimBfB0XUf9Vl28OQ=
|
||||
github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
|
||||
github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
|
||||
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
|
|
@ -40,6 +41,7 @@ go.uber.org/atomic v1.10.0 h1:9qC72Qh0+3MqyJbAn8YU5xVq1frD8bn3JtD2oXtafVQ=
|
|||
go.uber.org/atomic v1.10.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0=
|
||||
go.uber.org/multierr v1.5.0 h1:KCa4XfM8CWFCpxXRGok+Q0SS/0XBhMDbHHGABQLvD2A=
|
||||
go.uber.org/multierr v1.5.0/go.mod h1:FeouvMocqHpRaaGuG9EjoKcStLC43Zu/fmqdUMPcKYU=
|
||||
go.uber.org/tools v0.0.0-20190618225709-2cfd321de3ee h1:0mgffUl7nfd+FpvXMVz4IDEaUSmT1ysygQC7qYo7sG4=
|
||||
go.uber.org/tools v0.0.0-20190618225709-2cfd321de3ee/go.mod h1:vJERXedbb3MVM5f9Ejo0C68/HhF8uaILCdgjnY+goOA=
|
||||
go.uber.org/zap v1.15.0 h1:ZZCA22JRF2gQE5FoNmhmrf7jeJJ2uhqDUNRYKm8dvmM=
|
||||
go.uber.org/zap v1.15.0/go.mod h1:Mb2vm2krFEG5DV0W9qcHBYFtp/Wku1cvYaqPsS/WYfc=
|
||||
|
|
@ -49,8 +51,10 @@ golang.org/x/crypto v0.0.0-20220829220503-c86fa9a7ed90 h1:Y/gsMcFOcR+6S6f3YeMKl5
|
|||
golang.org/x/crypto v0.0.0-20220829220503-c86fa9a7ed90/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
|
||||
golang.org/x/exp v0.0.0-20230213192124-5e25df0256eb h1:PaBZQdo+iSDyHT053FjUCgZQ/9uqVwPOcl7KSWhKn6w=
|
||||
golang.org/x/exp v0.0.0-20230213192124-5e25df0256eb/go.mod h1:CxIveKay+FTh1D0yPZemJVgC/95VzuuOLq5Qi4xnoYc=
|
||||
golang.org/x/lint v0.0.0-20190930215403-16217165b5de h1:5hukYrvBGR8/eNkX5mdUezrA6JiaEZDtJb9Ei+1LlBs=
|
||||
golang.org/x/lint v0.0.0-20190930215403-16217165b5de/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
|
||||
golang.org/x/mod v0.0.0-20190513183733-4bf6d317e70e/go.mod h1:mXi4GBBbnImb6dmsKGUJ2LatrhH/nqhxcFungHvyanc=
|
||||
golang.org/x/mod v0.6.0 h1:b9gGHsz9/HhJ3HF5DHQytPpuwocVTChQJK3AvoLRD5I=
|
||||
golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
|
||||
golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
|
||||
golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
|
||||
|
|
@ -59,6 +63,7 @@ golang.org/x/sync v0.0.0-20220923202941-7f9b1623fab7 h1:ZrnxWX62AgTKOSagEqxvb3ff
|
|||
golang.org/x/sync v0.0.0-20220923202941-7f9b1623fab7/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
|
||||
golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
|
||||
golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.1.0 h1:kunALQeHf1/185U1i0GOB/fy1IPRDDpuoOOqRReG57U=
|
||||
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
|
||||
golang.org/x/text v0.3.8 h1:nAL+RVCQ9uMn3vJZbV+MRnydTJFPf8qqY42YiA6MrqY=
|
||||
golang.org/x/text v0.3.8/go.mod h1:E6s5w1FMmriuDzIBO73fBruAKo1PCIq6d2Q6DHfQ8WQ=
|
||||
|
|
@ -66,10 +71,12 @@ golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3
|
|||
golang.org/x/tools v0.0.0-20190621195816-6e04913cbbac/go.mod h1:/rFqwRUd4F7ZHNgwSSTFct+R/Kf4OFW1sUzUTQQTgfc=
|
||||
golang.org/x/tools v0.0.0-20191029041327-9cc4af7d6b2c/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
|
||||
golang.org/x/tools v0.0.0-20191029190741-b9c20aec41a5/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
|
||||
golang.org/x/tools v0.2.0 h1:G6AHpWxTMGY1KyEYoAQ5WTtIekUUvDNjan3ugu60JvE=
|
||||
golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
|
||||
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
|
||||
gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
|
||||
gopkg.in/errgo.v2 v2.1.0/go.mod h1:hNsd1EY+bozCKY1Ytp96fpM3vjJbqLJn88ws8XvfDNI=
|
||||
gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
|
||||
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
|
||||
honnef.co/go/tools v0.0.1-2019.2.3 h1:3JgtbtFHMiCmsznwGVTUWbgGov+pVqnlf1dEJTNAXeM=
|
||||
honnef.co/go/tools v0.0.1-2019.2.3/go.mod h1:a3bituU0lyd329TUQxRnasdCoJDkEUEAqEt0JzvZhAg=
|
||||
|
|
|
|||
37
neoq.go
37
neoq.go
|
|
@ -29,12 +29,13 @@ type contextKey int
|
|||
var varsKey contextKey
|
||||
|
||||
const (
|
||||
JobStatusNew = "new"
|
||||
JobStatusProcessed = "processed"
|
||||
JobStatusFailed = "failed"
|
||||
JobStatusNew = "new"
|
||||
JobStatusProcessed = "processed"
|
||||
JobStatusFailed = "failed"
|
||||
|
||||
DefaultPgConnectionString = "postgres://postgres:postgres@127.0.0.1:5432/neoq"
|
||||
DefaultTransactionTimeout = 60000 //ms
|
||||
DefaultHandlerDeadline = 30000 //ms
|
||||
DefaultTransactionTimeout = time.Minute
|
||||
DefaultHandlerDeadline = 30 * time.Second
|
||||
DuplicateJobID = -1
|
||||
postgresBackendName = "postgres"
|
||||
|
||||
|
|
@ -65,18 +66,18 @@ const (
|
|||
// Neoq interface is Neoq's primary API
|
||||
type Neoq interface {
|
||||
// Enqueue queues jobs to be executed asynchronously
|
||||
Enqueue(job Job) (jobID int64, err error)
|
||||
Enqueue(ctx context.Context, job Job) (jobID int64, err error)
|
||||
|
||||
// Listen listens for jobs on a queue and processes them with the given handler
|
||||
Listen(queue string, h Handler) (err error)
|
||||
Listen(ctx context.Context, queue string, h Handler) (err error)
|
||||
|
||||
// ListenCron listens for jobs on a cron schedule and processes them with the given handler
|
||||
//
|
||||
// See: https://pkg.go.dev/github.com/robfig/cron?#hdr-CRON_Expression_Format for details on the cron spec format
|
||||
ListenCron(cron string, h Handler) (err error)
|
||||
ListenCron(ctx context.Context, cron string, h Handler) (err error)
|
||||
|
||||
// Shutdown halts the worker
|
||||
Shutdown() error
|
||||
Shutdown(ctx context.Context) error
|
||||
|
||||
// WithConfigOpt configures neoq with with optional configuration
|
||||
WithConfigOpt(opt ConfigOption) Neoq
|
||||
|
|
@ -99,9 +100,9 @@ type HandlerFunc func(ctx context.Context) error
|
|||
|
||||
// Handler handles jobs on a queue
|
||||
type Handler struct {
|
||||
deadline time.Duration
|
||||
handle HandlerFunc
|
||||
concurrency int
|
||||
deadline time.Duration
|
||||
}
|
||||
|
||||
// HandlerOption is function that sets optional configuration for Handlers
|
||||
|
|
@ -143,7 +144,7 @@ func NewHandler(f HandlerFunc, opts ...HandlerOption) (h Handler) {
|
|||
|
||||
// always set a job deadline if none is set
|
||||
if h.deadline == 0 {
|
||||
h.deadline = time.Duration(DefaultHandlerDeadline * time.Millisecond)
|
||||
h.deadline = DefaultHandlerDeadline
|
||||
}
|
||||
|
||||
return
|
||||
|
|
@ -172,7 +173,7 @@ type Job struct {
|
|||
}
|
||||
|
||||
// New creates a new Neoq instance for listening to queues and enqueing new jobs
|
||||
func New(opts ...ConfigOption) (n Neoq, err error) {
|
||||
func New(ctx context.Context, opts ...ConfigOption) (n Neoq, err error) {
|
||||
ic := internalConfig{}
|
||||
for _, opt := range opts {
|
||||
opt(&ic)
|
||||
|
|
@ -189,13 +190,13 @@ func New(opts ...ConfigOption) (n Neoq, err error) {
|
|||
err = errors.New("your must provide a postgres connection string to initialize the postgres backend: see neoq.ConnectionString(...)")
|
||||
return
|
||||
}
|
||||
n, err = NewPgBackend(ic.connectionString, opts...)
|
||||
n, err = NewPgBackend(ctx, ic.connectionString, opts...)
|
||||
// TODO make the default an in-memory backend
|
||||
default:
|
||||
if ic.connectionString == "" {
|
||||
ic.connectionString = DefaultPgConnectionString
|
||||
}
|
||||
n, err = NewPgBackend(ic.connectionString, opts...)
|
||||
n, err = NewPgBackend(ctx, ic.connectionString, opts...)
|
||||
}
|
||||
|
||||
return
|
||||
|
|
@ -305,17 +306,17 @@ type internalConfig struct {
|
|||
connectionString string // a connection string to use connecting to a backend
|
||||
}
|
||||
|
||||
func (i internalConfig) Enqueue(job Job) (jobID int64, err error) {
|
||||
func (i internalConfig) Enqueue(ctx context.Context, job Job) (jobID int64, err error) {
|
||||
return
|
||||
}
|
||||
func (i internalConfig) Listen(queue string, h Handler) (err error) {
|
||||
func (i internalConfig) Listen(ctx context.Context, queue string, h Handler) (err error) {
|
||||
return
|
||||
}
|
||||
func (i internalConfig) ListenCron(cron string, h Handler) (err error) {
|
||||
func (i internalConfig) ListenCron(ctx context.Context, cron string, h Handler) (err error) {
|
||||
return
|
||||
}
|
||||
|
||||
func (i internalConfig) Shutdown() (err error) {
|
||||
func (i internalConfig) Shutdown(ctx context.Context) (err error) {
|
||||
return
|
||||
}
|
||||
|
||||
|
|
|
|||
14
neoq_test.go
14
neoq_test.go
|
|
@ -11,12 +11,13 @@ import (
|
|||
|
||||
func TestWorkerListenConn(t *testing.T) {
|
||||
const queue = "foobar"
|
||||
pgBackend, err := NewPgBackend("postgres://postgres:postgres@127.0.0.1:5432/neoq")
|
||||
ctx := context.TODO()
|
||||
pgBackend, err := NewPgBackend(ctx, "postgres://postgres:postgres@127.0.0.1:5432/neoq")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
nq, err := New(Backend(pgBackend))
|
||||
nq, err := New(ctx, Backend(pgBackend))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
|
@ -40,13 +41,13 @@ func TestWorkerListenConn(t *testing.T) {
|
|||
}
|
||||
|
||||
// Listen for jobs on the queue
|
||||
nq.Listen(queue, handler)
|
||||
nq.Listen(ctx, queue, handler)
|
||||
|
||||
// allow time for listener to start
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
|
||||
for i := 0; i < numJobs; i++ {
|
||||
jid, err := nq.Enqueue(Job{
|
||||
jid, err := nq.Enqueue(ctx, Job{
|
||||
Queue: queue,
|
||||
Payload: map[string]interface{}{
|
||||
"message": fmt.Sprintf("hello world: %d", i),
|
||||
|
|
@ -89,7 +90,8 @@ func TestWorkerListenConn(t *testing.T) {
|
|||
|
||||
func TestWorkerListenCron(t *testing.T) {
|
||||
const cron = "* * * * * *"
|
||||
nq, err := New()
|
||||
ctx := context.TODO()
|
||||
nq, err := New(ctx)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
|
@ -110,7 +112,7 @@ func TestWorkerListenCron(t *testing.T) {
|
|||
t.Error(err)
|
||||
}
|
||||
|
||||
nq.ListenCron(cron, handler)
|
||||
nq.ListenCron(ctx, cron, handler)
|
||||
|
||||
// allow time for listener to start
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
|
|
|
|||
|
|
@ -55,7 +55,7 @@ type PgBackend struct {
|
|||
// Available ConfigOption
|
||||
// - WithTransactionTimeout(timeout int): configure the idle_in_transaction_timeout for the worker's database
|
||||
// connection(s)
|
||||
func NewPgBackend(connectString string, opts ...ConfigOption) (n Neoq, err error) {
|
||||
func NewPgBackend(ctx context.Context, connectString string, opts ...ConfigOption) (n Neoq, err error) {
|
||||
w := PgBackend{
|
||||
mu: &sync.Mutex{},
|
||||
config: &pgConfig{connectString: connectString},
|
||||
|
|
@ -71,7 +71,7 @@ func NewPgBackend(connectString string, opts ...ConfigOption) (n Neoq, err error
|
|||
opt(&w)
|
||||
}
|
||||
|
||||
err = w.initializeDB()
|
||||
err = w.initializeDB(ctx)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
|
@ -96,7 +96,7 @@ func NewPgBackend(connectString string, opts ...ConfigOption) (n Neoq, err error
|
|||
return
|
||||
}
|
||||
|
||||
w.pool, err = pgxpool.NewWithConfig(context.Background(), poolConfig)
|
||||
w.pool, err = pgxpool.NewWithConfig(ctx, poolConfig)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
|
@ -114,17 +114,16 @@ type pgConfig struct {
|
|||
}
|
||||
|
||||
// initializeDB initializes the tables, types, and indices necessary to operate Neoq
|
||||
func (w PgBackend) initializeDB() (err error) {
|
||||
func (w PgBackend) initializeDB(ctx context.Context) (err error) {
|
||||
var pgxCfg *pgx.ConnConfig
|
||||
var tx pgx.Tx
|
||||
ctx := context.Background()
|
||||
pgxCfg, err = pgx.ParseConfig(w.config.connectString)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
connectStr := fmt.Sprintf("postgres://%s:%s@%s", pgxCfg.User, pgxCfg.Password, pgxCfg.Host)
|
||||
conn, err := pgx.Connect(context.Background(), connectStr)
|
||||
conn, err := pgx.Connect(ctx, connectStr)
|
||||
if err != nil {
|
||||
w.logger.Error("unableto connect to database", err)
|
||||
return
|
||||
|
|
@ -133,7 +132,7 @@ func (w PgBackend) initializeDB() (err error) {
|
|||
|
||||
var dbExists bool
|
||||
dbExistsQ := fmt.Sprintf(`SELECT EXISTS (SELECT datname FROM pg_catalog.pg_database WHERE datname = '%s');`, pgxCfg.Database)
|
||||
rows, err := conn.Query(context.Background(), dbExistsQ)
|
||||
rows, err := conn.Query(ctx, dbExistsQ)
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "unable to determne if jobs table exists: %v", err)
|
||||
return
|
||||
|
|
@ -148,7 +147,7 @@ func (w PgBackend) initializeDB() (err error) {
|
|||
defer rows.Close()
|
||||
|
||||
conn.Close(ctx)
|
||||
conn, err = pgx.Connect(context.Background(), connectStr)
|
||||
conn, err = pgx.Connect(ctx, connectStr)
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "Unable to connect to database: %v\n", err)
|
||||
os.Exit(1)
|
||||
|
|
@ -164,7 +163,7 @@ func (w PgBackend) initializeDB() (err error) {
|
|||
}
|
||||
}
|
||||
|
||||
conn, err = pgx.ConnectConfig(context.Background(), pgxCfg)
|
||||
conn, err = pgx.ConnectConfig(ctx, pgxCfg)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
|
@ -181,7 +180,7 @@ func (w PgBackend) initializeDB() (err error) {
|
|||
schemaname = 'public' AND
|
||||
tablename = 'neoq_jobs'
|
||||
);`
|
||||
rows, err = tx.Query(context.Background(), jobsTableExistsQ)
|
||||
rows, err = tx.Query(ctx, jobsTableExistsQ)
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "unable to determne if jobs table exists: %v", err)
|
||||
return
|
||||
|
|
@ -249,8 +248,7 @@ func (w PgBackend) initializeDB() (err error) {
|
|||
}
|
||||
|
||||
// Enqueue adds jobs to the specified queue
|
||||
func (w PgBackend) Enqueue(job Job) (jobID int64, err error) {
|
||||
ctx := context.Background()
|
||||
func (w PgBackend) Enqueue(ctx context.Context, job Job) (jobID int64, err error) {
|
||||
conn, err := w.pool.Acquire(ctx)
|
||||
if err != nil {
|
||||
return
|
||||
|
|
@ -308,12 +306,12 @@ func (w PgBackend) Enqueue(job Job) (jobID int64, err error) {
|
|||
// Listen sets the queue handler function for the specified queue.
|
||||
//
|
||||
// Neoq will not process any queues until Listen() is called at least once.
|
||||
func (w PgBackend) Listen(queue string, h Handler) (err error) {
|
||||
func (w PgBackend) Listen(ctx context.Context, queue string, h Handler) (err error) {
|
||||
w.mu.Lock()
|
||||
w.handlers[queue] = h
|
||||
w.mu.Unlock()
|
||||
|
||||
err = w.start(queue)
|
||||
err = w.start(ctx, queue)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
|
@ -323,7 +321,7 @@ func (w PgBackend) Listen(queue string, h Handler) (err error) {
|
|||
// ListenCron listens for jobs on a cron schedule and handles them with the provided handler
|
||||
//
|
||||
// See: https://pkg.go.dev/github.com/robfig/cron?#hdr-CRON_Expression_Format for details on the cron spec format
|
||||
func (w PgBackend) ListenCron(cronSpec string, h Handler) (err error) {
|
||||
func (w PgBackend) ListenCron(ctx context.Context, cronSpec string, h Handler) (err error) {
|
||||
cd, err := crondescriptor.NewCronDescriptor(cronSpec)
|
||||
if err != nil {
|
||||
return err
|
||||
|
|
@ -336,19 +334,19 @@ func (w PgBackend) ListenCron(cronSpec string, h Handler) (err error) {
|
|||
|
||||
queue := stripNonAlphanum(strcase.ToSnake(*cdStr))
|
||||
w.cron.AddFunc(cronSpec, func() {
|
||||
w.Enqueue(Job{Queue: queue})
|
||||
w.Enqueue(ctx, Job{Queue: queue})
|
||||
})
|
||||
|
||||
err = w.Listen(queue, h)
|
||||
err = w.Listen(ctx, queue, h)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
func (w PgBackend) Shutdown() (err error) {
|
||||
func (w PgBackend) Shutdown(ctx context.Context) (err error) {
|
||||
w.pool.Close()
|
||||
w.cron.Stop()
|
||||
|
||||
err = w.listenConn.Close(context.Background())
|
||||
err = w.listenConn.Close(ctx)
|
||||
|
||||
return
|
||||
}
|
||||
|
|
@ -470,14 +468,12 @@ func (w PgBackend) updateJob(ctx context.Context, jobErr error) (err error) {
|
|||
}
|
||||
|
||||
// start starts a queue listener, processes pending job, and fires up goroutines to process future jobs
|
||||
func (w PgBackend) start(queue string) (err error) {
|
||||
|
||||
func (w PgBackend) start(ctx context.Context, queue string) (err error) {
|
||||
var handler Handler
|
||||
var ok bool
|
||||
if handler, ok = w.handlers[queue]; !ok {
|
||||
return fmt.Errorf("no handler for queue: %s", queue)
|
||||
}
|
||||
ctx := context.Background()
|
||||
conn, err := w.pool.Acquire(ctx)
|
||||
if err != nil {
|
||||
return
|
||||
|
|
@ -542,7 +538,7 @@ func (w PgBackend) removeFutureJob(jobID int64) {
|
|||
// initFutureJobs is intended to be run once to initialize the list of future jobs that must be monitored for
|
||||
// execution. it should be run only during system startup.
|
||||
func (w PgBackend) initFutureJobs(ctx context.Context, queue string) {
|
||||
rows, err := w.pool.Query(context.Background(), FutureJobQuery, queue)
|
||||
rows, err := w.pool.Query(ctx, FutureJobQuery, queue)
|
||||
if err != nil {
|
||||
w.logger.Error("error fetching future jobs list", err)
|
||||
return
|
||||
|
|
|
|||
Loading…
Reference in a new issue