mirror of
https://github.com/acaloiaro/neoq
synced 2026-07-21 18:29:08 +00:00
Add periodic job support
This commit is contained in:
Adriano Caloiaro
parent
0984795a16
commit
f1f5ea2988
6 changed files with 183 additions and 10 deletions
|
|
@ -17,10 +17,11 @@ Neoq does not aim to be the _fastest_ background job processor. It aims to be _f
|
|||
# Features
|
||||
|
||||
- **Postgres-backed** job processing
|
||||
- **Job uniqueness**: jobs are fingerprinted based on their payload and status to prevent job duplication (multiple unprocessed jobs with the same payload cannot be queued)
|
||||
- **Retries**: Jobs may be retried a configurable number of times with exponential backoff and jitter to prevent thundering herds
|
||||
- **Job uniqueness**: jobs are fingerprinted based on their payload and status to prevent job duplication (multiple unprocessed jobs with the same payload cannot be queued)
|
||||
- **Deadlines**: Queue handlers can be configured with per-job time deadlines with millisecond accuracy
|
||||
- **Configurable transaction idle time**: Don't let your background worker transactions run away with db resources. By default, worker transactions may idle for 60 seconds.
|
||||
- **Periodic Jobs**: Jobs can be scheduled periodically using standard cron syntax
|
||||
- **Future Jobs**: Jobs can be scheduled either for the future or immediate execution
|
||||
- **Concurrency**: Concurrency is configurable for every queue
|
||||
|
||||
|
|
|
|||
25
examples/add_periodic_jobs/main.go
Normal file
25
examples/add_periodic_jobs/main.go
Normal file
|
|
@ -0,0 +1,25 @@
|
|||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log"
|
||||
"time"
|
||||
|
||||
"github.com/acaloiaro/neoq"
|
||||
)
|
||||
|
||||
func main() {
|
||||
nq, _ := neoq.New("postgres://postgres:postgres@127.0.0.1:5432/neoq?sslmode=disable", neoq.TransactionTimeoutOpt(1000))
|
||||
// run a job periodically
|
||||
handler := neoq.NewHandler(func(ctx context.Context) (err error) {
|
||||
log.Println("running periodic job")
|
||||
return
|
||||
})
|
||||
handler = handler.
|
||||
WithOption(neoq.HandlerDeadlineOpt(time.Duration(500 * time.Millisecond))).
|
||||
WithOption(neoq.HandlerConcurrencyOpt(1))
|
||||
|
||||
nq.ListenCron("* * * * * *", handler)
|
||||
|
||||
time.Sleep(5 * time.Second)
|
||||
}
|
||||
5
go.mod
5
go.mod
|
|
@ -10,10 +10,15 @@ require (
|
|||
)
|
||||
|
||||
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
|
||||
golang.org/x/crypto v0.0.0-20220829220503-c86fa9a7ed90 // indirect
|
||||
golang.org/x/sync v0.0.0-20220923202941-7f9b1623fab7 // indirect
|
||||
golang.org/x/text v0.3.8 // indirect
|
||||
|
|
|
|||
42
go.sum
42
go.sum
|
|
@ -1,7 +1,13 @@
|
|||
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=
|
||||
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
|
||||
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI=
|
||||
github.com/guregu/null v4.0.0+incompatible h1:4zw0ckM7ECd6FNNddc3Fu4aty9nTlpkkzH7dPn4/4Gw=
|
||||
github.com/guregu/null v4.0.0+incompatible/go.mod h1:ePGpQaN9cw0tj45IR5E5ehMvsFlLlQZAkkOXZurJ3NM=
|
||||
github.com/iancoleman/strcase v0.2.0 h1:05I4QRnGpI0m37iZQRuskXh+w77mr6Z41lwQzuHLwW0=
|
||||
github.com/iancoleman/strcase v0.2.0/go.mod h1:iwCmte+B7n89clKwxIoIXy/HfoL7AsD47ZCWhYzw7ho=
|
||||
github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
|
||||
github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
|
||||
github.com/jackc/pgservicefile v0.0.0-20200714003250-2b9c44734f2b h1:C8S2+VttkHFdOOCXJe+YGfa4vHYwlt4Zx+IVXQ97jYg=
|
||||
|
|
@ -10,24 +16,60 @@ github.com/jackc/pgx/v5 v5.2.0 h1:NdPpngX0Y6z6XDFKqmFQaE+bCtkqzvQIOt1wvBlAqs8=
|
|||
github.com/jackc/pgx/v5 v5.2.0/go.mod h1:Ptn7zmohNsWEsdxRawMzk3gaKma2obW+NWTnKa0S4nk=
|
||||
github.com/jackc/puddle/v2 v2.1.2 h1:0f7vaaXINONKTsxYDn4otOAiJanX/BMeAtY//BXqzlg=
|
||||
github.com/jackc/puddle/v2 v2.1.2/go.mod h1:2lpufsF5mRHO6SuZkm0fNYxM6SWHfvyFj62KwNzgels=
|
||||
github.com/jsuar/go-cron-descriptor v0.1.0 h1:Q97ujk+/xhcz1lA9nmUMq750FZV7RlV23TNyMB74xkQ=
|
||||
github.com/jsuar/go-cron-descriptor v0.1.0/go.mod h1:PFR+Y6Lr86uYZpwsWRoFMRA3CX4a6q7zfPltp3SLkSU=
|
||||
github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
|
||||
github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
|
||||
github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
|
||||
github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
|
||||
github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
|
||||
github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
|
||||
github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
|
||||
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
|
||||
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
|
||||
github.com/robfig/cron v1.2.0 h1:ZjScXvvxeQ63Dbyxy76Fj3AT3Ut0aKsyd2/tl3DTMuQ=
|
||||
github.com/robfig/cron v1.2.0/go.mod h1:JGuDeoQd7Z6yL4zQhZ3OPEVHB7fL6Ka6skscFHfmt2k=
|
||||
github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
|
||||
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
|
||||
github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
|
||||
github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
|
||||
github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
|
||||
github.com/stretchr/testify v1.8.0 h1:pSgiaMZlXftHpm5L7V1+rVB+AZJydKsMxsQBIJw4PKk=
|
||||
go.uber.org/atomic v1.6.0/go.mod h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ=
|
||||
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/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=
|
||||
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
|
||||
golang.org/x/crypto v0.0.0-20190510104115-cbcb75029529/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
|
||||
golang.org/x/crypto v0.0.0-20220829220503-c86fa9a7ed90 h1:Y/gsMcFOcR+6S6f3YeMKl5g+dZMEWqcz5Czj/GWYbkM=
|
||||
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/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
|
||||
golang.org/x/mod v0.0.0-20190513183733-4bf6d317e70e/go.mod h1:mXi4GBBbnImb6dmsKGUJ2LatrhH/nqhxcFungHvyanc=
|
||||
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=
|
||||
golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
|
||||
golang.org/x/sync v0.0.0-20220923202941-7f9b1623fab7 h1:ZrnxWX62AgTKOSagEqxvb3ffipvEDX2pl7E1TdqLqIc=
|
||||
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/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=
|
||||
golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
|
||||
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/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/go.mod h1:a3bituU0lyd329TUQxRnasdCoJDkEUEAqEt0JzvZhAg=
|
||||
|
|
|
|||
75
neoq.go
75
neoq.go
|
|
@ -7,6 +7,9 @@
|
|||
// Redis as their primary data stores will ideally use Neoq with corresponding backends.
|
||||
package neoq
|
||||
|
||||
// TODO dependencies to factor out
|
||||
// "github.com/iancoleman/strcase"
|
||||
// "github.com/jsuar/go-cron-descriptor/pkg/crondescriptor"
|
||||
import (
|
||||
"context"
|
||||
"crypto/md5"
|
||||
|
|
@ -18,15 +21,19 @@ import (
|
|||
"os"
|
||||
"runtime"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"math/rand"
|
||||
|
||||
"github.com/guregu/null"
|
||||
"github.com/iancoleman/strcase"
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
"github.com/jsuar/go-cron-descriptor/pkg/crondescriptor"
|
||||
"github.com/pkg/errors"
|
||||
"github.com/robfig/cron"
|
||||
"golang.org/x/exp/slog"
|
||||
)
|
||||
|
||||
|
|
@ -179,20 +186,25 @@ func txFromContext(ctx context.Context) (t pgx.Tx, err error) {
|
|||
|
||||
// Neoq interface is Neoq's primary API
|
||||
type Neoq interface {
|
||||
// Config retrieves neoq's configuration
|
||||
Config() *Config
|
||||
|
||||
// Enqueue queues jobs to be executed asynchronously
|
||||
Enqueue(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)
|
||||
|
||||
// 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)
|
||||
|
||||
// Shutdown halts the worker
|
||||
Shutdown() error
|
||||
|
||||
// WithConfigOpt configures neoq with with optional configuration
|
||||
WithConfigOpt(opt ConfigOption) Neoq
|
||||
|
||||
// Config retrieves neoq's configuration
|
||||
Config() *Config
|
||||
}
|
||||
|
||||
// Logger interface is the interface that neoq's logger must implement
|
||||
|
|
@ -265,6 +277,7 @@ func NewHandler(f HandlerFunc, opts ...HandlerOption) (h Handler) {
|
|||
// pgWorker is a concrete Neoq implementation based on Postgres, using pgx/pgxpool
|
||||
type pgWorker struct {
|
||||
config *Config
|
||||
cron *cron.Cron
|
||||
dbConnectString string
|
||||
idleTxTimeout int
|
||||
listenConn *pgx.Conn
|
||||
|
|
@ -308,7 +321,9 @@ func New(connection string, opts ...ConfigOption) (n Neoq, err error) {
|
|||
handlers: make(map[string]Handler),
|
||||
futureJobs: make(map[int64]time.Time),
|
||||
logger: slog.New(slog.NewTextHandler(os.Stdout)),
|
||||
cron: cron.New(),
|
||||
}
|
||||
w.cron.Start()
|
||||
|
||||
// Set all options
|
||||
for _, opt := range opts {
|
||||
|
|
@ -545,7 +560,7 @@ func (w pgWorker) Enqueue(job Job) (jobID int64, err error) {
|
|||
|
||||
// Listen sets the queue handler function for the specified queue.
|
||||
//
|
||||
// Neoq will not process any jobs until Listen() is called at least once.
|
||||
// Neoq will not process any queues until Listen() is called at least once.
|
||||
func (w pgWorker) Listen(queue string, h Handler) (err error) {
|
||||
w.handlers[queue] = h
|
||||
|
||||
|
|
@ -556,8 +571,33 @@ func (w pgWorker) Listen(queue string, h Handler) (err error) {
|
|||
return
|
||||
}
|
||||
|
||||
// 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 pgWorker) ListenCron(cronSpec string, h Handler) (err error) {
|
||||
cd, err := crondescriptor.NewCronDescriptor(cronSpec)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
cdStr, err := cd.GetDescription(crondescriptor.Full)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
queue := stripNonAlphanum(strcase.ToSnake(*cdStr))
|
||||
w.cron.AddFunc(cronSpec, func() {
|
||||
w.Enqueue(Job{Queue: queue})
|
||||
})
|
||||
|
||||
err = w.Listen(queue, h)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
func (w pgWorker) Shutdown() (err error) {
|
||||
w.pool.Close()
|
||||
w.cron.Stop()
|
||||
|
||||
err = w.listenConn.Close(context.Background())
|
||||
|
||||
|
|
@ -796,7 +836,7 @@ func (w pgWorker) scheduleFutureJobs(ctx context.Context, queue string) {
|
|||
}
|
||||
}
|
||||
|
||||
// annoucneJob announces jobs to queue listeners.
|
||||
// announceJob announces jobs to queue listeners.
|
||||
//
|
||||
// Announced jobs are executed by the first worker to respond to the announcement.
|
||||
func (w pgWorker) announceJob(ctx context.Context, queue string, jobID int64) {
|
||||
|
|
@ -941,7 +981,9 @@ func (w pgWorker) listen(ctx context.Context, queue string) (c chan int64) {
|
|||
// set this connection's idle in transaction timeout to infinite so it is not intermittently disconnected
|
||||
_, err = w.listenConn.Exec(ctx, fmt.Sprintf("SET idle_in_transaction_session_timeout = '0'; LISTEN %s", queue))
|
||||
if err != nil {
|
||||
w.logger.Error("unable to create database connection for listener", err)
|
||||
msg := "unable to create database connection for listener"
|
||||
err = errors.Wrap(err, msg)
|
||||
w.logger.Error(msg, err)
|
||||
return
|
||||
}
|
||||
|
||||
|
|
@ -982,6 +1024,11 @@ func (w pgWorker) getPendingJob(ctx context.Context, tx pgx.Tx, jobID int64) (jo
|
|||
return *j, err
|
||||
}
|
||||
|
||||
func (w pgWorker) getPendingJobID(ctx context.Context, conn *pgxpool.Conn, queue string) (jobID int64, err error) {
|
||||
err = conn.QueryRow(ctx, PendingJobIDQuery, queue).Scan(&jobID)
|
||||
return
|
||||
}
|
||||
|
||||
// calculateBackoff calculates the number of seconds to back off before the next retry
|
||||
// this formula is unabashedly taken from Sidekiq because it is good.
|
||||
func calculateBackoff(retryCount int) time.Time {
|
||||
|
|
@ -994,7 +1041,17 @@ func randInt(max int) int {
|
|||
return rand.Intn(max)
|
||||
}
|
||||
|
||||
func (w pgWorker) getPendingJobID(ctx context.Context, conn *pgxpool.Conn, queue string) (jobID int64, err error) {
|
||||
err = conn.QueryRow(ctx, PendingJobIDQuery, queue).Scan(&jobID)
|
||||
return
|
||||
func stripNonAlphanum(s string) string {
|
||||
var result strings.Builder
|
||||
for i := 0; i < len(s); i++ {
|
||||
b := s[i]
|
||||
if (b == '_') ||
|
||||
('a' <= b && b <= 'z') ||
|
||||
('A' <= b && b <= 'Z') ||
|
||||
('0' <= b && b <= '9') ||
|
||||
b == ' ' {
|
||||
result.WriteByte(b)
|
||||
}
|
||||
}
|
||||
return result.String()
|
||||
}
|
||||
|
|
|
|||
43
neoq_test.go
43
neoq_test.go
|
|
@ -82,3 +82,46 @@ func TestWorkerListenConn(t *testing.T) {
|
|||
}
|
||||
|
||||
}
|
||||
|
||||
func TestWorkerListenCron(t *testing.T) {
|
||||
const cron = "* * * * * *"
|
||||
nq, err := New("postgres://postgres:postgres@127.0.0.1:5432/neoq?sslmode=disable")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
jobRan := false
|
||||
var done = make(chan bool)
|
||||
handler := NewHandler(func(ctx context.Context) (err error) {
|
||||
slog.Info("got periodic job")
|
||||
done <- true
|
||||
return
|
||||
})
|
||||
handler = handler.
|
||||
WithOption(HandlerDeadlineOpt(time.Duration(500 * time.Millisecond))).
|
||||
WithOption(HandlerConcurrencyOpt(1))
|
||||
|
||||
if err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
|
||||
nq.ListenCron(cron, handler)
|
||||
|
||||
// allow time for listener to start
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
|
||||
select {
|
||||
case <-time.After(5 * time.Second):
|
||||
err = errors.New("timed out waiting for periodic job")
|
||||
case <-done:
|
||||
jobRan = true
|
||||
}
|
||||
|
||||
// Allow time for job status to be updated in the database
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
|
||||
if !jobRan {
|
||||
t.Error(err)
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue