mirror of
https://github.com/acaloiaro/neoq
synced 2026-07-21 18:29:08 +00:00
Remove *Opt suffix from WithConfig
This commit is contained in:
parent
4559513ee4
commit
d253b50a58
7 changed files with 42 additions and 40 deletions
|
|
@ -11,7 +11,7 @@ import (
|
||||||
func main() {
|
func main() {
|
||||||
const queue = "foobar"
|
const queue = "foobar"
|
||||||
ctx := context.Background()
|
ctx := context.Background()
|
||||||
nq, err := neoq.New(ctx, neoq.PgTransactionTimeoutOpt(1000))
|
nq, err := neoq.New(ctx, neoq.PgTransactionTimeout(1000))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Fatalf("error initializing neoq: %v", err)
|
log.Fatalf("error initializing neoq: %v", err)
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -29,10 +29,10 @@ func main() {
|
||||||
log.Println("got job id:", j.ID, "messsage:", j.Payload["message"])
|
log.Println("got job id:", j.ID, "messsage:", j.Payload["message"])
|
||||||
done <- true
|
done <- true
|
||||||
return
|
return
|
||||||
}, neoq.HandlerConcurrencyOpt(8))
|
}, neoq.HandlerConcurreny(8))
|
||||||
|
|
||||||
// Option 2: Set options after the handler is created
|
// Option 2: Set options after the handler is created
|
||||||
handler = handler.WithOption(neoq.HandlerConcurrencyOpt(8))
|
handler = handler.WithOption(neoq.HandlerConcurreny(8))
|
||||||
|
|
||||||
err = nq.Listen(ctx, queue, handler)
|
err = nq.Listen(ctx, queue, handler)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|
|
||||||
|
|
@ -35,7 +35,7 @@ func main() {
|
||||||
})
|
})
|
||||||
|
|
||||||
// this 10ms deadline will cause our job that sleeps for 1s to fail
|
// this 10ms deadline will cause our job that sleeps for 1s to fail
|
||||||
handler = handler.WithOption(neoq.HandlerDeadlineOpt(10 * time.Millisecond))
|
handler = handler.WithOpt(neoq.HandlerDeadline(10 * time.Millisecond))
|
||||||
|
|
||||||
err = nq.Listen(ctx, queue, handler)
|
err = nq.Listen(ctx, queue, handler)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|
|
||||||
|
|
@ -13,7 +13,7 @@ func main() {
|
||||||
// by default neoq connects to a local postgres server using: [neoq.DefaultPgConnectionString]
|
// by default neoq connects to a local postgres server using: [neoq.DefaultPgConnectionString]
|
||||||
// connection strings can be set explicitly as follows:
|
// connection strings can be set explicitly as follows:
|
||||||
// neoq.New(neoq.ConnectionString("postgres://username:passsword@hostname/database"))
|
// neoq.New(neoq.ConnectionString("postgres://username:passsword@hostname/database"))
|
||||||
nq, err := neoq.New(ctx, neoq.PgTransactionTimeoutOpt(1000))
|
nq, err := neoq.New(ctx, neoq.PgTransactionTimeout(1000))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Fatalf("error initializing neoq: %v", err)
|
log.Fatalf("error initializing neoq: %v", err)
|
||||||
}
|
}
|
||||||
|
|
@ -24,8 +24,8 @@ func main() {
|
||||||
return
|
return
|
||||||
})
|
})
|
||||||
handler = handler.
|
handler = handler.
|
||||||
WithOption(neoq.HandlerDeadlineOpt(time.Duration(500 * time.Millisecond))).
|
WithOption(neoq.HandlerDeadline(500 * time.Millisecond)).
|
||||||
WithOption(neoq.HandlerConcurrencyOpt(1))
|
WithOption(neoq.HandlerConcurreny(1))
|
||||||
|
|
||||||
nq.ListenCron(ctx, "* * * * * *", handler)
|
nq.ListenCron(ctx, "* * * * * *", handler)
|
||||||
|
|
||||||
|
|
|
||||||
36
neoq.go
36
neoq.go
|
|
@ -14,7 +14,6 @@ import (
|
||||||
"context"
|
"context"
|
||||||
"math"
|
"math"
|
||||||
"runtime"
|
"runtime"
|
||||||
"strings"
|
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"math/rand"
|
"math/rand"
|
||||||
|
|
@ -56,8 +55,8 @@ type Neoq interface {
|
||||||
// Shutdown halts the worker
|
// Shutdown halts the worker
|
||||||
Shutdown(ctx context.Context) error
|
Shutdown(ctx context.Context) error
|
||||||
|
|
||||||
// WithConfigOpt configures neoq with with optional configuration
|
// WithConfig configures neoq with with optional configuration
|
||||||
WithConfigOpt(opt ConfigOption) Neoq
|
WithConfig(opt ConfigOption) Neoq
|
||||||
}
|
}
|
||||||
|
|
||||||
// Logger interface is the interface that neoq's logger must implement
|
// Logger interface is the interface that neoq's logger must implement
|
||||||
|
|
@ -91,18 +90,18 @@ func (h Handler) WithOption(opt HandlerOption) (handler Handler) {
|
||||||
return h
|
return h
|
||||||
}
|
}
|
||||||
|
|
||||||
// HandlerDeadlineOpt configures handlers with a deadline for every job that it excutes
|
// HandlerDeadline configures handlers with a time deadline for every executed job
|
||||||
// The deadline is the amount of time (ms) that can be spent executing the handler's HandlerFunction
|
// The deadline is the amount of time that can be spent executing the handler's HandlerFunc
|
||||||
// when the deadline is exceeded, jobs are failed and begin the retry phase of their lifecycle
|
// when a deadline is exceeded, the job is failed and enters its retry phase
|
||||||
func HandlerDeadlineOpt(d time.Duration) HandlerOption {
|
func HandlerDeadline(d time.Duration) HandlerOption {
|
||||||
return func(h *Handler) {
|
return func(h *Handler) {
|
||||||
h.deadline = d
|
h.deadline = d
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// HandlerConcurrencyOpt configures Neoq handlers to process jobs concurrently
|
// HandlerConcurreny configures Neoq handlers to process jobs concurrently
|
||||||
// the default concurrency is the number of (v)CPUs on the machine running Neoq
|
// the default concurrency is the number of (v)CPUs on the machine running Neoq
|
||||||
func HandlerConcurrencyOpt(c int) HandlerOption {
|
func HandlerConcurreny(c int) HandlerOption {
|
||||||
return func(h *Handler) {
|
return func(h *Handler) {
|
||||||
h.concurrency = c
|
h.concurrency = c
|
||||||
}
|
}
|
||||||
|
|
@ -261,21 +260,6 @@ func randInt(max int) int {
|
||||||
return rand.Intn(max)
|
return rand.Intn(max)
|
||||||
}
|
}
|
||||||
|
|
||||||
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()
|
|
||||||
}
|
|
||||||
|
|
||||||
// internalConfig models internal neoq configuratio not exposed to users
|
// internalConfig models internal neoq configuratio not exposed to users
|
||||||
type internalConfig struct {
|
type internalConfig struct {
|
||||||
backendName string // the name of a known backend
|
backendName string // the name of a known backend
|
||||||
|
|
@ -286,9 +270,11 @@ type internalConfig struct {
|
||||||
func (i internalConfig) Enqueue(ctx context.Context, job Job) (jobID int64, err error) {
|
func (i internalConfig) Enqueue(ctx context.Context, job Job) (jobID int64, err error) {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
func (i internalConfig) Listen(ctx context.Context, queue string, h Handler) (err error) {
|
func (i internalConfig) Listen(ctx context.Context, queue string, h Handler) (err error) {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
func (i internalConfig) ListenCron(ctx context.Context, cron string, h Handler) (err error) {
|
func (i internalConfig) ListenCron(ctx context.Context, cron string, h Handler) (err error) {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
@ -297,6 +283,6 @@ func (i internalConfig) Shutdown(ctx context.Context) (err error) {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
func (i internalConfig) WithConfigOpt(opt ConfigOption) (n Neoq) {
|
func (i internalConfig) WithConfig(opt ConfigOption) (n Neoq) {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -33,8 +33,8 @@ func TestWorkerListenConn(t *testing.T) {
|
||||||
return
|
return
|
||||||
})
|
})
|
||||||
handler = handler.
|
handler = handler.
|
||||||
WithOption(HandlerDeadlineOpt(time.Duration(500 * time.Millisecond))).
|
WithOption(HandlerDeadline(500 * time.Millisecond)).
|
||||||
WithOption(HandlerConcurrencyOpt(1))
|
WithOption(HandlerConcurreny(1))
|
||||||
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Error(err)
|
t.Error(err)
|
||||||
|
|
@ -105,8 +105,8 @@ func TestWorkerListenCron(t *testing.T) {
|
||||||
})
|
})
|
||||||
|
|
||||||
handler = handler.
|
handler = handler.
|
||||||
WithOption(HandlerDeadlineOpt(time.Duration(500 * time.Millisecond))).
|
WithOption(HandlerDeadline(500 * time.Millisecond)).
|
||||||
WithOption(HandlerConcurrencyOpt(1))
|
WithOption(HandlerConcurreny(1))
|
||||||
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Error(err)
|
t.Error(err)
|
||||||
|
|
|
||||||
|
|
@ -9,6 +9,7 @@ import (
|
||||||
"log"
|
"log"
|
||||||
"os"
|
"os"
|
||||||
"strconv"
|
"strconv"
|
||||||
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
|
@ -376,7 +377,7 @@ func (w PgBackend) Shutdown(ctx context.Context) (err error) {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
func (w PgBackend) WithConfigOpt(opt ConfigOption) Neoq {
|
func (w PgBackend) WithConfig(opt ConfigOption) Neoq {
|
||||||
opt(&w)
|
opt(&w)
|
||||||
return &w
|
return &w
|
||||||
}
|
}
|
||||||
|
|
@ -804,15 +805,15 @@ func (w PgBackend) getPendingJobID(ctx context.Context, conn *pgxpool.Conn, queu
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
// PgTransactionTimeoutOpt sets the time that NeoqPg's transactions may be idle before its underlying connection is
|
// PgTransactionTimeout sets the time that NeoqPg's transactions may be idle before its underlying connection is
|
||||||
// closed
|
// closed
|
||||||
// The timeout is the number of milliseconds that a transaction may sit idle before postgres terminates the
|
// The timeout is the number of milliseconds that a transaction may sit idle before postgres terminates the
|
||||||
// transaction's underlying connection. The timeout should be longer than your longest job takes to complete. If set
|
// transaction's underlying connection. The timeout should be longer than your longest job takes to complete. If set
|
||||||
// too short, job state will become unpredictable, e.g. retry counts may become incorrect.
|
// too short, job state will become unpredictable, e.g. retry counts may become incorrect.
|
||||||
//
|
//
|
||||||
// PgTransactionTimeoutOpt is best set when calling neoq.New() rather than after creation using WithConfigOpt() because this
|
// PgTransactionTimeout is best set when calling neoq.New() rather than after creation using WithConfig() because this
|
||||||
// setting results in the creation of a new database connection pool.
|
// setting results in the creation of a new database connection pool.
|
||||||
func PgTransactionTimeoutOpt(txTimeout int) ConfigOption {
|
func PgTransactionTimeout(txTimeout int) ConfigOption {
|
||||||
return func(n Neoq) {
|
return func(n Neoq) {
|
||||||
var ok bool
|
var ok bool
|
||||||
var npg PgBackend
|
var npg PgBackend
|
||||||
|
|
@ -857,3 +858,18 @@ func PgTransactionTimeoutOpt(txTimeout int) ConfigOption {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
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()
|
||||||
|
}
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue