radix

func Pipeline ¶

func Pipeline(cmds ...CmdAction) Action

Pipeline returns an Action which first writes multiple commands to a Conn in a single write, then reads their responses in a single read. This reduces network delay into a single round-trip.

Run will not be called on any of the passed in CmdActions.

NOTE that, while a Pipeline performs all commands on a single Conn, it shouldn't be used by itself for MULTI/EXEC transactions, because if there's an error it won't discard the incomplete transaction. Use WithConn or EvalScript for transactional functionality instead.

client, err := NewPool("tcp", "127.0.0.1:6379", 10) // or any other client
if err != nil {
	// handle error
}
var fooVal string
p := Pipeline(
	FlatCmd(nil, "SET", "foo", 1),
	Cmd(&fooVal, "GET", "foo"),
)
if err := client.Do(p); err != nil {
	// handle error
}
fmt.Printf("fooVal: %q\n", fooVal)

Output:

fooVal: "1"

func WithConn ¶

func WithConn(key string, fn func(Conn) error) Action

WithConn is used to perform a set of independent Actions on the same Conn. key should be a key which one or more of the inner Actions is acting on, or "" if no keys are being acted on. The callback function is what should actually carry out the inner actions, and the error it returns will be passed back up immediately.

NOTE that WithConn only ensures all inner Actions are performed on the same Conn, it doesn't make them transactional. Use MULTI/WATCH/EXEC within a WithConn for transactions, or use EvalScript

client, err := NewPool("tcp", "127.0.0.1:6379", 10) // or any other client
if err != nil {
	// handle error
}

// This example retrieves the current integer value of `key` and sets its
// new value to be the increment of that, all using the same connection
// instance. NOTE that it does not do this atomically like the INCR command
// would.
key := "someKey"
err = client.Do(WithConn(key, func(conn Conn) error {
	var curr int
	if err := conn.Do(Cmd(&curr, "GET", key)); err != nil {
		return err
	}

	curr++
	return conn.Do(FlatCmd(nil, "SET", key, curr))
}))
if err != nil {
	// handle error
}

Output:

client, err := NewPool("tcp", "127.0.0.1:6379", 10) // or any other client
if err != nil {
	// handle error
}

// This example retrieves the current value of `key` and then sets a new
// value on it in an atomic transaction.
key := "someKey"
var prevVal string

err = client.Do(WithConn(key, func(c Conn) error {

	// Begin the transaction with a MULTI command
	if err := c.Do(Cmd(nil, "MULTI")); err != nil {
		return err
	}

	// If any of the calls after the MULTI call error it's important that
	// the transaction is discarded. This isn't strictly necessary if the
	// error was a network error, as the connection would be closed by the
	// client anyway, but it's important otherwise.
	var err error
	defer func() {
		if err != nil {
			// The return from DISCARD doesn't matter. If it's an error then
			// it's a network error and the Conn will be closed by the
			// client.
			c.Do(Cmd(nil, "DISCARD"))
		}
	}()

	// queue up the transaction's commands
	if err = c.Do(Cmd(nil, "GET", key)); err != nil {
		return err
	}
	if err = c.Do(Cmd(nil, "SET", key, "someOtherValue")); err != nil {
		return err
	}

	// execute the transaction, capturing the result
	var result []string
	if err = c.Do(Cmd(&result, "EXEC")); err != nil {
		return err
	}

	// capture the output of the first transaction command, i.e. the GET
	prevVal = result[0]
	return nil
}))
if err != nil {
	// handle error
}

fmt.Printf("the value of key %q was %q\n", key, prevVal)

Output:

func NewCluster ¶

func NewCluster(clusterAddrs []string, opts ...ClusterOpt) (*Cluster, error)

NewCluster initializes and returns a Cluster instance. It will try every address given until it finds a usable one. From there it uses CLUSTER SLOTS to discover the cluster topology and make all the necessary connections.

NewCluster takes in a number of options which can overwrite its default behavior. The default options NewCluster uses are:

