rmq

package module
v1.0.4 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 15, 2019 License: MIT Imports: 11 Imported by: 0

README

Build Status GoDoc

Note: We recently updated rmq to use latest go-redis client github.com/go-redis/redis. If you don't want to upgrade yet, you can continue using rmq branch v1:

  1. Using a dependency manager, i.e. dep

    # Gopkg.toml
[[constraint]]
  name = "github.com/adjust/rmq"
  branch = "v1"

  2. Using gopkg.in

    import "gopkg.in/adjust/rmq.v1"

    See https://gopkg.in/adjust/rmq.v1

Overview

rmq is short for Redis message queue. It's a message queue system written in Go and backed by Redis. It's similar to redismq, but implemented independently with a different interface in mind.

Basic Usage

Lets take a look at how to use rmq.

Import

Of course you need to import rmq wherever you want to use it.

import "github.com/adjust/rmq"

Connection

Before we get to queues, we first need to establish a connection. Each rmq connection has a name (used in statistics) and Redis connection details including which Redis database to use. The most basic Redis connection uses a TCP connection to a given host and a port.

connection := rmq.OpenConnection("my service", "tcp", "localhost:6379", 1)

But it's also possible to access a Redis listening on a Unix socket.

connection := rmq.OpenConnection("my service", "unix", "/tmp/redis.sock", 1)

Note: rmq panics on Redis connection errors. Your producers and consumers will crash if Redis goes down. Please let us know if you would see this handled differently.

Queue

Once we have a connection we can use it to finally access queues. Each queue must have a unique name by which we address it. Queues are created once they are accessed. There is no need to declare them in advance. Here we open a queue named "tasks":

taskQueue := connection.OpenQueue("tasks")

Producer

An empty queue is boring, lets add some deliveries! Internally all deliveries are saved to Redis as strings. This is how you can publish a string payload to a queue:

delivery := "task payload"
taskQueue.Publish(delivery)

In practice, however, it's more common to have instances of some struct that we want to publish to a queue. Assuming task is of some type like Kind, this is how to publish the JSON representation of that task:

// create task
taskBytes, err := json.Marshal(task)
if err != nil {
    // handle error
    return
}

taskQueue.PublishBytes(taskBytes)

For a full example see example/producer

Consumer

Now that our queue starts filling, lets add a consumer. After opening the queue as before, we need it to start consuming before we can add consumers.

taskQueue.StartConsuming(10, time.Second)

This sets the prefetch limit to 10 and the poll duration to one second. This means the queue will fetch up to 10 deliveries at a time before giving them to the consumers. To avoid idling producers in times of full queues, the prefetch limit should always be greater than the number of consumers you are going to add. If the queue gets empty, the poll duration sets how long to wait before checking for new deliveries in Redis.

Once this is set up, we can actually add consumers to the consuming queue.

taskConsumer := &TaskConsumer{}
taskQueue.AddConsumer("task consumer", taskConsumer)

For our example this assumes that you have a struct TaskConsumer that implements the rmq.Consumer interface like this:

func (consumer *TaskConsumer) Consume(delivery rmq.Delivery) {
    var task Task
    if err = json.Unmarshal([]byte(delivery.Payload()), &task); err != nil {
        // handle error
        delivery.Reject()
        return
    }

    // perform task
    log.Printf("performing task %s", task)
    delivery.Ack()
}

First we unmarshal the JSON package found in the delivery payload. If this fails we reject the delivery, otherwise we perform the task and ack the delivery.

If you don't actually need a consumer struct you can just call AddConsumerFunc instead and pass in a consumer function which directly handles an rmq.Delivery:

taskQueue.AddConsumerFunc(func(delivery rmq.Delivery) {
    // handle delivery and call Ack() or Reject() on it
})

For a full example see example/consumer

Stop Consuming

If you want to stop consuming from the queue, you can call StopConsuming:

finishedChan := taskQueue.StopConsuming()

When StopConsuming is called, it will continue fetching its current batch of deliveries. After that the consumers continue to consume the fetched deliveries until all unacked deliveries are fully consumed. If StopConsuming is called before consuming or after already stopped, it will return a closed channel. If you want to wait until all consumers are idle you can wait on the finishedChan:

  <-finishedChan

This is useful to implement a graceful shutdown of a consumer service. Please note that after calling StopConsuming the queue might not be in a state where you can add consumers and call StartConsuming again. If you have a use case where you actually need that sort of flexibility, please let us know. Currently for each queue you are only supposed to call StartConsuming and StopConsuming at most once.

