diff --git a/core/stores/redis/redis.go b/core/stores/redis/redis.go index b31eb874a21f..a779a883d695 100644 --- a/core/stores/redis/redis.go +++ b/core/stores/redis/redis.go @@ -267,6 +267,31 @@ func (s *Redis) BlpopExCtx(ctx context.Context, node RedisNode, key string) (str return vals[1], true, nil } +// BlpopWithTimeout uses passed in redis connection to execute blpop command. +// Control blocking query timeout +func (s *Redis) BlpopWithTimeout(node RedisNode, timeout time.Duration, key string) (string, error) { + return s.BlpopWithTimeoutCtx(context.Background(), node, timeout, key) +} + +// BlpopWithTimeoutCtx uses passed in redis connection to execute blpop command. +// Control blocking query timeout +func (s *Redis) BlpopWithTimeoutCtx(ctx context.Context, node RedisNode, timeout time.Duration, key string) (string, error) { + if node == nil { + return "", ErrNilNode + } + + vals, err := node.BLPop(ctx, timeout, key).Result() + if err != nil { + return "", err + } + + if len(vals) < 2 { + return "", fmt.Errorf("no value on key: %s", key) + } + + return vals[1], nil +} + // Decr is the implementation of redis decr command. func (s *Redis) Decr(key string) (int64, error) { return s.DecrCtx(context.Background(), key) diff --git a/core/stores/redis/redis_test.go b/core/stores/redis/redis_test.go index fdf90e71a4da..783fedde6eb4 100644 --- a/core/stores/redis/redis_test.go +++ b/core/stores/redis/redis_test.go @@ -1117,6 +1117,17 @@ func TestRedisBlpopEx(t *testing.T) { }) } +func TestRedisBlpopWithTimeout(t *testing.T) { + runOnRedis(t, func(client *Redis) { + client.Ping() + var node mockedNode + _, err := client.BlpopWithTimeout(nil, 10*time.Second, "foo") + assert.NotNil(t, err) + _, err = client.BlpopWithTimeout(node, 10*time.Second, "foo") + assert.NotNil(t, err) + }) +} + func TestRedisGeo(t *testing.T) { runOnRedis(t, func(client *Redis) { client.Ping()