ClusterPoolFunc(DefaultClientFunc)
ClusterSyncEvery(5 * time.Second)
ClusterOnDownDelayActionsBy(100 * time.Millisecond)

func ClusterOnDownDelayActionsBy ¶

func ClusterOnDownDelayActionsBy(d time.Duration) ClusterOpt

ClusterOnDownDelayActionsBy tells the Cluster to delay all commands by the given duration while the cluster is seen to be in the CLUSTERDOWN state. This allows fewer actions to be affected by brief outages, e.g. during a failover.

If the given duration is 0 then Cluster will not delay actions during the CLUSTERDOWN state. Note that calls to Sync will not be delayed regardless of this option.

func ClusterPoolFunc ¶

func ClusterPoolFunc(pf ClientFunc) ClusterOpt

ClusterPoolFunc tells the Cluster to use the given ClientFunc when creating pools of connections to cluster members.

func ClusterSyncEvery ¶

func ClusterSyncEvery(d time.Duration) ClusterOpt

ClusterSyncEvery tells the Cluster to synchronize itself with the cluster's topology at the given interval. On every synchronization Cluster will ask the cluster for its topology and make/destroy its connections as necessary.

func ClusterWithTrace ¶

func ClusterWithTrace(ct trace.ClusterTrace) ClusterOpt

ClusterWithTrace tells the Cluster to trace itself with the given ClusterTrace. Note that ClusterTrace will block every point that you set to trace.

func Cmd ¶

func Cmd(rcv interface{}, cmd string, args ...string) CmdAction

Cmd is used to perform a redis command and retrieve a result. It should not be passed into Do more than once.

If the receiver value of Cmd is a primitive, a slice/map, or a struct then a pointer must be passed in. It may also be an io.Writer, an encoding.Text/BinaryUnmarshaler, or a resp.Unmarshaler. See the package docs for more on how results are unmarshaled into the receiver.

client, err := NewPool("tcp", "127.0.0.1:6379", 10) // or any other client
if err != nil {
	panic(err)
}

if err := client.Do(Cmd(nil, "SET", "foo", "bar")); err != nil {
	panic(err)
}

var fooVal string
if err := client.Do(Cmd(&fooVal, "GET", "foo")); err != nil {
	panic(err)
}
fmt.Println(fooVal)

Output:

bar

func FlatCmd ¶

func FlatCmd(rcv interface{}, cmd, key string, args ...interface{}) CmdAction

FlatCmd is like Cmd, but the arguments can be of almost any type, and FlatCmd will automatically flatten them into a single array of strings. Like Cmd, a FlatCmd should not be passed into Do more than once.

FlatCmd does _not_ work for commands whose first parameter isn't a key, or (generally) for MSET. Use Cmd for those.

FlatCmd supports using a resp.LenReader (an io.Reader with a Len() method) as an argument. *bytes.Buffer is an example of a LenReader, and the resp package has a NewLenReader function which can wrap an existing io.Reader.

FlatCmd also supports encoding.Text/BinaryMarshalers. It does _not_ currently support resp.Marshaler.

The receiver to FlatCmd follows the same rules as for Cmd.

client, err := NewPool("tcp", "127.0.0.1:6379", 10) // or any other client
if err != nil {
	panic(err)
}

// performs "SET" "foo" "1"
err = client.Do(FlatCmd(nil, "SET", "foo", 1))
if err != nil {
	panic(err)
}

// performs "SADD" "fooSet" "1" "2" "3"
err = client.Do(FlatCmd(nil, "SADD", "fooSet", []string{"1", "2", "3"}))
if err != nil {
	panic(err)
}

// performs "HMSET" "foohash" "a" "1" "b" "2" "c" "3"
m := map[string]int{"a": 1, "b": 2, "c": 3}
err = client.Do(FlatCmd(nil, "HMSET", "fooHash", m))
if err != nil {
	panic(err)
}

Output:

func Dial ¶

func Dial(network, addr string, opts ...DialOpt) (Conn, error)

Dial is a ConnFunc which creates a Conn using net.Dial and NewConn. It takes in a number of options which can overwrite its default behavior as well.