Testing Included

To simplify testing of queue producers and consumers we include test mocks.

Test Connection

As before, we first need a queue connection, but this time we use a rmq.TestConnection that doesn't need any connection settings.

testConn := rmq.NewTestConnection()

If you are using a testing framework that uses test suites, you can reuse that test connection by setting it up once for the suite and resetting it with testConn.Reset before each test.

Producer Test

Now lets say we want to test the function publishTask that creates a task and publishes it to a queue from that connection.

// call the function that should publish a task
publishTask(testConn)

// check that the task is published
c.Check(suite.testConn.GetDelivery("tasks", 0), Equals, "task payload")

The c.Check part is from gocheck, but it will look similar for other testing frameworks. Given a rmq.TestConnection, we can check the deliveries that were published to it's queues (since the last Reset() call) with GetDelivery(queue, index). In this case we want to extract the first (and possibly only) delivery that was published to queue tasks and just check the payload string.

If the payload is JSON again, the unmarshalling and check might look like this:

var task Task
err := json.Unmarshal([]byte(suite.testConn.GetDelivery("tasks", 0)), &task)
c.Assert(err, IsNil)
c.Assert(task, NotNil)
c.Check(task.Property, Equals, "value")

If you expect a producer to create multiple deliveries you can use different indexes to access them all.

c.Check(suite.testConn.GetDelivery("tasks", 0), Equals, "task1")
c.Check(suite.testConn.GetDelivery("tasks", 1), Equals, "task2")

For convenience there's also a function GetDeliveries that returns all published deliveries to a queue as string array.

c.Check(suite.testConn.GetDeliveries("tasks"), DeepEquals, []string{"task1", "task2"})

If your producer doesn't have guarantees about the order of its deliveries, you could implement a selector function like findByPrefix and then check each delivery regardless of their index.

tasks := suite.testConn.GetDeliveries("tasks")
c.Assert(tasks, HasLen, 2)
xTask := findByPrefix(tasks, "x")
yTask := findByPrefix(tasks, "y")
c.Check(xTask.Id, Equals, "3")
c.Check(yTask.Id, Equals, "4")

These examples assumed that you inject the rmq.Connection into your testable functions. If you inject instances of rmq.Queue instead, you can use rmq.TestQueue instances in tests and access their LastDeliveries (since Reset()) directly.

Consumer Test

Testing consumers is a bit easier because consumers must implement the rmq.Consumer interface. In the tests just create rmq.TestDelivery and pass it to your Consume implementation. This example creates a test delivery from a string and then checks that the delivery was acked.

consumer := &TaskConsumer{}
delivery := rmq.NewTestDeliveryString("task payload")

consumer.Consume(delivery)

c.Check(delivery.State, Equals, rmq.Acked)

The State field will always be one of these values:

  • rmq.Acked: The delivery was acked
  • rmq.Rejected: The delivery was rejected
  • rmq.Pushed: The delivery was pushed (see below)
  • rmq.Unacked: Nothing of the above

If your packages are JSON marshalled objects, then you can create test deliveries out of those like this:

task := Task{Property: "bad value"}
delivery := rmq.NewTestDelivery(task)

Integration tests

If you want to write integration tests which exercise both producers and consumers at the same time, you can use the OpenConnectionWithTestRedisClient constructor. It returns a real rmq.Connection instance which is backed by an in memory Redis client implementation. That way it behaves exactly as in production, just without the durability of a real Redis client. Don't use this in production!

Statistics

Given a connection, you can call connection.CollectStats to receive rmq.Stats about all open queues, connections and consumers. If you run example/handler you can see what's available:

In this example you see three connections consuming things, each wich 10 consumers each. Two of them have 8 packages unacked. Below the marker you see connections which are not consuming. One of the handler connections died because I stopped the handler. Running the cleaner would clean that up (see below).

TODO

There are some features and aspects not properly documented yet. I will quickly list those here, hopefully they will be expanded in the future. 😉

  • Batch Consumers: Use queue.AddBatchConsumer() to register a consumer that receives batches of deliveries to be consumed at once (database bulk insert) See example/batch_consumer
  • Push Queues: When consuming queue A you can set up its push queue to be queue B. The consumer can then call delivery.Push() to push this delivery (originally from queue A) to the associated push queue B. (useful for retries)
  • Cleaner: Run this regularly to return unacked deliveries of stopped or crashed consumers back to ready so they can be consumed by a new consumer. See example/cleaner
  • Returner: Imagine there was some error that made you reject a lot of deliveries by accident. Just call queue.ReturnRejected() to return all rejected deliveries of that queue back to ready. (Similar to ReturnUnacked which is used by the cleaner) Consider using push queues if you do this regularly. See example/returner
  • Purger: If deliveries failed you don't want to retry them anymore for whatever reason, you can call queue.PurgeRejected() to dispose of them for good. There's also queue.PurgeReady if you want to get a queue clean without consuming possibly bad deliveries. See example/purger

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func ActiveSign 

