mirror of https://github.com/go-redis/redis.git
Merge pull request #219 from go-redis/feature/cluster-watch
cluster: add Watch support.
This commit is contained in:
commit
ba44d4d158
11
cluster.go
11
cluster.go
|
@ -44,6 +44,17 @@ func NewClusterClient(opt *ClusterOptions) *ClusterClient {
|
||||||
return client
|
return client
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Watch creates new transaction and marks the keys to be watched
|
||||||
|
// for conditional execution of a transaction.
|
||||||
|
func (c *ClusterClient) Watch(keys ...string) (*Multi, error) {
|
||||||
|
addr := c.slotMasterAddr(hashSlot(keys[0]))
|
||||||
|
client, err := c.getClient(addr)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
return client.Watch(keys...)
|
||||||
|
}
|
||||||
|
|
||||||
// Close closes the cluster client, releasing any open resources.
|
// Close closes the cluster client, releasing any open resources.
|
||||||
//
|
//
|
||||||
// It is rare to Close a ClusterClient, as the ClusterClient is meant
|
// It is rare to Close a ClusterClient, as the ClusterClient is meant
|
||||||
|
|
|
@ -4,9 +4,10 @@ package redis
|
||||||
type ClusterPipeline struct {
|
type ClusterPipeline struct {
|
||||||
commandable
|
commandable
|
||||||
|
|
||||||
cmds []Cmder
|
|
||||||
cluster *ClusterClient
|
cluster *ClusterClient
|
||||||
closed bool
|
|
||||||
|
cmds []Cmder
|
||||||
|
closed bool
|
||||||
}
|
}
|
||||||
|
|
||||||
// Pipeline creates a new pipeline which is able to execute commands
|
// Pipeline creates a new pipeline which is able to execute commands
|
||||||
|
|
|
@ -5,7 +5,9 @@ import (
|
||||||
"math/rand"
|
"math/rand"
|
||||||
"net"
|
"net"
|
||||||
"reflect"
|
"reflect"
|
||||||
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
|
"sync"
|
||||||
|
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
@ -317,6 +319,49 @@ var _ = Describe("Cluster", func() {
|
||||||
Expect(err).To(HaveOccurred())
|
Expect(err).To(HaveOccurred())
|
||||||
Expect(err.Error()).To(ContainSubstring("MOVED"))
|
Expect(err.Error()).To(ContainSubstring("MOVED"))
|
||||||
})
|
})
|
||||||
|
|
||||||
|
It("should Watch", func() {
|
||||||
|
var incr func(string) error
|
||||||
|
|
||||||
|
// Transactionally increments key using GET and SET commands.
|
||||||
|
incr = func(key string) error {
|
||||||
|
tx, err := client.Watch(key)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
defer tx.Close()
|
||||||
|
|
||||||
|
n, err := tx.Get(key).Int64()
|
||||||
|
if err != nil && err != redis.Nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
_, err = tx.Exec(func() error {
|
||||||
|
tx.Set(key, strconv.FormatInt(n+1, 10), 0)
|
||||||
|
return nil
|
||||||
|
})
|
||||||
|
if err == redis.TxFailedErr {
|
||||||
|
return incr(key)
|
||||||
|
}
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
var wg sync.WaitGroup
|
||||||
|
for i := 0; i < 100; i++ {
|
||||||
|
wg.Add(1)
|
||||||
|
go func() {
|
||||||
|
defer wg.Done()
|
||||||
|
|
||||||
|
err := incr("key")
|
||||||
|
Expect(err).NotTo(HaveOccurred())
|
||||||
|
}()
|
||||||
|
}
|
||||||
|
wg.Wait()
|
||||||
|
|
||||||
|
n, err := client.Get("key").Int64()
|
||||||
|
Expect(err).NotTo(HaveOccurred())
|
||||||
|
Expect(n).To(Equal(int64(100)))
|
||||||
|
})
|
||||||
})
|
})
|
||||||
})
|
})
|
||||||
|
|
||||||
|
|
4
multi.go
4
multi.go
|
@ -23,8 +23,8 @@ type Multi struct {
|
||||||
closed bool
|
closed bool
|
||||||
}
|
}
|
||||||
|
|
||||||
// Watch marks the keys to be watched for conditional execution
|
// Watch creates new transaction and marks the keys to be watched
|
||||||
// of a transaction.
|
// for conditional execution of a transaction.
|
||||||
func (c *Client) Watch(keys ...string) (*Multi, error) {
|
func (c *Client) Watch(keys ...string) (*Multi, error) {
|
||||||
tx := c.Multi()
|
tx := c.Multi()
|
||||||
if err := tx.Watch(keys...).Err(); err != nil {
|
if err := tx.Watch(keys...).Err(); err != nil {
|
||||||
|
|
|
@ -1,6 +1,9 @@
|
||||||
package redis_test
|
package redis_test
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"strconv"
|
||||||
|
"sync"
|
||||||
|
|
||||||
. "github.com/onsi/ginkgo"
|
. "github.com/onsi/ginkgo"
|
||||||
. "github.com/onsi/gomega"
|
. "github.com/onsi/gomega"
|
||||||
|
|
||||||
|
@ -21,29 +24,47 @@ var _ = Describe("Multi", func() {
|
||||||
Expect(client.Close()).NotTo(HaveOccurred())
|
Expect(client.Close()).NotTo(HaveOccurred())
|
||||||
})
|
})
|
||||||
|
|
||||||
It("should exec", func() {
|
It("should Watch", func() {
|
||||||
multi := client.Multi()
|
var incr func(string) error
|
||||||
defer func() {
|
|
||||||
Expect(multi.Close()).NotTo(HaveOccurred())
|
|
||||||
}()
|
|
||||||
|
|
||||||
var (
|
// Transactionally increments key using GET and SET commands.
|
||||||
set *redis.StatusCmd
|
incr = func(key string) error {
|
||||||
get *redis.StringCmd
|
tx, err := client.Watch(key)
|
||||||
)
|
if err != nil {
|
||||||
cmds, err := multi.Exec(func() error {
|
return err
|
||||||
set = multi.Set("key", "hello", 0)
|
}
|
||||||
get = multi.Get("key")
|
defer tx.Close()
|
||||||
return nil
|
|
||||||
})
|
n, err := tx.Get(key).Int64()
|
||||||
|
if err != nil && err != redis.Nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
_, err = tx.Exec(func() error {
|
||||||
|
tx.Set(key, strconv.FormatInt(n+1, 10), 0)
|
||||||
|
return nil
|
||||||
|
})
|
||||||
|
if err == redis.TxFailedErr {
|
||||||
|
return incr(key)
|
||||||
|
}
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
var wg sync.WaitGroup
|
||||||
|
for i := 0; i < 100; i++ {
|
||||||
|
wg.Add(1)
|
||||||
|
go func() {
|
||||||
|
defer wg.Done()
|
||||||
|
|
||||||
|
err := incr("key")
|
||||||
|
Expect(err).NotTo(HaveOccurred())
|
||||||
|
}()
|
||||||
|
}
|
||||||
|
wg.Wait()
|
||||||
|
|
||||||
|
n, err := client.Get("key").Int64()
|
||||||
Expect(err).NotTo(HaveOccurred())
|
Expect(err).NotTo(HaveOccurred())
|
||||||
Expect(cmds).To(HaveLen(2))
|
Expect(n).To(Equal(int64(100)))
|
||||||
|
|
||||||
Expect(set.Err()).NotTo(HaveOccurred())
|
|
||||||
Expect(set.Val()).To(Equal("OK"))
|
|
||||||
|
|
||||||
Expect(get.Err()).NotTo(HaveOccurred())
|
|
||||||
Expect(get.Val()).To(Equal("hello"))
|
|
||||||
})
|
})
|
||||||
|
|
||||||
It("should discard", func() {
|
It("should discard", func() {
|
||||||
|
|
Loading…
Reference in New Issue