In place of a host:port address, Dial also accepts a URI, as per:

https://www.iana.org/assignments/uri-schemes/prov/redis

If the URI has an AUTH password or db specified Dial will attempt to perform the AUTH and/or SELECT as well.

If either DialAuthPass or DialSelectDB is used it overwrites the associated value passed in by the URI.

The default options Dial uses are:

DialTimeout(10 * time.Second)

func NewConn ¶

func NewConn(conn net.Conn) Conn

NewConn takes an existing net.Conn and wraps it to support the Conn interface of this package. The Read and Write methods on the original net.Conn should not be used after calling this method.

func PubSubStub ¶

func PubSubStub(remoteNetwork, remoteAddr string, fn func([]string) interface{}) (Conn, chan<- PubSubMessage)

PubSubStub returns a (fake) Conn, much like Stub does, which pretends it is a Conn to a real redis instance, but is instead using the given callback to service requests. It is primarily useful for writing tests.

PubSubStub differes from Stub in that Encode calls for (P)SUBSCRIBE, (P)UNSUBSCRIBE, MESSAGE, and PING will be intercepted and handled as per redis' expected pubsub functionality. A PubSubMessage may be written to the returned channel at any time, and if the PubSubStub has had (P)SUBSCRIBE called matching that PubSubMessage it will be written to the PubSubStub's internal buffer as expected.

This is intended to be used so that it can mock services which can perform both normal redis commands and pubsub (e.g. a real redis instance, redis sentinel). Once created this stub can be passed into PubSub and treated like a real connection.

// Make a pubsub stub conn which will return nil for everything except
// pubsub commands (which will be handled automatically)
stub, stubCh := PubSubStub("tcp", "127.0.0.1:6379", func([]string) interface{} {
	return nil
})

// These writes shouldn't do anything, initially, since we haven't
// subscribed to anything
go func() {
	for {
		stubCh <- PubSubMessage{
			Channel: "foo",
			Message: []byte("bar"),
		}
		time.Sleep(1 * time.Second)
	}
}()

// Use PubSub to wrap the stub like we would for a normal redis connection
pstub := PubSub(stub)

// Subscribe msgCh to "foo"
msgCh := make(chan PubSubMessage)
if err := pstub.Subscribe(msgCh, "foo"); err != nil {
	log.Fatal(err)
}

// now msgCh is subscribed the publishes being made by the go-routine above
// will start being written to it
for m := range msgCh {
	log.Printf("read m: %#v", m)
}

Output:

func Stub ¶

func Stub(remoteNetwork, remoteAddr string, fn func([]string) interface{}) Conn

Stub returns a (fake) Conn which pretends it is a Conn to a real redis instance, but is instead using the given callback to service requests. It is primarily useful for writing tests.

When Encode is called the given value is marshalled into bytes then unmarshalled into a []string, which is passed to the callback. The return from the callback is then marshalled and buffered interanlly, and will be unmarshalled in the next call to Decode.

remoteNetwork and remoteAddr can be empty, but if given will be used as the return from the RemoteAddr method.

If the internal buffer is empty then Decode will block until Encode is called in a separate go-routine. The SetDeadline and SetReadDeadline methods can be used as usual to limit how long Decode blocks. All other inherited net.Conn methods will panic.

m := map[string]string{}
stub := Stub("tcp", "127.0.0.1:6379", func(args []string) interface{} {
	switch args[0] {
	case "GET":
		return m[args[1]]
	case "SET":
		m[args[1]] = args[2]
		return nil
	default:
		return errors.Errorf("this stub doesn't support command %q", args[0])
	}
})

stub.Do(Cmd(nil, "SET", "foo", "1"))

var foo int
stub.Do(Cmd(&foo, "GET", "foo"))
fmt.Printf("foo: %d\n", foo)

Output:

func DialAuthPass ¶

func DialAuthPass(pass string) DialOpt

DialAuthPass will cause Dial to perform an AUTH command once the connection is created, using the given pass.