func ActiveSign(active bool) string

func CleanConnection added in v1.0.1 

func CleanConnection(connection *redisConnection) error

func CleanQueue added in v1.0.1 

func CleanQueue(queue *redisQueue)

func OpenConnection 

func OpenConnection(tag, network, address string, db int) *redisConnection

OpenConnection opens and returns a new connection

func OpenConnectionWithRedisClient 

func OpenConnectionWithRedisClient(tag string, redisClient *redis.Client) *redisConnection

OpenConnectionWithRedisClient opens and returns a new connection

func OpenConnectionWithTestRedisClient added in v1.0.1 

func OpenConnectionWithTestRedisClient(tag string) *redisConnection

OpenConnectionWithTestRedisClient opens and returns a new connection which uses a test redis client internally. This is useful in integration tests.

Types

type BatchConsumer 

type BatchConsumer interface {
	Consume(batch Deliveries)
}

type Cleaner 

type Cleaner struct {
	// contains filtered or unexported fields
}

func NewCleaner 

func NewCleaner(connection Connection) *Cleaner

func (*Cleaner) Clean 

func (cleaner *Cleaner) Clean() error

type Connection 

type Connection interface {
	OpenQueue(name string) Queue
	CollectStats(queueList []string) Stats
	GetOpenQueues() []string
}

Connection is an interface that can be used to test publishing

type ConnectionStat 

type ConnectionStat struct {
	// contains filtered or unexported fields
}

func (ConnectionStat) String 

func (stat ConnectionStat) String() string

type ConnectionStats 

type ConnectionStats map[string]ConnectionStat

type Consumer 

type Consumer interface {
	Consume(delivery Delivery)
}

type ConsumerFunc added in v1.0.1 

type ConsumerFunc func(Delivery)

func (ConsumerFunc) Consume added in v1.0.1 

func (consumerFunc ConsumerFunc) Consume(delivery Delivery)

type Deliveries 

type Deliveries []Delivery

func (Deliveries) Ack 

func (deliveries Deliveries) Ack() int

func (Deliveries) Push added in v1.0.1 

func (deliveries Deliveries) Push() int

func (Deliveries) Reject 

func (deliveries Deliveries) Reject() int

type Delivery 

type Delivery interface {
	Payload() string
	Ack() bool
	Reject() bool
	Push() bool
}

type Queue 

type Queue interface {
	Publish(payload string) bool
	PublishBytes(payload []byte) bool
	SetPushQueue(pushQueue Queue)
	StartConsuming(prefetchLimit int, pollDuration time.Duration) bool
	StopConsuming() <-chan struct{}
	AddConsumer(tag string, consumer Consumer) string
	AddConsumerFunc(tag string, consumerFunc ConsumerFunc) string
	AddBatchConsumer(tag string, batchSize int, consumer BatchConsumer) string
	AddBatchConsumerWithTimeout(tag string, batchSize int, timeout time.Duration, consumer BatchConsumer) string
	PurgeReady() int
	PurgeRejected() int
	ReturnRejected(count int) int
	ReturnAllRejected() int
	ReturnAllUnacked() int
	Close() bool
}

type QueueStat 

type QueueStat struct {
	ReadyCount    int `json:"ready"`
	RejectedCount int `json:"rejected"`
	// contains filtered or unexported fields
}

func NewQueueStat 

func NewQueueStat(readyCount, rejectedCount int) QueueStat

func (QueueStat) ConnectionCount 

func (stat QueueStat) ConnectionCount() int

func (QueueStat) ConsumerCount 

func (stat QueueStat) ConsumerCount() int

func (QueueStat) String 

func (stat QueueStat) String() string

func (QueueStat) UnackedCount 

func (stat QueueStat) UnackedCount() int

type QueueStats 

type QueueStats map[string]QueueStat

type RedisClient added in v1.0.1 

