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

Inspector support Go context

Browse Source
This commit is contained in:
youngxu 2022-06-21 14:10:31 +08:00
parent 7d783d6ae9
commit 7922c858f7
6 changed files with 9 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
inspector.go
View File

@ -948,7 +948,7 @@ func (i *Inspector) CancelProcessing(id string) error {
// guarantee that the task with the given id will be canceled. The return // guarantee that the task with the given id will be canceled. The return
// value only indicates whether the cancelation signal has been sent. // value only indicates whether the cancelation signal has been sent.
func (i *Inspector) CancelProcessingContext(ctx context.Context, id string) error { func (i *Inspector) CancelProcessingContext(ctx context.Context, id string) error {
return i.rdb.PublishCancelationContext(ctx, id) return i.rdb.PublishCancelation(ctx, id)
} }
// PauseQueue pauses task processing on the specified queue. // PauseQueue pauses task processing on the specified queue.

2
internal/base/base.go
View File

@ -749,7 +749,7 @@ type Broker interface {
// Cancelation related methods // Cancelation related methods
CancelationPubSub() (*redis.PubSub, error) // TODO: Need to decouple from redis to support other brokers CancelationPubSub() (*redis.PubSub, error) // TODO: Need to decouple from redis to support other brokers
PublishCancelation(id string) error PublishCancelation(ctx context.Context, id string) error
WriteResult(qname, id string, data []byte) (n int, err error) WriteResult(qname, id string, data []byte) (n int, err error)
} }

8
internal/rdb/rdb.go
View File

@ -1459,13 +1459,7 @@ func (r *RDB) CancelationPubSub() (*redis.PubSub, error) {
// PublishCancelation publish cancelation message to all subscribers. // PublishCancelation publish cancelation message to all subscribers.
// The message is the ID for the task to be canceled. // The message is the ID for the task to be canceled.
func (r *RDB) PublishCancelation(id string) error { func (r *RDB) PublishCancelation(ctx context.Context, id string) error {
return r.PublishCancelationContext(context.Background(), id)
}
// PublishCancelationContext publish cancelation message to all subscribers.
// The message is the ID for the task to be canceled.
func (r *RDB) PublishCancelationContext(ctx context.Context, id string) error {
var op errors.Op = "rdb.PublishCancelation" var op errors.Op = "rdb.PublishCancelation"
if err := r.client.Publish(ctx, base.CancelChannel, id).Err(); err != nil { if err := r.client.Publish(ctx, base.CancelChannel, id).Err(); err != nil {
return errors.E(op, errors.Unknown, fmt.Sprintf("redis pubsub publish error: %v", err)) return errors.E(op, errors.Unknown, fmt.Sprintf("redis pubsub publish error: %v", err))

2
internal/rdb/rdb_test.go
View File

@ -3053,7 +3053,7 @@ func TestCancelationPubSub(t *testing.T) {
publish := []string{"one", "two", "three"} publish := []string{"one", "two", "three"}
for _, msg := range publish { for _, msg := range publish {
r.PublishCancelation(msg) r.PublishCancelation(context.Background(), msg)
} }
// allow for message to reach subscribers. // allow for message to reach subscribers.

4
internal/testbroker/testbroker.go
View File

@ -199,13 +199,13 @@ func (tb *TestBroker) CancelationPubSub() (*redis.PubSub, error) {
return tb.real.CancelationPubSub() return tb.real.CancelationPubSub()
} }
func (tb *TestBroker) PublishCancelation(id string) error { func (tb *TestBroker) PublishCancelation(ctx context.Context, id string) error {
tb.mu.Lock() tb.mu.Lock()
defer tb.mu.Unlock() defer tb.mu.Unlock()
if tb.sleeping { if tb.sleeping {
return errRedisDown return errRedisDown
} }
return tb.real.PublishCancelation(id) return tb.real.PublishCancelation(ctx, id)
} }
func (tb *TestBroker) WriteResult(qname, id string, data []byte) (int, error) { func (tb *TestBroker) WriteResult(qname, id string, data []byte) (int, error) {

5
subscriber_test.go
View File

@ -5,6 +5,7 @@
package asynq package asynq
import ( import (
"context"
"sync" "sync"
"testing" "testing"
"time" "time"
@ -51,7 +52,7 @@ func TestSubscriber(t *testing.T) {
// wait for subscriber to establish connection to pubsub channel // wait for subscriber to establish connection to pubsub channel
time.Sleep(time.Second) time.Sleep(time.Second)
if err := rdbClient.PublishCancelation(tc.publishID); err != nil { if err := rdbClient.PublishCancelation(context.Background(), tc.publishID); err != nil {
t.Fatalf("could not publish cancelation message: %v", err) t.Fatalf("could not publish cancelation message: %v", err)
} }
@ -110,7 +111,7 @@ func TestSubscriberWithRedisDown(t *testing.T) {
called = true called = true
}) })
if err := r.PublishCancelation(id); err != nil { if err := r.PublishCancelation(context.Background(), id); err != nil {
t.Fatalf("could not publish cancelation message: %v", err) t.Fatalf("could not publish cancelation message: %v", err)
} }