Merge pull request #21 from tinh-tinh/feat/ren/20-recovery-error-prev…

…ent-panic

feat: recovery error to prevent panic app

job.go

@@ -71,6 +71,24 @@ func (job *Job) Process(cb Callback) {
 	job.ProcessedOn = time.Now()
 	fmt.Printf("⌛ Running job %s progress\n", job.Id)
 	println()
+	defer func() {
+		if r := recover(); r != nil {
+			job.FailedReason = fmt.Sprintf("%v", r)
+			job.Status = FailedStatus
+			// Store error
+			client := job.queue.client
+			client.HSet(context.Background(), fmt.Sprintf("%sstore", job.queue.Name), job.Id, job.FailedReason).Result()
+			if job.RetryFailures > 0 {
+				job.Status = DelayedStatus
+				job.RetryFailures--
+				fmt.Printf("Add job %s for retry (%d remains) 🕛\n", job.Id, job.RetryFailures)
+				println()
+			} else {
+				fmt.Printf("Failed job %s ❌\n", job.Id)
+				println()
+			}
+		}
+	}()
 	err := cb()
 	if err == nil {
 		job.FinishedOn = time.Now()
@@ -82,7 +100,7 @@ func (job *Job) Process(cb Callback) {
 		job.Status = FailedStatus
 		// Store error
 		client := job.queue.client
-		client.Set(context.Background(), job.Id, job.FailedReason, time.Hour)
+		client.HSet(context.Background(), fmt.Sprintf("%sstore", job.queue.Name), job.Id, job.FailedReason).Result()
 		if job.RetryFailures > 0 {
 			job.Status = DelayedStatus
 			job.RetryFailures--

queue_test.go

@@ -160,3 +160,30 @@ func HeaveTask(key string) (string, error) {
 	bytes, err := bcrypt.GenerateFromPassword([]byte(key), 14)
 	return string(bytes), err
 }
+
+func Test_Crash(t *testing.T) {
+	addr := "localhost:6379"
+	userQueue := queue.New("crash", &queue.Options{
+		Connect: &redis.Options{
+			Addr:     addr,
+			Password: "",
+			DB:       0,
+		},
+		Workers:       3,
+		RetryFailures: 3,
+	})
+
+	userQueue.Process(func(job *queue.Job) {
+		job.Process(func() error {
+			panic("error by test")
+		})
+	})
+
+	t.Parallel()
+
+	// t.Run("test", func(t *testing.T) {
+	userQueue.AddJob(queue.AddJobOptions{
+		Id:   "1",
+		Data: "value 1",
+	})
+}