type RedisClient interface {
	// simple keys
	Set(key string, value string, expiration time.Duration) bool
	Del(key string) (affected int, ok bool)      // default affected: 0
	TTL(key string) (ttl time.Duration, ok bool) // default ttl: 0

	// lists
	LPush(key, value string) bool
	LLen(key string) (affected int, ok bool)
	LRem(key string, count int, value string) (affected int, ok bool)
	LTrim(key string, start, stop int)
	RPopLPush(source, destination string) (value string, ok bool)

	// sets
	SAdd(key, value string) bool
	SMembers(key string) (members []string)         // default members: []string{}
	SRem(key, value string) (affected int, ok bool) // default affected: 0

	// special
	FlushDb()
}

type RedisWrapper added in v1.0.1 

type RedisWrapper struct {
	// contains filtered or unexported fields
}

func (RedisWrapper) Del added in v1.0.1 

func (wrapper RedisWrapper) Del(key string) (affected int, ok bool)

func (RedisWrapper) FlushDb added in v1.0.1 

func (wrapper RedisWrapper) FlushDb()

func (RedisWrapper) LLen added in v1.0.1 

func (wrapper RedisWrapper) LLen(key string) (affected int, ok bool)

func (RedisWrapper) LPush added in v1.0.1 

func (wrapper RedisWrapper) LPush(key, value string) bool

func (RedisWrapper) LRem added in v1.0.1 

func (wrapper RedisWrapper) LRem(key string, count int, value string) (affected int, ok bool)

func (RedisWrapper) LTrim added in v1.0.1 

func (wrapper RedisWrapper) LTrim(key string, start, stop int)

func (RedisWrapper) RPopLPush added in v1.0.1 

func (wrapper RedisWrapper) RPopLPush(source, destination string) (value string, ok bool)

func (RedisWrapper) SAdd added in v1.0.1 

func (wrapper RedisWrapper) SAdd(key, value string) bool

func (RedisWrapper) SMembers added in v1.0.1 

func (wrapper RedisWrapper) SMembers(key string) []string

func (RedisWrapper) SRem added in v1.0.1 

func (wrapper RedisWrapper) SRem(key, value string) (affected int, ok bool)

func (RedisWrapper) Set added in v1.0.1 

func (wrapper RedisWrapper) Set(key string, value string, expiration time.Duration) bool

func (RedisWrapper) TTL added in v1.0.1 

func (wrapper RedisWrapper) TTL(key string) (ttl time.Duration, ok bool)

type State 

type State int
const (
	Unacked State = iota
	Acked
	Rejected
	Pushed
)

func (State) String 

func (i State) String() string

type Stats 

type Stats struct {
	QueueStats QueueStats `json:"queues"`
	// contains filtered or unexported fields
}

func CollectStats 

func CollectStats(queueList []string, mainConnection *redisConnection) Stats

func NewStats 

func NewStats() Stats

func (Stats) GetHtml 

func (stats Stats) GetHtml(layout, refresh string) string

func (Stats) String 

func (stats Stats) String() string

type TestBatchConsumer 

type TestBatchConsumer struct {
	LastBatch     Deliveries
	ConsumedCount int
	AutoFinish    bool
	// contains filtered or unexported fields
}

func NewTestBatchConsumer 

func NewTestBatchConsumer() *TestBatchConsumer

func (*TestBatchConsumer) Consume 

func (consumer *TestBatchConsumer) Consume(batch Deliveries)

func (*TestBatchConsumer) Finish 

func (consumer *TestBatchConsumer) Finish()

type TestConnection 

type TestConnection struct {
	// contains filtered or unexported fields
}

func NewTestConnection 

func NewTestConnection() TestConnection

func (TestConnection) CollectStats 

func (connection TestConnection) CollectStats(queueList []string) Stats

func (TestConnection) GetDeliveries 

func (connection TestConnection) GetDeliveries(queueName string) []string

func (TestConnection) GetDelivery 

func (connection TestConnection) GetDelivery(queueName string, index int) string

func (TestConnection) GetOpenQueues 

func (connection TestConnection) GetOpenQueues() []string

func (TestConnection) OpenQueue 

func (connection TestConnection) OpenQueue(name string) Queue

func (TestConnection) Reset 

func (connection TestConnection) Reset()

type TestConsumer 

type TestConsumer struct {
	AutoAck       bool
	AutoFinish    bool
	SleepDuration time.Duration

	LastDelivery   Delivery
	LastDeliveries []Delivery
	// contains filtered or unexported fields
}

func NewTestConsumer 

