From a5a9f4ba50ce555bf0744f2b6c8cfe8864902397 Mon Sep 17 00:00:00 2001 From: Jose Cely Date: Mon, 22 Jun 2026 11:46:36 -0500 Subject: [PATCH] add maxruns and option for loop forever --- controllers/schedsample.go | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) diff --git a/controllers/schedsample.go b/controllers/schedsample.go index 44856d0..1480a19 100644 --- a/controllers/schedsample.go +++ b/controllers/schedsample.go @@ -23,40 +23,40 @@ func SchedulerSample(c *core.Context) *core.Response { ctx := context.Background() - // ── Enqueue a "send_email" task ───────────────────────────────────────── + // ── Enqueue a "send_email" task (runs once, sequential) ──────────────── emailPayload, _ := json.Marshal(map[string]interface{}{ "to": "user@example.com", "subject": "Hello from the scheduler!", }) - item1, err := store.Enqueue(ctx, "send_email", string(emailPayload), 0) + item1, err := store.Enqueue(ctx, "send_email", string(emailPayload), 0, 1, 0) if err != nil { c.GetLogger().Error(fmt.Sprintf("failed to enqueue send_email: %v", err)) return c.Response.SetStatusCode(500).Json(`{"message": "failed to enqueue send_email"}`) } - c.GetLogger().Info(fmt.Sprintf("[scheduler] enqueued send_email task id=%d", item1.ID)) + c.GetLogger().Info(fmt.Sprintf("[scheduler] enqueued send_email task id=%d (maxRuns=1, thread=0)", item1.ID)) - // ── Enqueue a "flaky_task" (demonstrates error recording) ─────────────── + // ── Enqueue a "flaky_task" (runs 3 times, sequential) ────────────────── flakyPayload, _ := json.Marshal(map[string]interface{}{ "description": "This task will fail and the error will be recorded", }) - item2, err := store.Enqueue(ctx, "flaky_task", string(flakyPayload), 0) + item2, err := store.Enqueue(ctx, "flaky_task", string(flakyPayload), 0, 3, 0) if err != nil { c.GetLogger().Error(fmt.Sprintf("failed to enqueue flaky_task: %v", err)) return c.Response.SetStatusCode(500).Json(`{"message": "failed to enqueue flaky_task"}`) } - c.GetLogger().Info(fmt.Sprintf("[scheduler] enqueued flaky_task id=%d", item2.ID)) + c.GetLogger().Info(fmt.Sprintf("[scheduler] enqueued flaky_task id=%d (maxRuns=3, thread=0)", item2.ID)) - // ── Enqueue a high-priority "send_email" task ─────────────────────────── + // ── Enqueue a high-priority "send_email" (repeats indefinitely, concurrent) ── urgentPayload, _ := json.Marshal(map[string]interface{}{ "to": "urgent@example.com", "subject": "URGENT: Priority message", }) - item3, err := store.Enqueue(ctx, "send_email", string(urgentPayload), 10) + item3, err := store.Enqueue(ctx, "send_email", string(urgentPayload), 10, -1, 1) if err != nil { c.GetLogger().Error(fmt.Sprintf("failed to enqueue urgent send_email: %v", err)) return c.Response.SetStatusCode(500).Json(`{"message": "failed to enqueue urgent send_email"}`) } - c.GetLogger().Info(fmt.Sprintf("[scheduler] enqueued high-priority send_email task id=%d", item3.ID)) + c.GetLogger().Info(fmt.Sprintf("[scheduler] enqueued high-priority send_email id=%d (maxRuns=-1/infinite, thread=1/concurrent)", item3.ID)) message := fmt.Sprintf(`{"message": "Enqueued 3 scheduler tasks (send_email x2, flaky_task x1)", "tasks": [%d, %d, %d]}`, item1.ID, item2.ID, item3.ID)