chore: testing panic in worker

Signed-off-by: Bo-Yi Wu <[email protected]>

Author: appleboy

queue.go

@@ -121,6 +121,7 @@ func (q *Queue) work() {
 			atomic.AddInt32(&q.runningWorkers, -1)
 			if err := recover(); err != nil {
 				q.logger.Error(err)
+				q.logger.Infof("restart the new worker: %d", num)
 				go q.work()
 			}
 		}()

queue_test.go

@@ -93,3 +93,31 @@ func TestWorkerStatus(t *testing.T) {
 	q.Shutdown()
 	q.Wait()
 }
+
+func TestWorkerPanic(t *testing.T) {
+	w := &queueWorker{
+		messages: make(chan QueuedMessage, 10),
+	}
+	q, err := NewQueue(
+		WithWorker(w),
+		WithWorkerCount(2),
+	)
+	assert.NoError(t, err)
+	assert.NotNil(t, q)
+
+	q.Queue(mockMessage{
+		message: "foobar",
+	})
+	q.Queue(mockMessage{
+		message: "foobar",
+	})
+	q.Queue(mockMessage{
+		message: "panic",
+	})
+	q.Start()
+	time.Sleep(30 * time.Millisecond)
+	assert.Equal(t, 2, q.Workers())
+	q.Shutdown()
+	q.Wait()
+	assert.Equal(t, 0, q.Workers())
+}

worker.go

@@ -44,6 +44,9 @@ func (w *queueWorker) AfterRun() error  { return nil }
 func (w *queueWorker) Run(chan struct{}) error {
 	for msg := range w.messages {
 		log.Println("got message", msg)
+		if string(msg.Bytes()) == "panic" {
+			panic("show panic")
+		}
 		time.Sleep(100 * time.Millisecond)
 	}
 	return nil