func NewTestConsumer(name string) *TestConsumer

func (*TestConsumer) Consume 

func (consumer *TestConsumer) Consume(delivery Delivery)

func (*TestConsumer) Finish 

func (consumer *TestConsumer) Finish()

func (*TestConsumer) String 

func (consumer *TestConsumer) String() string

type TestDelivery 

type TestDelivery struct {
	State State
	// contains filtered or unexported fields
}

func NewTestDelivery 

func NewTestDelivery(content interface{}) *TestDelivery

func NewTestDeliveryString 

func NewTestDeliveryString(payload string) *TestDelivery

func (*TestDelivery) Ack 

func (delivery *TestDelivery) Ack() bool

func (*TestDelivery) Payload 

func (delivery *TestDelivery) Payload() string

func (*TestDelivery) Push 

func (delivery *TestDelivery) Push() bool

func (*TestDelivery) Reject 

func (delivery *TestDelivery) Reject() bool

type TestQueue 

type TestQueue struct {
	LastDeliveries []string
	// contains filtered or unexported fields
}

func NewTestQueue 

func NewTestQueue(name string) *TestQueue

func (*TestQueue) AddBatchConsumer 

func (queue *TestQueue) AddBatchConsumer(tag string, batchSize int, consumer BatchConsumer) string

func (*TestQueue) AddBatchConsumerWithTimeout 

func (queue *TestQueue) AddBatchConsumerWithTimeout(tag string, batchSize int, timeout time.Duration, consumer BatchConsumer) string

func (*TestQueue) AddConsumer 

func (queue *TestQueue) AddConsumer(tag string, consumer Consumer) string

func (*TestQueue) AddConsumerFunc added in v1.0.1 

func (queue *TestQueue) AddConsumerFunc(tag string, consumerFunc ConsumerFunc) string

func (*TestQueue) Close 

func (queue *TestQueue) Close() bool

func (*TestQueue) Publish 

func (queue *TestQueue) Publish(payload string) bool

func (*TestQueue) PublishBytes 

func (queue *TestQueue) PublishBytes(payload []byte) bool

func (*TestQueue) PurgeReady 

func (queue *TestQueue) PurgeReady() int

func (*TestQueue) PurgeRejected 

func (queue *TestQueue) PurgeRejected() int

func (*TestQueue) Reset 

func (queue *TestQueue) Reset()

func (*TestQueue) ReturnAllRejected 

func (queue *TestQueue) ReturnAllRejected() int

func (*TestQueue) ReturnAllUnacked added in v1.0.4 

func (queue *TestQueue) ReturnAllUnacked() int

func (*TestQueue) ReturnRejected 

func (queue *TestQueue) ReturnRejected(count int) int

func (*TestQueue) SetPushQueue 

func (queue *TestQueue) SetPushQueue(pushQueue Queue)

func (*TestQueue) StartConsuming 

func (queue *TestQueue) StartConsuming(prefetchLimit int, pollDuration time.Duration) bool

func (*TestQueue) StopConsuming 

func (queue *TestQueue) StopConsuming() <-chan struct{}

func (*TestQueue) String 

func (queue *TestQueue) String() string

type TestRedisClient added in v1.0.1 

type TestRedisClient struct {
	// contains filtered or unexported fields
}

TestRedisClient is a mock for redis

func NewTestRedisClient added in v1.0.1 

func NewTestRedisClient() *TestRedisClient

NewTestRedisClient returns a NewTestRedisClient

func (*TestRedisClient) Del added in v1.0.1 

func (client *TestRedisClient) Del(key string) (affected int, ok bool)

Del removes the specified key. A key is ignored if it does not exist.

func (*TestRedisClient) FlushDb added in v1.0.1 

func (client *TestRedisClient) FlushDb()

FlushDb delete all the keys of the currently selected DB. This command never fails.

func (*TestRedisClient) Get added in v1.0.1 

func (client *TestRedisClient) Get(key string) string

Get the value of key. If the key does not exist or isn't a string the special value nil is returned.

func (*TestRedisClient) LLen added in v1.0.1 

func (client *TestRedisClient) LLen(key string) (affected int, ok bool)

LLen returns the length of the list stored at key. If key does not exist, it is interpreted as an empty list and 0 is returned. An error is returned when the value stored at key is not a list.

func (*TestRedisClient) LPush added in v1.0.1 

func (client *TestRedisClient) LPush(key, value string) bool

