forked from hibiken/asynq
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Renamed previously called scheduler to forwarder to resolve name conflicts
- Loading branch information
Showing
19 changed files
with
989 additions
and
187 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,75 @@ | ||
| // Copyright 2020 Kentaro Hibino. All rights reserved. | ||
| // Use of this source code is governed by a MIT license | ||
| // that can be found in the LICENSE file. | ||
|
|
||
| package asynq | ||
|
|
||
| import ( | ||
| "sync" | ||
| "time" | ||
|
|
||
| "github.com/hibiken/asynq/internal/base" | ||
| "github.com/hibiken/asynq/internal/log" | ||
| ) | ||
|
|
||
| // A forwarder is responsible for moving scheduled and retry tasks to pending state | ||
| // so that the tasks get processed by the workers. | ||
| type forwarder struct { | ||
| logger *log.Logger | ||
| broker base.Broker | ||
|
|
||
| // channel to communicate back to the long running "forwarder" goroutine. | ||
| done chan struct{} | ||
|
|
||
| // list of queue names to check and enqueue. | ||
| queues []string | ||
|
|
||
| // poll interval on average | ||
| avgInterval time.Duration | ||
| } | ||
|
|
||
| type forwarderParams struct { | ||
| logger *log.Logger | ||
| broker base.Broker | ||
| queues []string | ||
| interval time.Duration | ||
| } | ||
|
|
||
| func newForwarder(params forwarderParams) *forwarder { | ||
| return &forwarder{ | ||
| logger: params.logger, | ||
| broker: params.broker, | ||
| done: make(chan struct{}), | ||
| queues: params.queues, | ||
| avgInterval: params.interval, | ||
| } | ||
| } | ||
|
|
||
| func (f *forwarder) terminate() { | ||
| f.logger.Debug("Forwarder shutting down...") | ||
| // Signal the forwarder goroutine to stop polling. | ||
| f.done <- struct{}{} | ||
| } | ||
|
|
||
| // start starts the "forwarder" goroutine. | ||
| func (f *forwarder) start(wg *sync.WaitGroup) { | ||
| wg.Add(1) | ||
| go func() { | ||
| defer wg.Done() | ||
| for { | ||
| select { | ||
| case <-f.done: | ||
| f.logger.Debug("Forwarder done") | ||
| return | ||
| case <-time.After(f.avgInterval): | ||
| f.exec() | ||
| } | ||
| } | ||
| }() | ||
| } | ||
|
|
||
| func (f *forwarder) exec() { | ||
| if err := f.broker.CheckAndEnqueue(f.queues...); err != nil { | ||
| f.logger.Errorf("Could not enqueue scheduled tasks: %v", err) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,137 @@ | ||
| // Copyright 2020 Kentaro Hibino. All rights reserved. | ||
| // Use of this source code is governed by a MIT license | ||
| // that can be found in the LICENSE file. | ||
|
|
||
| package asynq | ||
|
|
||
| import ( | ||
| "sync" | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/google/go-cmp/cmp" | ||
| h "github.com/hibiken/asynq/internal/asynqtest" | ||
| "github.com/hibiken/asynq/internal/base" | ||
| "github.com/hibiken/asynq/internal/rdb" | ||
| ) | ||
|
|
||
| func TestForwarder(t *testing.T) { | ||
| r := setup(t) | ||
| defer r.Close() | ||
| rdbClient := rdb.NewRDB(r) | ||
| const pollInterval = time.Second | ||
| s := newForwarder(forwarderParams{ | ||
| logger: testLogger, | ||
| broker: rdbClient, | ||
| queues: []string{"default", "critical"}, | ||
| interval: pollInterval, | ||
| }) | ||
| t1 := h.NewTaskMessageWithQueue("gen_thumbnail", nil, "default") | ||
| t2 := h.NewTaskMessageWithQueue("send_email", nil, "critical") | ||
| t3 := h.NewTaskMessageWithQueue("reindex", nil, "default") | ||
| t4 := h.NewTaskMessageWithQueue("sync", nil, "critical") | ||
| now := time.Now() | ||
|
|
||
| tests := []struct { | ||
| initScheduled map[string][]base.Z // scheduled queue initial state | ||
| initRetry map[string][]base.Z // retry queue initial state | ||
| initPending map[string][]*base.TaskMessage // default queue initial state | ||
| wait time.Duration // wait duration before checking for final state | ||
| wantScheduled map[string][]*base.TaskMessage // schedule queue final state | ||
| wantRetry map[string][]*base.TaskMessage // retry queue final state | ||
| wantPending map[string][]*base.TaskMessage // default queue final state | ||
| }{ | ||
| { | ||
| initScheduled: map[string][]base.Z{ | ||
| "default": {{Message: t1, Score: now.Add(time.Hour).Unix()}}, | ||
| "critical": {{Message: t2, Score: now.Add(-2 * time.Second).Unix()}}, | ||
| }, | ||
| initRetry: map[string][]base.Z{ | ||
| "default": {{Message: t3, Score: time.Now().Add(-500 * time.Millisecond).Unix()}}, | ||
| "critical": {}, | ||
| }, | ||
| initPending: map[string][]*base.TaskMessage{ | ||
| "default": {}, | ||
| "critical": {t4}, | ||
| }, | ||
| wait: pollInterval * 2, | ||
| wantScheduled: map[string][]*base.TaskMessage{ | ||
| "default": {t1}, | ||
| "critical": {}, | ||
| }, | ||
| wantRetry: map[string][]*base.TaskMessage{ | ||
| "default": {}, | ||
| "critical": {}, | ||
| }, | ||
| wantPending: map[string][]*base.TaskMessage{ | ||
| "default": {t3}, | ||
| "critical": {t2, t4}, | ||
| }, | ||
| }, | ||
| { | ||
| initScheduled: map[string][]base.Z{ | ||
| "default": { | ||
| {Message: t1, Score: now.Unix()}, | ||
| {Message: t3, Score: now.Add(-500 * time.Millisecond).Unix()}, | ||
| }, | ||
| "critical": { | ||
| {Message: t2, Score: now.Add(-2 * time.Second).Unix()}, | ||
| }, | ||
| }, | ||
| initRetry: map[string][]base.Z{ | ||
| "default": {}, | ||
| "critical": {}, | ||
| }, | ||
| initPending: map[string][]*base.TaskMessage{ | ||
| "default": {}, | ||
| "critical": {t4}, | ||
| }, | ||
| wait: pollInterval * 2, | ||
| wantScheduled: map[string][]*base.TaskMessage{ | ||
| "default": {}, | ||
| "critical": {}, | ||
| }, | ||
| wantRetry: map[string][]*base.TaskMessage{ | ||
| "default": {}, | ||
| "critical": {}, | ||
| }, | ||
| wantPending: map[string][]*base.TaskMessage{ | ||
| "default": {t1, t3}, | ||
| "critical": {t2, t4}, | ||
| }, | ||
| }, | ||
| } | ||
|
|
||
| for _, tc := range tests { | ||
| h.FlushDB(t, r) // clean up db before each test case. | ||
| h.SeedAllScheduledQueues(t, r, tc.initScheduled) // initialize scheduled queue | ||
| h.SeedAllRetryQueues(t, r, tc.initRetry) // initialize retry queue | ||
| h.SeedAllPendingQueues(t, r, tc.initPending) // initialize default queue | ||
|
|
||
| var wg sync.WaitGroup | ||
| s.start(&wg) | ||
| time.Sleep(tc.wait) | ||
| s.terminate() | ||
|
|
||
| for qname, want := range tc.wantScheduled { | ||
| gotScheduled := h.GetScheduledMessages(t, r, qname) | ||
| if diff := cmp.Diff(want, gotScheduled, h.SortMsgOpt); diff != "" { | ||
| t.Errorf("mismatch found in %q after running forwarder: (-want, +got)\n%s", base.ScheduledKey(qname), diff) | ||
| } | ||
| } | ||
|
|
||
| for qname, want := range tc.wantRetry { | ||
| gotRetry := h.GetRetryMessages(t, r, qname) | ||
| if diff := cmp.Diff(want, gotRetry, h.SortMsgOpt); diff != "" { | ||
| t.Errorf("mismatch found in %q after running forwarder: (-want, +got)\n%s", base.RetryKey(qname), diff) | ||
| } | ||
| } | ||
|
|
||
| for qname, want := range tc.wantPending { | ||
| gotPending := h.GetPendingMessages(t, r, qname) | ||
| if diff := cmp.Diff(want, gotPending, h.SortMsgOpt); diff != "" { | ||
| t.Errorf("mismatch found in %q after running forwarder: (-want, +got)\n%s", base.QueueKey(qname), diff) | ||
| } | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.