If this is set and a redis URI is passed to Dial which also has a password set, this takes precedence.

func DialConnectTimeout ¶

func DialConnectTimeout(d time.Duration) DialOpt

DialConnectTimeout determines the timeout value to pass into net.DialTimeout when creating the connection. If not set then net.Dial is called instead.

func DialReadTimeout ¶

func DialReadTimeout(d time.Duration) DialOpt

DialReadTimeout determines the deadline to set when reading from a dialed connection. If not set then SetReadDeadline is never called.

func DialSelectDB ¶

func DialSelectDB(db int) DialOpt

DialSelectDB will cause Dial to perform a SELECT command once the connection is created, using the given database index.

If this is set and a redis URI is passed to Dial which also has a database index set, this takes precedence.

func DialTimeout ¶

func DialTimeout(d time.Duration) DialOpt

DialTimeout is the equivalent to using DialConnectTimeout, DialReadTimeout, and DialWriteTimeout all with the same value.

func DialUseTLS ¶

func DialUseTLS(config *tls.Config) DialOpt

DialUseTLS will cause Dial to perform a TLS handshake using the provided config. If config is nil the config is interpreted as equivalent to the zero configuration. See https://golang.org/pkg/crypto/tls/#Config

func DialWriteTimeout ¶

func DialWriteTimeout(d time.Duration) DialOpt

DialWriteTimeout determines the deadline to set when writing to a dialed connection. If not set then SetWriteDeadline is never called.

func NewEvalScript ¶

func NewEvalScript(numKeys int, script string) EvalScript

NewEvalScript initializes a EvalScript instance. numKeys corresponds to the number of arguments which will be keys when Cmd is called

func PersistentPubSubAbortAfter ¶

func PersistentPubSubAbortAfter(attempts int) PersistentPubSubOpt

PersistentPubSubAbortAfter changes PersistentPubSub's reconnect behavior. Usually PersistentPubSub will try to reconnect forever upon a disconnect, blocking any methods which have been called until reconnect is successful.

When PersistentPubSubAbortAfter is used, it will give up after that many attempts and return the error to the method which has been blocked the longest. Another method will need to be called in order for PersistentPubSub to resume trying to reconnect.

func PersistentPubSubConnFunc ¶

func PersistentPubSubConnFunc(connFn ConnFunc) PersistentPubSubOpt

PersistentPubSubConnFunc causes PersistentPubSub to use the given ConnFunc when connecting to its destination.

func NewPool ¶

func NewPool(network, addr string, size int, opts ...PoolOpt) (*Pool, error)

NewPool creates a *Pool which will keep open at least the given number of connections to the redis instance at the given address.

NewPool takes in a number of options which can overwrite its default behavior. The default options NewPool uses are:

PoolConnFunc(DefaultConnFunc)
PoolOnEmptyCreateAfter(1 * time.Second)
PoolRefillInterval(1 * time.Second)
PoolOnFullBuffer((size / 3)+1, 1 * time.Second)
PoolPingInterval(5 * time.Second / (size+1))
PoolPipelineConcurrency(size)
PoolPipelineWindow(150 * time.Microsecond, 0)

The recommended size of the pool depends on the number of concurrent goroutines that will use the pool and whether implicit pipelining is enabled or not.

As a general rule, when implicit pipelining is enabled (the default) the size of the pool can be kept low without problems to reduce resource and file descriptor usage.

func PoolConnFunc ¶

func PoolConnFunc(cf ConnFunc) PoolOpt

PoolConnFunc tells the Pool to use the given ConnFunc when creating new Conns to its redis instance. The ConnFunc can be used to set timeouts, perform AUTH, or even use custom Conn implementations.

func PoolOnEmptyCreateAfter ¶

func PoolOnEmptyCreateAfter(wait time.Duration) PoolOpt

PoolOnEmptyCreateAfter effects the Pool's behavior when there are no available connections in the Pool. The effect is to cause actions to block until a connection becomes available or until the duration has passed. If the duration is passed a new connection is created and used.