LPush inserts the specified value at the head of the list stored at key. If key does not exist, it is created as empty list before performing the push operations. When key holds a value that is not a list, an error is returned. It is possible to push multiple elements using a single command call just specifying multiple arguments at the end of the command. Elements are inserted one after the other to the head of the list, from the leftmost element to the rightmost element.

func (*TestRedisClient) LRange added in v1.0.1 

func (client *TestRedisClient) LRange(key string, start, end int) []string

LRange returns the specified elements of the list stored at key. The offsets start and stop are zero-based indexes, with 0 being the first element of the list (the head of the list), 1 being the next element and so on. These offsets can also be negative numbers indicating offsets starting at the end of the list. For example, -1 is the last element of the list, -2 the penultimate, and so on.

func (*TestRedisClient) LRem added in v1.0.1 

func (client *TestRedisClient) LRem(key string, count int, value string) (affected int, ok bool)

LRem removes the first count occurrences of elements equal to value from the list stored at key. The count argument influences the operation in the following ways: count > 0: Remove elements equal to value moving from head to tail. count < 0: Remove elements equal to value moving from tail to head. count = 0: Remove all elements equal to value. For example, LREM list -2 "hello" will remove the last two occurrences of "hello" in the list stored at list. Note that non-existing keys are treated like empty lists, so when key does not exist, the command will always return 0.

func (*TestRedisClient) LTrim added in v1.0.1 

func (client *TestRedisClient) LTrim(key string, start, stop int)

LTrim trims an existing list so that it will contain only the specified range of elements specified. Both start and stop are zero-based indexes, where 0 is the first element of the list (the head), 1 the next element and so on. For example: LTRIM foobar 0 2 will modify the list stored at foobar so that only the first three elements of the list will remain. start and end can also be negative numbers indicating offsets from the end of the list, where -1 is the last element of the list, -2 the penultimate element and so on. Out of range indexes will not produce an error: if start is larger than the end of the list, or start > end, the result will be an empty list (which causes key to be removed). If end is larger than the end of the list, Redis will treat it like the last element of the list

func (*TestRedisClient) RPopLPush added in v1.0.1 

func (client *TestRedisClient) RPopLPush(source, destination string) (value string, ok bool)

RPopLPush atomically returns and removes the last element (tail) of the list stored at source, and pushes the element at the first element (head) of the list stored at destination. For example: consider source holding the list a,b,c, and destination holding the list x,y,z. Executing RPOPLPUSH results in source holding a,b and destination holding c,x,y,z. If source does not exist, the value nil is returned and no operation is performed. If source and destination are the same, the operation is equivalent to removing the last element from the list and pushing it as first element of the list, so it can be considered as a list rotation command.

func (*TestRedisClient) SAdd added in v1.0.1 

func (client *TestRedisClient) SAdd(key, value string) bool

SAdd adds the specified members to the set stored at key. Specified members that are already a member of this set are ignored. If key does not exist, a new set is created before adding the specified members. An error is returned when the value stored at key is not a set.

func (*TestRedisClient) SMembers added in v1.0.1 

func (client *TestRedisClient) SMembers(key string) (members []string)

SMembers returns all the members of the set value stored at key. This has the same effect as running SINTER with one argument key.

func (*TestRedisClient) SRem added in v1.0.1 

func (client *TestRedisClient) SRem(key, value string) (affected int, ok bool)

SRem removes the specified members from the set stored at key. Specified members that are not a member of this set are ignored. If key does not exist, it is treated as an empty set and this command returns 0. An error is returned when the value stored at key is not a set.

func (*TestRedisClient) Set added in v1.0.1 

func (client *TestRedisClient) Set(key string, value string, expiration time.Duration) bool

Set sets key to hold the string value. If key already holds a value, it is overwritten, regardless of its type. Any previous time to live associated with the key is discarded on successful SET operation.

func (*TestRedisClient) TTL added in v1.0.1 

func (client *TestRedisClient) TTL(key string) (ttl time.Duration, ok bool)

TTL returns the remaining time to live of a key that has a timeout. This introspection capability allows a Redis client to check how many seconds a given key will continue to be part of the dataset. In Redis 2.6 or older the command returns -1 if the key does not exist or if the key exist but has no associated expire. Starting with Redis 2.8 the return value in case of error changed: The command returns -2 if the key does not exist. The command returns -1 if the key exists but has no associated expire.

Source Files

View all Source files

Directories

Path Synopsis
example
batch_consumer
cleaner
consumer
handler
producer
purger
returner

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.