go-pkg/asynq
2
0
Fork 0
mirror of https://github.com/hibiken/asynq.git synced 2024-12-27 00:02:19 +08:00
Code Issues Projects Releases Wiki Activity

Fix build

Browse Source
This commit is contained in:
Ken Hibino 2021-05-10 21:19:57 -07:00
parent 4af65d5fa5
commit 1812d05d21
4 changed files with 17 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
client.go
View File

@ -5,7 +5,6 @@
package asynq package asynq
import ( import (
"errors"
"fmt" "fmt"
"strings" "strings"
"sync" "sync"
@ -14,6 +13,7 @@ import (
"github.com/go-redis/redis/v7" "github.com/go-redis/redis/v7"
"github.com/google/uuid" "github.com/google/uuid"
"github.com/hibiken/asynq/internal/base" "github.com/hibiken/asynq/internal/base"
"github.com/hibiken/asynq/internal/errors"
"github.com/hibiken/asynq/internal/rdb" "github.com/hibiken/asynq/internal/rdb"
) )
@ -346,7 +346,7 @@ func (c *Client) Enqueue(task *Task, opts ...Option) (*Result, error) {
err = c.schedule(msg, opt.processAt, opt.uniqueTTL) err = c.schedule(msg, opt.processAt, opt.uniqueTTL)
} }
switch { switch {
case err == rdb.ErrDuplicateTask: case errors.Is(err, errors.ErrDuplicateTask):
return nil, fmt.Errorf("%w", ErrDuplicateTask) return nil, fmt.Errorf("%w", ErrDuplicateTask)
case err != nil: case err != nil:
return nil, err return nil, err

20
inspeq/inspector.go
View File

@ -5,7 +5,6 @@
package inspeq package inspeq
import ( import (
"errors"
"fmt" "fmt"
"strconv" "strconv"
"strings" "strings"
@ -15,6 +14,7 @@ import (
"github.com/google/uuid" "github.com/google/uuid"
"github.com/hibiken/asynq" "github.com/hibiken/asynq"
"github.com/hibiken/asynq/internal/base" "github.com/hibiken/asynq/internal/base"
"github.com/hibiken/asynq/internal/errors"
"github.com/hibiken/asynq/internal/rdb" "github.com/hibiken/asynq/internal/rdb"
) )
@ -154,10 +154,10 @@ var (
// returns ErrQueueNotEmpty. // returns ErrQueueNotEmpty.
func (i *Inspector) DeleteQueue(qname string, force bool) error { func (i *Inspector) DeleteQueue(qname string, force bool) error {
err := i.rdb.RemoveQueue(qname, force) err := i.rdb.RemoveQueue(qname, force)
if _, ok := err.(*rdb.QueueNotFoundError); ok { if errors.IsQueueNotFound(err) {
return fmt.Errorf("%w: queue=%q", ErrQueueNotFound, qname) return fmt.Errorf("%w: queue=%q", ErrQueueNotFound, qname)
} }
if _, ok := err.(*rdb.QueueNotEmptyError); ok { if errors.IsQueueNotEmpty(err) {
return fmt.Errorf("%w: queue=%q", ErrQueueNotEmpty, qname) return fmt.Errorf("%w: queue=%q", ErrQueueNotEmpty, qname)
} }
return err return err
@ -549,13 +549,13 @@ func (i *Inspector) DeleteTaskByKey(qname, key string) error {
} }
switch prefix { switch prefix {
case keyPrefixPending: case keyPrefixPending:
return i.rdb.DeletePendingTask(qname, id) return i.rdb.DeleteTask(qname, id)
case keyPrefixScheduled: case keyPrefixScheduled:
return i.rdb.DeleteScheduledTask(qname, id) return i.rdb.DeleteTask(qname, id)
case keyPrefixRetry: case keyPrefixRetry:
return i.rdb.DeleteRetryTask(qname, id) return i.rdb.DeleteTask(qname, id)
case keyPrefixArchived: case keyPrefixArchived:
return i.rdb.DeleteArchivedTask(qname, id) return i.rdb.DeleteTask(qname, id)
default: default:
return fmt.Errorf("invalid key") return fmt.Errorf("invalid key")
} }
@ -657,11 +657,11 @@ func (i *Inspector) ArchiveTaskByKey(qname, key string) error {
} }
switch prefix { switch prefix {
case keyPrefixPending: case keyPrefixPending:
return i.rdb.ArchivePendingTask(qname, id) return i.rdb.ArchiveTask(qname, id)
case keyPrefixScheduled: case keyPrefixScheduled:
return i.rdb.ArchiveScheduledTask(qname, id) return i.rdb.ArchiveTask(qname, id)
case keyPrefixRetry: case keyPrefixRetry:
return i.rdb.ArchiveRetryTask(qname, id) return i.rdb.ArchiveTask(qname, id)
case keyPrefixArchived: case keyPrefixArchived:
return fmt.Errorf("task is already archived") return fmt.Errorf("task is already archived")
default: default:

5
inspeq/inspector_test.go
View File

@ -5,6 +5,7 @@
package inspeq package inspeq
import ( import (
"errors"
"flag" "flag"
"fmt" "fmt"
"math" "math"
@ -271,7 +272,7 @@ func TestInspectorDeleteQueueErrorQueueNotEmpty(t *testing.T) {
h.SeedAllArchivedQueues(t, r, tc.archived) h.SeedAllArchivedQueues(t, r, tc.archived)
err := inspector.DeleteQueue(tc.qname, tc.force) err := inspector.DeleteQueue(tc.qname, tc.force)
if _, ok := err.(*ErrQueueNotEmpty); !ok { if !errors.Is(err, ErrQueueNotEmpty) {
t.Errorf("DeleteQueue(%v, %t) did not return ErrQueueNotEmpty", t.Errorf("DeleteQueue(%v, %t) did not return ErrQueueNotEmpty",
tc.qname, tc.force) tc.qname, tc.force)
} }
@ -327,7 +328,7 @@ func TestInspectorDeleteQueueErrorQueueNotFound(t *testing.T) {
h.SeedAllArchivedQueues(t, r, tc.archived) h.SeedAllArchivedQueues(t, r, tc.archived)
err := inspector.DeleteQueue(tc.qname, tc.force) err := inspector.DeleteQueue(tc.qname, tc.force)
if _, ok := err.(*ErrQueueNotFound); !ok { if !errors.Is(err, ErrQueueNotFound) {
t.Errorf("DeleteQueue(%v, %t) did not return ErrQueueNotFound", t.Errorf("DeleteQueue(%v, %t) did not return ErrQueueNotFound",
tc.qname, tc.force) tc.qname, tc.force)
} }

5
processor.go
View File

@ -6,7 +6,6 @@ package asynq
import ( import (
"context" "context"
"errors"
"fmt" "fmt"
"math/rand" "math/rand"
"runtime" "runtime"
@ -17,8 +16,8 @@ import (
"time" "time"
"github.com/hibiken/asynq/internal/base" "github.com/hibiken/asynq/internal/base"
"github.com/hibiken/asynq/internal/errors"
"github.com/hibiken/asynq/internal/log" "github.com/hibiken/asynq/internal/log"
"github.com/hibiken/asynq/internal/rdb"
"golang.org/x/time/rate" "golang.org/x/time/rate"
) )
@ -163,7 +162,7 @@ func (p *processor) exec() {
qnames := p.queues() qnames := p.queues()
msg, deadline, err := p.broker.Dequeue(qnames...) msg, deadline, err := p.broker.Dequeue(qnames...)
switch { switch {
case err == rdb.ErrNoProcessableTask: case errors.Is(err, errors.ErrNoProcessableTask):
p.logger.Debug("All queues are empty") p.logger.Debug("All queues are empty")
// Queues are empty, this is a normal behavior. // Queues are empty, this is a normal behavior.
// Sleep to avoid slamming redis and let scheduler move tasks into queues. // Sleep to avoid slamming redis and let scheduler move tasks into queues.