diff --git a/examples/add_future_job/main.go b/examples/add_future_job/main.go index adf02eb..0d58f7d 100644 --- a/examples/add_future_job/main.go +++ b/examples/add_future_job/main.go @@ -11,7 +11,7 @@ import ( func main() { const queue = "foobar" ctx := context.Background() - nq, err := neoq.New(ctx, neoq.PgTransactionTimeoutOpt(1000)) + nq, err := neoq.New(ctx, neoq.PgTransactionTimeout(1000)) if err != nil { log.Fatalf("error initializing neoq: %v", err) } diff --git a/examples/add_job_with_custom_concurrency/main.go b/examples/add_job_with_custom_concurrency/main.go index b3ab923..a32059b 100644 --- a/examples/add_job_with_custom_concurrency/main.go +++ b/examples/add_job_with_custom_concurrency/main.go @@ -29,10 +29,10 @@ func main() { log.Println("got job id:", j.ID, "messsage:", j.Payload["message"]) done <- true return - }, neoq.HandlerConcurrencyOpt(8)) + }, neoq.HandlerConcurreny(8)) // 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) if err != nil { diff --git a/examples/add_job_with_deadline/main.go b/examples/add_job_with_deadline/main.go index 8fadce3..c43ee52 100644 --- a/examples/add_job_with_deadline/main.go +++ b/examples/add_job_with_deadline/main.go @@ -35,7 +35,7 @@ func main() { }) // 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) if err != nil { diff --git a/examples/add_periodic_jobs/main.go b/examples/add_periodic_jobs/main.go index 737acde..7d60e26 100644 --- a/examples/add_periodic_jobs/main.go +++ b/examples/add_periodic_jobs/main.go @@ -13,7 +13,7 @@ func main() { // 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, err := neoq.New(ctx, neoq.PgTransactionTimeoutOpt(1000)) + nq, err := neoq.New(ctx, neoq.PgTransactionTimeout(1000)) if err != nil { log.Fatalf("error initializing neoq: %v", err) } @@ -24,8 +24,8 @@ func main() { return }) handler = handler. - WithOption(neoq.HandlerDeadlineOpt(time.Duration(500 * time.Millisecond))). - WithOption(neoq.HandlerConcurrencyOpt(1)) + WithOption(neoq.HandlerDeadline(500 * time.Millisecond)). + WithOption(neoq.HandlerConcurreny(1)) nq.ListenCron(ctx, "* * * * * *", handler) diff --git a/neoq.go b/neoq.go index 9d114b7..17dc3d7 100644 --- a/neoq.go +++ b/neoq.go @@ -14,7 +14,6 @@ import ( "context" "math" "runtime" - "strings" "time" "math/rand" @@ -56,8 +55,8 @@ type Neoq interface { // Shutdown halts the worker Shutdown(ctx context.Context) error - // WithConfigOpt configures neoq with with optional configuration - WithConfigOpt(opt ConfigOption) Neoq + // WithConfig configures neoq with with optional configuration + WithConfig(opt ConfigOption) Neoq } // 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 } -// HandlerDeadlineOpt configures handlers with a deadline for every job that it excutes -// The deadline is the amount of time (ms) that can be spent executing the handler's HandlerFunction -// when the deadline is exceeded, jobs are failed and begin the retry phase of their lifecycle -func HandlerDeadlineOpt(d time.Duration) HandlerOption { +// HandlerDeadline configures handlers with a time deadline for every executed job +// The deadline is the amount of time that can be spent executing the handler's HandlerFunc +// when a deadline is exceeded, the job is failed and enters its retry phase +func HandlerDeadline(d time.Duration) HandlerOption { return func(h *Handler) { 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 -func HandlerConcurrencyOpt(c int) HandlerOption { +func HandlerConcurreny(c int) HandlerOption { return func(h *Handler) { h.concurrency = c } @@ -261,21 +260,6 @@ func randInt(max int) int { 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 type internalConfig struct { 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) { return } + func (i internalConfig) Listen(ctx context.Context, queue string, h Handler) (err error) { return } + func (i internalConfig) ListenCron(ctx context.Context, cron string, h Handler) (err error) { return } @@ -297,6 +283,6 @@ func (i internalConfig) Shutdown(ctx context.Context) (err error) { return } -func (i internalConfig) WithConfigOpt(opt ConfigOption) (n Neoq) { +func (i internalConfig) WithConfig(opt ConfigOption) (n Neoq) { return } diff --git a/neoq_test.go b/neoq_test.go index 1dab9d4..ab8e791 100644 --- a/neoq_test.go +++ b/neoq_test.go @@ -33,8 +33,8 @@ func TestWorkerListenConn(t *testing.T) { return }) handler = handler. - WithOption(HandlerDeadlineOpt(time.Duration(500 * time.Millisecond))). - WithOption(HandlerConcurrencyOpt(1)) + WithOption(HandlerDeadline(500 * time.Millisecond)). + WithOption(HandlerConcurreny(1)) if err != nil { t.Error(err) @@ -105,8 +105,8 @@ func TestWorkerListenCron(t *testing.T) { }) handler = handler. - WithOption(HandlerDeadlineOpt(time.Duration(500 * time.Millisecond))). - WithOption(HandlerConcurrencyOpt(1)) + WithOption(HandlerDeadline(500 * time.Millisecond)). + WithOption(HandlerConcurreny(1)) if err != nil { t.Error(err) diff --git a/postgres_backend.go b/postgres_backend.go index 1ecc488..2b3bb9c 100644 --- a/postgres_backend.go +++ b/postgres_backend.go @@ -9,6 +9,7 @@ import ( "log" "os" "strconv" + "strings" "sync" "time" @@ -376,7 +377,7 @@ func (w PgBackend) Shutdown(ctx context.Context) (err error) { return } -func (w PgBackend) WithConfigOpt(opt ConfigOption) Neoq { +func (w PgBackend) WithConfig(opt ConfigOption) Neoq { opt(&w) return &w } @@ -804,15 +805,15 @@ func (w PgBackend) getPendingJobID(ctx context.Context, conn *pgxpool.Conn, queu 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 // 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 // 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. -func PgTransactionTimeoutOpt(txTimeout int) ConfigOption { +func PgTransactionTimeout(txTimeout int) ConfigOption { return func(n Neoq) { var ok bool 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() +}