If wait is 0 then a new connection is created immediately upon an empty Pool.

func PoolOnEmptyErrAfter ¶

func PoolOnEmptyErrAfter(wait time.Duration) PoolOpt

PoolOnEmptyErrAfter effects the Pool's behavior when there are no available connections in the Pool. The effect is to cause actions to block until a connection becomes available or until the duration has passed. If the duration is passed then ErrEmptyPool is returned.

If wait is 0 then ErrEmptyPool is returned immediately upon an empty Pool.

func PoolOnEmptyWait ¶

func PoolOnEmptyWait() PoolOpt

PoolOnEmptyWait effects the Pool's behavior when there are no available connections in the Pool. The effect is to cause actions to block as long as it takes until a connection becomes available.

func PoolOnFullBuffer ¶

func PoolOnFullBuffer(size int, drainInterval time.Duration) PoolOpt

PoolOnFullBuffer effects the Pool's behavior when it is full. The effect is to give the pool an additional buffer for connections, called the overflow. If a connection is being put back into a full pool it will be put into the overflow. If the overflow is also full then the connection will be closed and discarded.

drainInterval specifies the interval at which a drain event happens. On each drain event a connection will be removed from the overflow buffer (if any are present in it), closed, and discarded.

If drainInterval is zero then drain events will never occur.

func PoolOnFullClose ¶

func PoolOnFullClose() PoolOpt

PoolOnFullClose effects the Pool's behavior when it is full. The effect is to cause any connection which is being put back into a full pool to be closed and discarded.

func PoolPingInterval ¶

func PoolPingInterval(d time.Duration) PoolOpt

PoolPingInterval specifies the interval at which a ping event happens. On each ping event the Pool calls the PING redis command over one of it's available connections.

Since connections are used in LIFO order, the ping interval * pool size is the duration of time it takes to ping every connection once when the pool is idle.

A shorter interval means connections are pinged more frequently, but also means more traffic with the server.

func PoolPipelineConcurrency ¶

func PoolPipelineConcurrency(limit int) PoolOpt

PoolPipelineConcurrency sets the maximum number of pipelines that can be executed concurrently.

If limit is greater than the pool size or less than 1, the limit will be set to the pool size.

func PoolPipelineWindow ¶

func PoolPipelineWindow(window time.Duration, limit int) PoolOpt

PoolPipelineWindow sets the duration after which internal pipelines will be flushed and the maximum number of commands that can be pipelined before flushing.

If window is zero then implicit pipelining will be disabled. If limit is zero then no limit will be used and pipelines will only be limited by the specified time window.

func PoolRefillInterval ¶

func PoolRefillInterval(d time.Duration) PoolOpt

PoolRefillInterval specifies the interval at which a refill event happens. On each refill event the Pool checks to see if it is full, and if it's not a single connection is created and added to it.

func PoolWithTrace ¶

func PoolWithTrace(pt trace.PoolTrace) PoolOpt

PoolWithTrace tells the Pool to trace itself with the given PoolTrace Note that PoolTrace will block every point that you set to trace.

func PersistentPubSub ¶

func PersistentPubSub(network, addr string, connFn ConnFunc) PubSubConn

PersistentPubSub is deprecated in favor of PersistentPubSubWithOpts instead.

// Example of how to use PersistentPubSub with a Cluster instance.

// Initialize the cluster in any way you see fit
cluster, err := NewCluster([]string{"127.0.0.1:6379"})
if err != nil {
	panic(err)
}

// Have PersistentPubSub pick a random cluster node everytime it wants to
// make a new connection. If the node fails PersistentPubSub will
// automatically pick a new node to connect to.
ps := PersistentPubSub("", "", func(string, string) (Conn, error) {
	topo := cluster.Topo()
	node := topo[rand.Intn(len(topo))]
	return Dial("tcp", node.Addr)
})

// Use the PubSubConn as normal.
msgCh := make(chan PubSubMessage)
ps.Subscribe(msgCh, "myChannel")
for msg := range msgCh {
	log.Printf("publish to channel %q received: %q", msg.Channel, msg.Message)
}

