Latest development

README.md

@@ -4,9 +4,9 @@ Background job processing for Go
 
 [![Go Reference](https://pkg.go.dev/badge/github.com/acaloiaro/neoq.svg)](https://pkg.go.dev/github.com/acaloiaro/neoq) [![Gitter chat](https://badges.gitter.im/gitterHQ/gitter.png)](https://app.gitter.im/#/room/#neoq:gitter.im)
 
-# Usage
+# Getting Started 
 
-`go get -u github.com/acaloiaro/neoq`
+See the [Getting Started](https://github.com/acaloiaro/neoq/wiki/Getting-Started) wiki to get started. 
 
 # About
 
@@ -46,7 +46,7 @@ Queue Handlers are simple Go functions that accept a `Context` parameter.
 ```go
 ctx := context.Background()
 nq, _ := neoq.New(ctx, neoq.WithBackend(memory.Backend))
-nq.Start(ctx, "hello_world", handler.New(func(ctx context.Context) (err error) {
+nq.Start(ctx, handler.New("greetings", func(ctx context.Context) (err error) {
   j, _ := jobs.FromContext(ctx)
   log.Println("got job id:", j.ID, "messsage:", j.Payload["message"])
   return
@@ -57,22 +57,22 @@ nq.Start(ctx, "hello_world", handler.New(func(ctx context.Context) (err error) {
 
 Enqueuing adds jobs to the specified queue to be processed asynchronously.
 
-**Example**: Add a "Hello World" job to the `hello_world` queue using the default in-memory backend.
+**Example**: Add a "Hello World" job to the `greetings` queue using the default in-memory backend.
 
 ```go
 ctx := context.Background()
 nq, _ := neoq.New(ctx, neoq.WithBackend(memory.Backend))
 nq.Enqueue(ctx, &jobs.Job{
-  Queue: "hello_world",
-  Payload: map[string]interface{}{
+  Queue: "greetings",
+  Payload: map[string]any{
     "message": "hello world",
   },
 })
 ```
 
 ## Redis
 
-**Example**: Process jobs on the "hello_world" queue and add a job to it using the redis backend
+**Example**: Process jobs on the "greetings" queue and add a job to it using the redis backend
 
 ```go
 ctx := context.Background()
@@ -81,14 +81,14 @@ nq, _ := neoq.New(ctx,
   redis.WithAddr("localhost:6379"),
   redis.WithPassword(""))
 
-nq.Start(ctx, "hello_world", handler.New(func(ctx context.Context) (err error) {
+nq.Start(ctx, handler.New("greetings", func(ctx context.Context) (err error) {
   j, _ := jobs.FromContext(ctx)
   log.Println("got job id:", j.ID, "messsage:", j.Payload["message"])
   return
 }))
 
 nq.Enqueue(ctx, &jobs.Job{
-  Queue: "hello_world",
+  Queue: "greetings",
   Payload: map[string]interface{}{
     "message": "hello world",
   },
@@ -97,7 +97,7 @@ nq.Enqueue(ctx, &jobs.Job{
 
 ## Postgres
 
-**Example**: Process jobs on the "hello_world" queue and add a job to it using the postgres backend
+**Example**: Process jobs on the "greetings" queue and add a job to it using the postgres backend
 
 ```go
 ctx := context.Background()
@@ -106,14 +106,14 @@ nq, _ := neoq.New(ctx,
   postgres.WithConnectionString("postgres://postgres:[email protected]:5432/neoq"),
 )
 
-nq.Start(ctx, "hello_world", handler.New(func(ctx context.Context) (err error) {
+nq.Start(ctx, handler.New("greetings", func(ctx context.Context) (err error) {
   j, _ := jobs.FromContext(ctx)
   log.Println("got job id:", j.ID, "messsage:", j.Payload["message"])
   return
 }))
 
 nq.Enqueue(ctx, &jobs.Job{
-  Queue: "hello_world",
+  Queue: "greetings",
   Payload: map[string]interface{}{
     "message": "hello world",
   },

backends/memory/memory_backend.go

@@ -72,6 +72,11 @@ func (m *MemBackend) Enqueue(ctx context.Context, job *jobs.Job) (jobID string,
 	var qc any
 	var ok bool
 
+	if job.Queue == "" {
+		err = jobs.ErrNoQueueSpecified
+		return
+	}
+
 	m.logger.Debug("adding a new job", "queue", job.Queue)
 
 	if qc, ok = m.queues.Load(job.Queue); !ok {
@@ -89,7 +94,6 @@ func (m *MemBackend) Enqueue(ctx context.Context, job *jobs.Job) (jobID string,
 
 	if job.Queue == "" {
 		err = jobs.ErrNoQueueSpecified
-
 		return
 	}
 
@@ -121,22 +125,22 @@ func (m *MemBackend) Enqueue(ctx context.Context, job *jobs.Job) (jobID string,
 }
 
 // Start starts processing jobs with the specified queue and handler
-func (m *MemBackend) Start(ctx context.Context, queue string, h handler.Handler) (err error) {
+func (m *MemBackend) Start(ctx context.Context, h handler.Handler) (err error) {
 	queueCapacity := h.QueueCapacity
 	if queueCapacity == emptyCapacity {
 		queueCapacity = defaultMemQueueCapacity
 	}
 
-	m.handlers.Store(queue, h)
-	m.queues.Store(queue, make(chan *jobs.Job, queueCapacity))
+	m.handlers.Store(h.Queue, h)
+	m.queues.Store(h.Queue, make(chan *jobs.Job, queueCapacity))
 
 	ctx, cancel := context.WithCancel(ctx)
 
 	m.mu.Lock()
 	m.cancelFuncs = append(m.cancelFuncs, cancel)
 	m.mu.Unlock()
 
-	err = m.start(ctx, queue)
+	err = m.start(ctx, h.Queue)
 	if err != nil {
 		return
 	}
@@ -164,8 +168,9 @@ func (m *MemBackend) StartCron(ctx context.Context, cronSpec string, h handler.H
 	m.mu.Lock()
 	m.cancelFuncs = append(m.cancelFuncs, cancel)
 	m.mu.Unlock()
+	h.Queue = queue
 
-	err = m.Start(ctx, queue, h)
+	err = m.Start(ctx, h)
 	if err != nil {
 		return fmt.Errorf("error processing queue '%s': %w", queue, err)
 	}
@@ -176,7 +181,7 @@ func (m *MemBackend) StartCron(ctx context.Context, cronSpec string, h handler.H
 		return fmt.Errorf("error adding cron: %w", err)
 	}
 
-	return
+	return err
 }
 
 // SetLogger sets this backend's logger
@@ -203,13 +208,12 @@ func (m *MemBackend) start(ctx context.Context, queue string) (err error) {
 
 	if ht, ok = m.handlers.Load(queue); !ok {
 		err = fmt.Errorf("%w: %s", handler.ErrNoHandlerForQueue, queue)
-		m.logger.Error("error loading handler for queue", queue)
+		m.logger.Error("error loading handler for queue", "queue", queue)
 		return
 	}
 
 	if qc, ok = m.queues.Load(queue); !ok {
-		err = fmt.Errorf("%w: %s", handler.ErrNoProcessorForQueue, queue)
-		m.logger.Error("error loading channel for queue", queue)
+		m.logger.Error("error loading channel for queue", "queue", queue, "error", handler.ErrNoHandlerForQueue)
 		return err
 	}
 
@@ -222,13 +226,12 @@ func (m *MemBackend) start(ctx context.Context, queue string) (err error) {
 		go func() {
 			var err error
 			var job *jobs.Job
-
 			for {
 				select {
 				case job = <-queueChan:
 					err = m.handleJob(ctx, job, h)
 				case <-ctx.Done():
-					return
+					err = ctx.Err()
 				}
 
 				if err != nil {

backends/memory/memory_backend_test.go

@@ -52,12 +52,12 @@ func TestBasicJobProcessing(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	h := handler.New(func(_ context.Context) (err error) {
+	h := handler.New(queue, func(_ context.Context) (err error) {
 		done <- true
 		return
 	})
 
-	if err := nq.Start(ctx, queue, h); err != nil {
+	if err := nq.Start(ctx, h); err != nil {
 		t.Fatal(err)
 	}
 
@@ -126,12 +126,12 @@ func TestBackendConfiguration(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	h := handler.New(func(_ context.Context) (err error) {
+	h := handler.New(queue, func(_ context.Context) (err error) {
 		time.Sleep(100 * time.Millisecond)
 		return
 	}, handler.Concurrency(1), handler.MaxQueueCapacity(1))
 
-	if err := nq.Start(ctx, queue, h); err != nil {
+	if err := nq.Start(ctx, h); err != nil {
 		t.Fatal(err)
 	}
 
@@ -184,11 +184,11 @@ func TestFutureJobScheduling(t *testing.T) {
 	}
 	defer nq.Shutdown(ctx)
 
-	h := handler.New(func(ctx context.Context) (err error) {
+	h := handler.New(queue, func(ctx context.Context) (err error) {
 		return
 	})
 
-	if err := nq.Start(ctx, queue, h); err != nil {
+	if err := nq.Start(ctx, h); err != nil {
 		t.Fatal(err)
 	}
 
@@ -232,7 +232,7 @@ func TestFutureJobSchedulingMultipleQueues(t *testing.T) {
 	done1 := make(chan bool)
 	done2 := make(chan bool)
 
-	h1 := handler.New(func(ctx context.Context) (err error) {
+	h1 := handler.New(q1, func(ctx context.Context) (err error) {
 		var j *jobs.Job
 		j, err = jobs.FromContext(ctx)
 		if err != nil {
@@ -247,7 +247,7 @@ func TestFutureJobSchedulingMultipleQueues(t *testing.T) {
 		return
 	})
 
-	h2 := handler.New(func(ctx context.Context) (err error) {
+	h2 := handler.New(q2, func(ctx context.Context) (err error) {
 		var j *jobs.Job
 		j, err = jobs.FromContext(ctx)
 		if err != nil {
@@ -262,11 +262,11 @@ func TestFutureJobSchedulingMultipleQueues(t *testing.T) {
 		return
 	})
 
-	if err := nq.Start(ctx, q1, h1); err != nil {
+	if err := nq.Start(ctx, h1); err != nil {
 		t.Fatal(err)
 	}
 
-	if err := nq.Start(ctx, q2, h2); err != nil {
+	if err := nq.Start(ctx, h2); err != nil {
 		t.Fatal(err)
 	}
 
@@ -339,7 +339,7 @@ func TestCron(t *testing.T) {
 	defer nq.Shutdown(ctx)
 
 	done := make(chan bool)
-	h := handler.New(func(ctx context.Context) (err error) {
+	h := handler.New("foobar", func(ctx context.Context) (err error) {
 		done <- true
 		return
 	})