Output:

func PersistentPubSubWithOpts ¶

func PersistentPubSubWithOpts(
	network, addr string, options ...PersistentPubSubOpt,
) (
	PubSubConn, error,
)

PersistentPubSubWithOpts is like PubSub, but instead of taking in an existing Conn to wrap it will create one on the fly. If the connection is ever terminated then a new one will be created and will be reset to the previous connection's state.

This is effectively a way to have a permanent PubSubConn established which supports subscribing/unsubscribing but without the hassle of implementing reconnect/re-subscribe logic.

With default options, none of the methods on the returned PubSubConn will ever return an error, they will instead block until a connection can be successfully reinstated.

PersistentPubSubWithOpts takes in a number of options which can overwrite its default behavior. The default options PersistentPubSubWithOpts uses are:

PersistentPubSubConnFunc(DefaultConnFunc)

func PubSub ¶

func PubSub(rc Conn) PubSubConn

PubSub wraps the given Conn so that it becomes a PubSubConn. The passed in Conn should not be used after this call.

// Create a normal redis connection
conn, err := Dial("tcp", "127.0.0.1:6379")
if err != nil {
	panic(err)
}

// Pass that connection into PubSub, conn should never get used after this
ps := PubSub(conn)

// Subscribe to a channel called "myChannel". All publishes to "myChannel"
// will get sent to msgCh after this
msgCh := make(chan PubSubMessage)
if err := ps.Subscribe(msgCh, "myChannel"); err != nil {
	panic(err)
}

for msg := range msgCh {
	log.Printf("publish to channel %q received: %q", msg.Channel, msg.Message)
}

Output:

func NewScanner ¶

func NewScanner(c Client, o ScanOpts) Scanner

NewScanner creates a new Scanner instance which will iterate over the redis instance's Client using the ScanOpts.

NOTE if Client is a *Cluster this will not work correctly, use the NewScanner method on Cluster instead.

client, err := DefaultClientFunc("tcp", "126.0.0.1:6379")
if err != nil {
	log.Fatal(err)
}

s := NewScanner(client, ScanOpts{Command: "HSCAN", Key: "somekey"})
var key string
for s.Next(&key) {
	log.Printf("key: %q", key)
}
if err := s.Close(); err != nil {
	log.Fatal(err)
}

Output:

client, err := DefaultClientFunc("tcp", "126.0.0.1:6379")
if err != nil {
	log.Fatal(err)
}

s := NewScanner(client, ScanAllKeys)
var key string
for s.Next(&key) {
	log.Printf("key: %q", key)
}
if err := s.Close(); err != nil {
	log.Fatal(err)
}

Output:

func NewSentinel ¶

func NewSentinel(primaryName string, sentinelAddrs []string, opts ...SentinelOpt) (*Sentinel, error)

NewSentinel creates and returns a *Sentinel instance. NewSentinel takes in a number of options which can overwrite its default behavior. The default options NewSentinel uses are:

SentinelConnFunc(DefaultConnFunc)
SentinelPoolFunc(DefaultClientFunc)

func SentinelConnFunc ¶

func SentinelConnFunc(cf ConnFunc) SentinelOpt

SentinelConnFunc tells the Sentinel to use the given ConnFunc when connecting to sentinel instances.

NOTE that if SentinelConnFunc is not used then Sentinel will attempt to retrieve AUTH and SELECT information from the address provided to NewSentinel, and use that for dialing all Sentinels. If SentinelConnFunc is provided, however, those options must be given through DialAuthPass/DialSelectDB within the ConnFunc.

func SentinelPoolFunc ¶

func SentinelPoolFunc(pf ClientFunc) SentinelOpt

SentinelPoolFunc tells the Sentinel to use the given ClientFunc when creating a pool of connections to the sentinel's primary.

func NewStreamReader ¶

func NewStreamReader(c Client, opts StreamReaderOpts) StreamReader

NewStreamReader returns a new StreamReader for the given client.

Any changes on opts after calling NewStreamReader will have no effect.