mirrors/mostlymatter
1
0
Форкнуть 0
Код Задачи Запросы на слияние Действия Пакеты Проекты Релизы Вики Активность

Updating dependancies. (#9303)

Просмотр исходного кода
Этот коммит содержится в:
Christopher Speller
2018-08-28 10:05:26 -07:00
коммит произвёл GitHub
родитель 347ee1d205
Коммит 61e27beabc
255 изменённых файлов: 11076 добавлений и 5145 удалений
Скачать Patch файл Скачать Diff файл Показать все файлы Свернуть все файлы

21
vendor/github.com/go-redis/redis/CHANGELOG.md сгенерированный поставляемый Обычный файл
Просмотреть файл

@@ -0,0 +1,21 @@
# Changelog
## 6.14
- Added Options.MinIdleConns.
- Added Options.MaxConnAge.
- PoolStats.FreeConns is renamed to PoolStats.IdleConns.
- Add Client.Do to simplify creating custom commands.
- Add Cmd.String, Cmd.Int, Cmd.Int64, Cmd.Uint64, Cmd.Float64, and Cmd.Bool helpers.
- Lower memory usage.
## v6.13
- Ring got new options called `HashReplicas` and `Hash`. It is recommended to set `HashReplicas = 1000` for better keys distribution between shards.
- Cluster client was optimized to use much less memory when reloading cluster state.
- PubSub.ReceiveMessage is re-worked to not use ReceiveTimeout so it does not lose data when timeout occurres. In most cases it is recommended to use PubSub.Channel instead.
- Dialer.KeepAlive is set to 5 minutes by default.
## v6.12
- ClusterClient got new option called `ClusterSlots` which allows to build cluster of normal Redis Servers that don't have cluster mode enabled. See https://godoc.org/github.com/go-redis/redis#example-NewClusterClient--ManualSetup

33
vendor/github.com/go-redis/redis/README.md сгенерированный поставляемый
Просмотреть файл

@@ -15,6 +15,7 @@ Supports:
- [Timeouts](https://godoc.org/github.com/go-redis/redis#Options).
- [Redis Sentinel](https://godoc.org/github.com/go-redis/redis#NewFailoverClient).
- [Redis Cluster](https://godoc.org/github.com/go-redis/redis#NewClusterClient).
- [Cluster of Redis Servers](https://godoc.org/github.com/go-redis/redis#example-NewClusterClient--ManualSetup) without using cluster mode and Redis Sentinel.
- [Ring](https://godoc.org/github.com/go-redis/redis#NewRing).
- [Instrumentation](https://godoc.org/github.com/go-redis/redis#ex-package--Instrumentation).
- [Cache friendly](https://github.com/go-redis/cache).
@@ -86,25 +87,27 @@ Please go through [examples](https://godoc.org/github.com/go-redis/redis#pkg-exa
Some corner cases:
SET key value EX 10 NX
set, err := client.SetNX("key", "value", 10*time.Second).Result()
```go
// SET key value EX 10 NX
set, err := client.SetNX("key", "value", 10*time.Second).Result()
SORT list LIMIT 0 2 ASC
vals, err := client.Sort("list", redis.Sort{Offset: 0, Count: 2, Order: "ASC"}).Result()
// SORT list LIMIT 0 2 ASC
vals, err := client.Sort("list", redis.Sort{Offset: 0, Count: 2, Order: "ASC"}).Result()
ZRANGEBYSCORE zset -inf +inf WITHSCORES LIMIT 0 2
vals, err := client.ZRangeByScoreWithScores("zset", redis.ZRangeBy{
Min: "-inf",
Max: "+inf",
Offset: 0,
Count: 2,
}).Result()
// ZRANGEBYSCORE zset -inf +inf WITHSCORES LIMIT 0 2
vals, err := client.ZRangeByScoreWithScores("zset", redis.ZRangeBy{
Min: "-inf",
Max: "+inf",
Offset: 0,
Count: 2,
}).Result()
ZINTERSTORE out 2 zset1 zset2 WEIGHTS 2 3 AGGREGATE SUM
vals, err := client.ZInterStore("out", redis.ZStore{Weights: []int64{2, 3}}, "zset1", "zset2").Result()
// ZINTERSTORE out 2 zset1 zset2 WEIGHTS 2 3 AGGREGATE SUM
vals, err := client.ZInterStore("out", redis.ZStore{Weights: []int64{2, 3}}, "zset1", "zset2").Result()
EVAL "return {KEYS[1],ARGV[1]}" 1 "key" "hello"
vals, err := client.Eval("return {KEYS[1],ARGV[1]}", []string{"key"}, "hello").Result()
// EVAL "return {KEYS[1],ARGV[1]}" 1 "key" "hello"
vals, err := client.Eval("return {KEYS[1],ARGV[1]}", []string{"key"}, "hello").Result()
```
## Benchmark

325
vendor/github.com/go-redis/redis/cluster.go сгенерированный поставляемый
Просмотреть файл

@@ -8,7 +8,8 @@ import (
"math"
"math/rand"
"net"
"strings"
"runtime"
"sort"
"sync"
"sync/atomic"
"time"
@@ -30,7 +31,7 @@ type ClusterOptions struct {
// The maximum number of retries before giving up. Command is retried
// on network errors and MOVED/ASK redirects.
// Default is 8.
// Default is 8 retries.
MaxRedirects int
// Enables read-only commands on slave nodes.
@@ -39,16 +40,25 @@ type ClusterOptions struct {
// It automatically enables ReadOnly.
RouteByLatency bool
// Allows routing read-only commands to the random master or slave node.
// It automatically enables ReadOnly.
RouteRandomly bool
// Optional function that returns cluster slots information.
// It is useful to manually create cluster of standalone Redis servers
// and load-balance read/write operations between master and slaves.
// It can use service like ZooKeeper to maintain configuration information
// and Cluster.ReloadState to manually trigger state reloading.
ClusterSlots func() ([]ClusterSlot, error)
// Following options are copied from Options struct.
OnConnect func(*Conn) error
Password string
MaxRetries int
MinRetryBackoff time.Duration
MaxRetryBackoff time.Duration
Password string
DialTimeout time.Duration
ReadTimeout time.Duration
@@ -56,6 +66,8 @@ type ClusterOptions struct {
// PoolSize applies per cluster node and not for the whole cluster.
PoolSize int
MinIdleConns int
MaxConnAge time.Duration
PoolTimeout time.Duration
IdleTimeout time.Duration
IdleCheckFrequency time.Duration
@@ -70,10 +82,14 @@ func (opt *ClusterOptions) init() {
opt.MaxRedirects = 8
}
if opt.RouteByLatency {
if opt.RouteByLatency || opt.RouteRandomly {
opt.ReadOnly = true
}
if opt.PoolSize == 0 {
opt.PoolSize = 5 * runtime.NumCPU()
}
switch opt.ReadTimeout {
case -1:
opt.ReadTimeout = 0
@@ -117,10 +133,11 @@ func (opt *ClusterOptions) clientOptions() *Options {
ReadTimeout: opt.ReadTimeout,
WriteTimeout: opt.WriteTimeout,
PoolSize: opt.PoolSize,
PoolTimeout: opt.PoolTimeout,
IdleTimeout: opt.IdleTimeout,
PoolSize: opt.PoolSize,
MinIdleConns: opt.MinIdleConns,
MaxConnAge: opt.MaxConnAge,
PoolTimeout: opt.PoolTimeout,
IdleTimeout: opt.IdleTimeout,
IdleCheckFrequency: disableIdleCheck,
TLSConfig: opt.TLSConfig,
@@ -160,10 +177,6 @@ func (n *clusterNode) Close() error {
return n.Client.Close()
}
func (n *clusterNode) Test() error {
return n.Client.ClusterInfo().Err()
}
func (n *clusterNode) updateLatency() {
const probes = 10
@@ -330,7 +343,7 @@ func (c *clusterNodes) GetOrCreate(addr string) (*clusterNode, error) {
v, err := c.nodeCreateGroup.Do(addr, func() (interface{}, error) {
node := newClusterNode(c.opt, addr)
return node, node.Test()
return node, nil
})
c.mu.Lock()
@@ -383,12 +396,31 @@ func (c *clusterNodes) Random() (*clusterNode, error) {
//------------------------------------------------------------------------------
type clusterSlot struct {
start, end int
nodes []*clusterNode
}
type clusterSlotSlice []*clusterSlot
func (p clusterSlotSlice) Len() int {
return len(p)
}
func (p clusterSlotSlice) Less(i, j int) bool {
return p[i].start < p[j].start
}
func (p clusterSlotSlice) Swap(i, j int) {
p[i], p[j] = p[j], p[i]
}
type clusterState struct {
nodes *clusterNodes
Masters []*clusterNode
Slaves []*clusterNode
slots [][]*clusterNode
slots []*clusterSlot
generation uint32
createdAt time.Time
@@ -400,19 +432,21 @@ func newClusterState(
c := clusterState{
nodes: nodes,
slots: make([][]*clusterNode, hashtag.SlotNumber),
slots: make([]*clusterSlot, 0, len(slots)),
generation: nodes.NextGeneration(),
createdAt: time.Now(),
}
isLoopbackOrigin := isLoopbackAddr(origin)
originHost, _, _ := net.SplitHostPort(origin)
isLoopbackOrigin := isLoopback(originHost)
for _, slot := range slots {
var nodes []*clusterNode
for i, slotNode := range slot.Nodes {
addr := slotNode.Addr
if !isLoopbackOrigin && useOriginAddr(origin, addr) {
addr = origin
if !isLoopbackOrigin {
addr = replaceLoopbackHost(addr, originHost)
}
node, err := c.nodes.GetOrCreate(addr)
@@ -430,11 +464,15 @@ func newClusterState(
}
}
for i := slot.Start; i <= slot.End; i++ {
c.slots[i] = nodes
}
c.slots = append(c.slots, &clusterSlot{
start: slot.Start,
end: slot.End,
nodes: nodes,
})
}
sort.Sort(clusterSlotSlice(c.slots))
time.AfterFunc(time.Minute, func() {
nodes.GC(c.generation)
})
@@ -442,6 +480,33 @@ func newClusterState(
return &c, nil
}
func replaceLoopbackHost(nodeAddr, originHost string) string {
nodeHost, nodePort, err := net.SplitHostPort(nodeAddr)
if err != nil {
return nodeAddr
}
nodeIP := net.ParseIP(nodeHost)
if nodeIP == nil {
return nodeAddr
}
if !nodeIP.IsLoopback() {
return nodeAddr
}
// Use origin host which is not loopback and node port.
return net.JoinHostPort(originHost, nodePort)
}
func isLoopback(host string) bool {
ip := net.ParseIP(host)
if ip == nil {
return true
}
return ip.IsLoopback()
}
func (c *clusterState) slotMasterNode(slot int) (*clusterNode, error) {
nodes := c.slotNodes(slot)
if len(nodes) > 0 {
@@ -502,32 +567,24 @@ func (c *clusterState) slotRandomNode(slot int) *clusterNode {
}
func (c *clusterState) slotNodes(slot int) []*clusterNode {
if slot >= 0 && slot < len(c.slots) {
return c.slots[slot]
i := sort.Search(len(c.slots), func(i int) bool {
return c.slots[i].end >= slot
})
if i >= len(c.slots) {
return nil
}
x := c.slots[i]
if slot >= x.start && slot <= x.end {
return x.nodes
}
return nil
}
func (c *clusterState) IsConsistent() bool {
if len(c.Masters) > len(c.Slaves) {
return false
if c.nodes.opt.ClusterSlots != nil {
return true
}
for _, master := range c.Masters {
s := master.Client.Info("replication").Val()
if !strings.Contains(s, "role:master") {
return false
}
}
for _, slave := range c.Slaves {
s := slave.Client.Info("replication").Val()
if !strings.Contains(s, "role:slave") {
return false
}
}
return true
return len(c.Masters) <= len(c.Slaves)
}
//------------------------------------------------------------------------------
@@ -555,7 +612,7 @@ func (c *clusterStateHolder) Reload() (*clusterState, error) {
return nil, err
}
if !state.IsConsistent() {
c.LazyReload()
time.AfterFunc(time.Second, c.LazyReload)
}
return state, nil
}
@@ -614,6 +671,14 @@ func (c *clusterStateHolder) Get() (*clusterState, error) {
return nil, errors.New("redis: cluster has no state")
}
func (c *clusterStateHolder) ReloadOrGet() (*clusterState, error) {
state, err := c.Reload()
if err == nil {
return state, nil
}
return c.Get()
}
//------------------------------------------------------------------------------
// ClusterClient is a Redis Cluster client representing a pool of zero
@@ -653,6 +718,8 @@ func NewClusterClient(opt *ClusterOptions) *ClusterClient {
c.init()
_, _ = c.state.Reload()
_, _ = c.cmdsInfoCache.Get()
if opt.IdleCheckFrequency > 0 {
go c.reaper(opt.IdleCheckFrequency)
}
@@ -660,6 +727,13 @@ func NewClusterClient(opt *ClusterOptions) *ClusterClient {
return c
}
// ReloadState reloads cluster state. It calls ClusterSlots func
// to get cluster slots information.
func (c *ClusterClient) ReloadState() error {
_, err := c.state.Reload()
return err
}
func (c *ClusterClient) init() {
c.cmdable.setProcessor(c.Process)
}
@@ -818,6 +892,7 @@ func (c *ClusterClient) Watch(fn func(*Tx) error, keys ...string) error {
}
if internal.IsRetryableError(err, true) {
c.state.LazyReload()
continue
}
@@ -853,6 +928,13 @@ func (c *ClusterClient) Close() error {
return c.nodes.Close()
}
// Do creates a Cmd from the args and processes the cmd.
func (c *ClusterClient) Do(args ...interface{}) *Cmd {
cmd := NewCmd(args...)
c.Process(cmd)
return cmd
}
func (c *ClusterClient) WrapProcess(
fn func(oldProcess func(Cmder) error) func(Cmder) error,
) {
@@ -904,12 +986,14 @@ func (c *ClusterClient) defaultProcess(cmd Cmder) error {
}
if internal.IsRetryableError(err, true) {
// Firstly retry the same node.
c.state.LazyReload()
// First retry the same node.
if attempt == 0 {
continue
}
// Secondly try random node.
// Second try random node.
node, err = c.nodes.Random()
if err != nil {
break
@@ -944,12 +1028,9 @@ func (c *ClusterClient) defaultProcess(cmd Cmder) error {
// ForEachMaster concurrently calls the fn on each master node in the cluster.
// It returns the first error if any.
func (c *ClusterClient) ForEachMaster(fn func(client *Client) error) error {
state, err := c.state.Reload()
state, err := c.state.ReloadOrGet()
if err != nil {
state, err = c.state.Get()
if err != nil {
return err
}
return err
}
var wg sync.WaitGroup
@@ -980,12 +1061,9 @@ func (c *ClusterClient) ForEachMaster(fn func(client *Client) error) error {
// ForEachSlave concurrently calls the fn on each slave node in the cluster.
// It returns the first error if any.
func (c *ClusterClient) ForEachSlave(fn func(client *Client) error) error {
state, err := c.state.Reload()
state, err := c.state.ReloadOrGet()
if err != nil {
state, err = c.state.Get()
if err != nil {
return err
}
return err
}
var wg sync.WaitGroup
@@ -1016,12 +1094,9 @@ func (c *ClusterClient) ForEachSlave(fn func(client *Client) error) error {
// ForEachNode concurrently calls the fn on each known node in the cluster.
// It returns the first error if any.
func (c *ClusterClient) ForEachNode(fn func(client *Client) error) error {
state, err := c.state.Reload()
state, err := c.state.ReloadOrGet()
if err != nil {
state, err = c.state.Get()
if err != nil {
return err
}
return err
}
var wg sync.WaitGroup
@@ -1071,7 +1146,7 @@ func (c *ClusterClient) PoolStats() *PoolStats {
acc.Timeouts += s.Timeouts
acc.TotalConns += s.TotalConns
acc.FreeConns += s.FreeConns
acc.IdleConns += s.IdleConns
acc.StaleConns += s.StaleConns
}
@@ -1082,7 +1157,7 @@ func (c *ClusterClient) PoolStats() *PoolStats {
acc.Timeouts += s.Timeouts
acc.TotalConns += s.TotalConns
acc.FreeConns += s.FreeConns
acc.IdleConns += s.IdleConns
acc.StaleConns += s.StaleConns
}
@@ -1090,6 +1165,14 @@ func (c *ClusterClient) PoolStats() *PoolStats {
}
func (c *ClusterClient) loadState() (*clusterState, error) {
if c.opt.ClusterSlots != nil {
slots, err := c.opt.ClusterSlots()
if err != nil {
return nil, err
}
return newClusterState(c.nodes, slots, "")
}
addrs, err := c.nodes.Addrs()
if err != nil {
return nil, err
@@ -1196,7 +1279,7 @@ func (c *ClusterClient) defaultProcessPipeline(cmds []Cmder) error {
cmdsMap = failedCmds
}
return firstCmdsErr(cmds)
return cmdsFirstErr(cmds)
}
func (c *ClusterClient) mapCmdsByNode(cmds []Cmder) (map[*clusterNode][]Cmder, error) {
@@ -1207,9 +1290,16 @@ func (c *ClusterClient) mapCmdsByNode(cmds []Cmder) (map[*clusterNode][]Cmder, e
}
cmdsMap := make(map[*clusterNode][]Cmder)
cmdsAreReadOnly := c.cmdsAreReadOnly(cmds)
for _, cmd := range cmds {
slot := c.cmdSlot(cmd)
node, err := state.slotMasterNode(slot)
var node *clusterNode
var err error
if cmdsAreReadOnly {
_, node, err = c.cmdSlotAndNode(cmd)
} else {
slot := c.cmdSlot(cmd)
node, err = state.slotMasterNode(slot)
}
if err != nil {
return nil, err
}
@@ -1218,6 +1308,16 @@ func (c *ClusterClient) mapCmdsByNode(cmds []Cmder) (map[*clusterNode][]Cmder, e
return cmdsMap, nil
}
func (c *ClusterClient) cmdsAreReadOnly(cmds []Cmder) bool {
for _, cmd := range cmds {
cmdInfo := c.cmdInfo(cmd.Name())
if cmdInfo == nil || !cmdInfo.ReadOnly {
return false
}
}
return true
}
func (c *ClusterClient) remapCmds(cmds []Cmder, failedCmds map[*clusterNode][]Cmder) {
remappedCmds, err := c.mapCmdsByNode(cmds)
if err != nil {
@@ -1233,26 +1333,26 @@ func (c *ClusterClient) remapCmds(cmds []Cmder, failedCmds map[*clusterNode][]Cm
func (c *ClusterClient) pipelineProcessCmds(
node *clusterNode, cn *pool.Conn, cmds []Cmder, failedCmds map[*clusterNode][]Cmder,
) error {
_ = cn.SetWriteTimeout(c.opt.WriteTimeout)
err := writeCmd(cn, cmds...)
err := cn.WithWriter(c.opt.WriteTimeout, func(wr *proto.Writer) error {
return writeCmd(wr, cmds...)
})
if err != nil {
setCmdsErr(cmds, err)
failedCmds[node] = cmds
return err
}
// Set read timeout for all commands.
_ = cn.SetReadTimeout(c.opt.ReadTimeout)
return c.pipelineReadCmds(cn, cmds, failedCmds)
err = cn.WithReader(c.opt.ReadTimeout, func(rd *proto.Reader) error {
return c.pipelineReadCmds(rd, cmds, failedCmds)
})
return err
}
func (c *ClusterClient) pipelineReadCmds(
cn *pool.Conn, cmds []Cmder, failedCmds map[*clusterNode][]Cmder,
rd *proto.Reader, cmds []Cmder, failedCmds map[*clusterNode][]Cmder,
) error {
for _, cmd := range cmds {
err := cmd.readReply(cn)
err := cmd.readReply(rd)
if err == nil {
continue
}
@@ -1361,7 +1461,7 @@ func (c *ClusterClient) defaultProcessTxPipeline(cmds []Cmder) error {
}
}
return firstCmdsErr(cmds)
return cmdsFirstErr(cmds)
}
func (c *ClusterClient) mapCmdsBySlot(cmds []Cmder) map[int][]Cmder {
@@ -1376,35 +1476,37 @@ func (c *ClusterClient) mapCmdsBySlot(cmds []Cmder) map[int][]Cmder {
func (c *ClusterClient) txPipelineProcessCmds(
node *clusterNode, cn *pool.Conn, cmds []Cmder, failedCmds map[*clusterNode][]Cmder,
) error {
cn.SetWriteTimeout(c.opt.WriteTimeout)
if err := txPipelineWriteMulti(cn, cmds); err != nil {
err := cn.WithWriter(c.opt.WriteTimeout, func(wr *proto.Writer) error {
return txPipelineWriteMulti(wr, cmds)
})
if err != nil {
setCmdsErr(cmds, err)
failedCmds[node] = cmds
return err
}
// Set read timeout for all commands.
cn.SetReadTimeout(c.opt.ReadTimeout)
if err := c.txPipelineReadQueued(cn, cmds, failedCmds); err != nil {
setCmdsErr(cmds, err)
return err
}
return pipelineReadCmds(cn, cmds)
err = cn.WithReader(c.opt.ReadTimeout, func(rd *proto.Reader) error {
err := c.txPipelineReadQueued(rd, cmds, failedCmds)
if err != nil {
setCmdsErr(cmds, err)
return err
}
return pipelineReadCmds(rd, cmds)
})
return err
}
func (c *ClusterClient) txPipelineReadQueued(
cn *pool.Conn, cmds []Cmder, failedCmds map[*clusterNode][]Cmder,
rd *proto.Reader, cmds []Cmder, failedCmds map[*clusterNode][]Cmder,
) error {
// Parse queued replies.
var statusCmd StatusCmd
if err := statusCmd.readReply(cn); err != nil {
if err := statusCmd.readReply(rd); err != nil {
return err
}
for _, cmd := range cmds {
err := statusCmd.readReply(cn)
err := statusCmd.readReply(rd)
if err == nil {
continue
}
@@ -1417,7 +1519,7 @@ func (c *ClusterClient) txPipelineReadQueued(
}
// Parse number of replies.
line, err := cn.Rd.ReadLine()
line, err := rd.ReadLine()
if err != nil {
if err == Nil {
err = TxFailedErr
@@ -1445,11 +1547,9 @@ func (c *ClusterClient) txPipelineReadQueued(
}
func (c *ClusterClient) pubSub(channels []string) *PubSub {
opt := c.opt.clientOptions()
var node *clusterNode
return &PubSub{
opt: opt,
pubsub := &PubSub{
opt: c.opt.clientOptions(),
newConn: func(channels []string) (*pool.Conn, error) {
if node == nil {
@@ -1472,6 +1572,8 @@ func (c *ClusterClient) pubSub(channels []string) *PubSub {
return node.Client.connPool.CloseConn(cn)
},
}
pubsub.init()
return pubsub
}
// Subscribe subscribes the client to the specified channels.
@@ -1494,43 +1596,6 @@ func (c *ClusterClient) PSubscribe(channels ...string) *PubSub {
return pubsub
}
func useOriginAddr(originAddr, nodeAddr string) bool {
nodeHost, nodePort, err := net.SplitHostPort(nodeAddr)
if err != nil {
return false
}
nodeIP := net.ParseIP(nodeHost)
if nodeIP == nil {
return false
}
if !nodeIP.IsLoopback() {
return false
}
_, originPort, err := net.SplitHostPort(originAddr)
if err != nil {
return false
}
return nodePort == originPort
}
func isLoopbackAddr(addr string) bool {
host, _, err := net.SplitHostPort(addr)
if err != nil {
return false
}
ip := net.ParseIP(host)
if ip == nil {
return false
}
return ip.IsLoopback()
}
func appendUniqueNode(nodes []*clusterNode, node *clusterNode) []*clusterNode {
for _, n := range nodes {
if n == node {

956
vendor/github.com/go-redis/redis/command.go сгенерированный поставляемый
Просмотреть файл

Разница между файлами не показана из-за своего большого размера Загрузить разницу

296
vendor/github.com/go-redis/redis/commands.go сгенерированный поставляемый
Просмотреть файл

@@ -62,6 +62,7 @@ type Cmdable interface {
TxPipelined(fn func(Pipeliner) error) ([]Cmder, error)
TxPipeline() Pipeliner
Command() *CommandsInfoCmd
ClientGetName() *StringCmd
Echo(message interface{}) *StringCmd
Ping() *StatusCmd
@@ -171,6 +172,26 @@ type Cmdable interface {
SRem(key string, members ...interface{}) *IntCmd
SUnion(keys ...string) *StringSliceCmd
SUnionStore(destination string, keys ...string) *IntCmd
XAdd(a *XAddArgs) *StringCmd
XLen(stream string) *IntCmd
XRange(stream, start, stop string) *XMessageSliceCmd
XRangeN(stream, start, stop string, count int64) *XMessageSliceCmd
XRevRange(stream string, start, stop string) *XMessageSliceCmd
XRevRangeN(stream string, start, stop string, count int64) *XMessageSliceCmd
XRead(a *XReadArgs) *XStreamSliceCmd
XReadStreams(streams ...string) *XStreamSliceCmd
XGroupCreate(stream, group, start string) *StatusCmd
XGroupSetID(stream, group, start string) *StatusCmd
XGroupDestroy(stream, group string) *IntCmd
XGroupDelConsumer(stream, group, consumer string) *IntCmd
XReadGroup(a *XReadGroupArgs) *XStreamSliceCmd
XAck(stream, group string, ids ...string) *IntCmd
XPending(stream, group string) *XPendingCmd
XPendingExt(a *XPendingExtArgs) *XPendingExtCmd
XClaim(a *XClaimArgs) *XMessageSliceCmd
XClaimJustID(a *XClaimArgs) *StringSliceCmd
XTrim(key string, maxLen int64) *IntCmd
XTrimApprox(key string, maxLen int64) *IntCmd
ZAdd(key string, members ...Z) *IntCmd
ZAddNX(key string, members ...Z) *IntCmd
ZAddXX(key string, members ...Z) *IntCmd
@@ -209,6 +230,7 @@ type Cmdable interface {
BgRewriteAOF() *StatusCmd
BgSave() *StatusCmd
ClientKill(ipPort string) *StatusCmd
ClientKillByFilter(keys ...string) *IntCmd
ClientList() *StringCmd
ClientPause(dur time.Duration) *BoolCmd
ConfigGet(parameter string) *SliceCmd
@@ -265,9 +287,9 @@ type Cmdable interface {
GeoRadiusByMemberRO(key, member string, query *GeoRadiusQuery) *GeoLocationCmd
GeoDist(key string, member1, member2, unit string) *FloatCmd
GeoHash(key string, members ...string) *StringSliceCmd
Command() *CommandsInfoCmd
ReadOnly() *StatusCmd
ReadWrite() *StatusCmd
MemoryUsage(key string, samples ...int) *IntCmd
}
type StatefulCmdable interface {
@@ -345,6 +367,12 @@ func (c *statefulCmdable) SwapDB(index1, index2 int) *StatusCmd {
//------------------------------------------------------------------------------
func (c *cmdable) Command() *CommandsInfoCmd {
cmd := NewCommandsInfoCmd("command")
c.process(cmd)
return cmd
}
func (c *cmdable) Del(keys ...string) *IntCmd {
args := make([]interface{}, 1+len(keys))
args[0] = "del"
@@ -411,7 +439,7 @@ func (c *cmdable) Migrate(host, port, key string, db int64, timeout time.Duratio
db,
formatMs(timeout),
)
cmd.setReadTimeout(readTimeout(timeout))
cmd.setReadTimeout(timeout)
c.process(cmd)
return cmd
}
@@ -985,7 +1013,7 @@ func (c *cmdable) BLPop(timeout time.Duration, keys ...string) *StringSliceCmd {
}
args[len(args)-1] = formatSec(timeout)
cmd := NewStringSliceCmd(args...)
cmd.setReadTimeout(readTimeout(timeout))
cmd.setReadTimeout(timeout)
c.process(cmd)
return cmd
}
@@ -998,7 +1026,7 @@ func (c *cmdable) BRPop(timeout time.Duration, keys ...string) *StringSliceCmd {
}
args[len(keys)+1] = formatSec(timeout)
cmd := NewStringSliceCmd(args...)
cmd.setReadTimeout(readTimeout(timeout))
cmd.setReadTimeout(timeout)
c.process(cmd)
return cmd
}
@@ -1010,7 +1038,7 @@ func (c *cmdable) BRPopLPush(source, destination string, timeout time.Duration)
destination,
formatSec(timeout),
)
cmd.setReadTimeout(readTimeout(timeout))
cmd.setReadTimeout(timeout)
c.process(cmd)
return cmd
}
@@ -1282,6 +1310,239 @@ func (c *cmdable) SUnionStore(destination string, keys ...string) *IntCmd {
//------------------------------------------------------------------------------
type XAddArgs struct {
Stream string
MaxLen int64 // MAXLEN N
MaxLenApprox int64 // MAXLEN ~ N
ID string
Values map[string]interface{}
}
func (c *cmdable) XAdd(a *XAddArgs) *StringCmd {
args := make([]interface{}, 0, 6+len(a.Values)*2)
args = append(args, "xadd")
args = append(args, a.Stream)
if a.MaxLen > 0 {
args = append(args, "maxlen", a.MaxLen)
} else if a.MaxLenApprox > 0 {
args = append(args, "maxlen", "~", a.MaxLenApprox)
}
if a.ID != "" {
args = append(args, a.ID)
} else {
args = append(args, "*")
}
for k, v := range a.Values {
args = append(args, k)
args = append(args, v)
}
cmd := NewStringCmd(args...)
c.process(cmd)
return cmd
}
func (c *cmdable) XLen(stream string) *IntCmd {
cmd := NewIntCmd("xlen", stream)
c.process(cmd)
return cmd
}
func (c *cmdable) XRange(stream, start, stop string) *XMessageSliceCmd {
cmd := NewXMessageSliceCmd("xrange", stream, start, stop)
c.process(cmd)
return cmd
}
func (c *cmdable) XRangeN(stream, start, stop string, count int64) *XMessageSliceCmd {
cmd := NewXMessageSliceCmd("xrange", stream, start, stop, "count", count)
c.process(cmd)
return cmd
}
func (c *cmdable) XRevRange(stream, start, stop string) *XMessageSliceCmd {
cmd := NewXMessageSliceCmd("xrevrange", stream, start, stop)
c.process(cmd)
return cmd
}
func (c *cmdable) XRevRangeN(stream, start, stop string, count int64) *XMessageSliceCmd {
cmd := NewXMessageSliceCmd("xrevrange", stream, start, stop, "count", count)
c.process(cmd)
return cmd
}
type XReadArgs struct {
Streams []string
Count int64
Block time.Duration
}
func (c *cmdable) XRead(a *XReadArgs) *XStreamSliceCmd {
args := make([]interface{}, 0, 5+len(a.Streams))
args = append(args, "xread")
if a.Count > 0 {
args = append(args, "count")
args = append(args, a.Count)
}
if a.Block >= 0 {
args = append(args, "block")
args = append(args, int64(a.Block/time.Millisecond))
}
args = append(args, "streams")
for _, s := range a.Streams {
args = append(args, s)
}
cmd := NewXStreamSliceCmd(args...)
c.process(cmd)
return cmd
}
func (c *cmdable) XReadStreams(streams ...string) *XStreamSliceCmd {
return c.XRead(&XReadArgs{
Streams: streams,
Block: -1,
})
}
func (c *cmdable) XGroupCreate(stream, group, start string) *StatusCmd {
cmd := NewStatusCmd("xgroup", "create", stream, group, start)
c.process(cmd)
return cmd
}
func (c *cmdable) XGroupSetID(stream, group, start string) *StatusCmd {
cmd := NewStatusCmd("xgroup", "setid", stream, group, start)
c.process(cmd)
return cmd
}
func (c *cmdable) XGroupDestroy(stream, group string) *IntCmd {
cmd := NewIntCmd("xgroup", "destroy", stream, group)
c.process(cmd)
return cmd
}
func (c *cmdable) XGroupDelConsumer(stream, group, consumer string) *IntCmd {
cmd := NewIntCmd("xgroup", "delconsumer", stream, group, consumer)
c.process(cmd)
return cmd
}
type XReadGroupArgs struct {
Group string
Consumer string
Streams []string
Count int64
Block time.Duration
}
func (c *cmdable) XReadGroup(a *XReadGroupArgs) *XStreamSliceCmd {
args := make([]interface{}, 0, 8+len(a.Streams))
args = append(args, "xreadgroup", "group", a.Group, a.Consumer)
if a.Count > 0 {
args = append(args, "count", a.Count)
}
if a.Block >= 0 {
args = append(args, "block", int64(a.Block/time.Millisecond))
}
args = append(args, "streams")
for _, s := range a.Streams {
args = append(args, s)
}
cmd := NewXStreamSliceCmd(args...)
c.process(cmd)
return cmd
}
func (c *cmdable) XAck(stream, group string, ids ...string) *IntCmd {
args := []interface{}{"xack", stream, group}
for _, id := range ids {
args = append(args, id)
}
cmd := NewIntCmd(args...)
c.process(cmd)
return cmd
}
func (c *cmdable) XPending(stream, group string) *XPendingCmd {
cmd := NewXPendingCmd("xpending", stream, group)
c.process(cmd)
return cmd
}
type XPendingExtArgs struct {
Stream string
Group string
Start string
End string
Count int64
Consumer string
}
func (c *cmdable) XPendingExt(a *XPendingExtArgs) *XPendingExtCmd {
args := make([]interface{}, 0, 7)
args = append(args, "xpending", a.Stream, a.Group, a.Start, a.End, a.Count)
if a.Consumer != "" {
args = append(args, a.Consumer)
}
cmd := NewXPendingExtCmd(args...)
c.process(cmd)
return cmd
}
type XClaimArgs struct {
Stream string
Group string
Consumer string
MinIdle time.Duration
Messages []string
}
func (c *cmdable) XClaim(a *XClaimArgs) *XMessageSliceCmd {
args := xClaimArgs(a)
cmd := NewXMessageSliceCmd(args...)
c.process(cmd)
return cmd
}
func (c *cmdable) XClaimJustID(a *XClaimArgs) *StringSliceCmd {
args := xClaimArgs(a)
args = append(args, "justid")
cmd := NewStringSliceCmd(args...)
c.process(cmd)
return cmd
}
func xClaimArgs(a *XClaimArgs) []interface{} {
args := make([]interface{}, 0, 4+len(a.Messages))
args = append(args,
"xclaim",
a.Stream,
a.Group, a.Consumer,
int64(a.MinIdle/time.Millisecond))
for _, id := range a.Messages {
args = append(args, id)
}
return args
}
func (c *cmdable) XTrim(key string, maxLen int64) *IntCmd {
cmd := NewIntCmd("xtrim", key, "maxlen", maxLen)
c.process(cmd)
return cmd
}
func (c *cmdable) XTrimApprox(key string, maxLen int64) *IntCmd {
cmd := NewIntCmd("xtrim", key, "maxlen", "~", maxLen)
c.process(cmd)
return cmd
}
//------------------------------------------------------------------------------
// Z represents sorted set member.
type Z struct {
Score float64
@@ -1682,6 +1943,20 @@ func (c *cmdable) ClientKill(ipPort string) *StatusCmd {
return cmd
}
// ClientKillByFilter is new style synx, while the ClientKill is old
// CLIENT KILL <option> [value] ... <option> [value]
func (c *cmdable) ClientKillByFilter(keys ...string) *IntCmd {
args := make([]interface{}, 2+len(keys))
args[0] = "client"
args[1] = "kill"
for i, key := range keys {
args[2+i] = key
}
cmd := NewIntCmd(args...)
c.process(cmd)
return cmd
}
func (c *cmdable) ClientList() *StringCmd {
cmd := NewStringCmd("client", "list")
c.process(cmd)
@@ -2168,8 +2443,15 @@ func (c *cmdable) GeoPos(key string, members ...string) *GeoPosCmd {
//------------------------------------------------------------------------------
func (c *cmdable) Command() *CommandsInfoCmd {
cmd := NewCommandsInfoCmd("command")
func (c *cmdable) MemoryUsage(key string, samples ...int) *IntCmd {
args := []interface{}{"memory", "usage", key}
if len(samples) > 0 {
if len(samples) != 1 {
panic("MemoryUsage expects single sample count")
}
args = append(args, "SAMPLES", samples[0])
}
cmd := NewIntCmd(args...)
c.process(cmd)
return cmd
}

29
vendor/github.com/go-redis/redis/internal/error.go сгенерированный поставляемый
Просмотреть файл

@@ -8,9 +8,15 @@ import (
"github.com/go-redis/redis/internal/proto"
)
func IsRetryableError(err error, retryNetError bool) bool {
if IsNetworkError(err) {
return retryNetError
func IsRetryableError(err error, retryTimeout bool) bool {
if err == io.EOF {
return true
}
if netErr, ok := err.(net.Error); ok {
if netErr.Timeout() {
return retryTimeout
}
return true
}
s := err.Error()
if s == "ERR max number of clients reached" {
@@ -19,6 +25,9 @@ func IsRetryableError(err error, retryNetError bool) bool {
if strings.HasPrefix(s, "LOADING ") {
return true
}
if strings.HasPrefix(s, "READONLY ") {
return true
}
if strings.HasPrefix(s, "CLUSTERDOWN ") {
return true
}
@@ -30,24 +39,12 @@ func IsRedisError(err error) bool {
return ok
}
func IsNetworkError(err error) bool {
if err == io.EOF {
return true
}
_, ok := err.(net.Error)
return ok
}
func IsReadOnlyError(err error) bool {
return strings.HasPrefix(err.Error(), "READONLY ")
}
func IsBadConn(err error, allowTimeout bool) bool {
if err == nil {
return false
}
if IsRedisError(err) {
return false
return strings.HasPrefix(err.Error(), "READONLY ")
}
if allowTimeout {
if netErr, ok := err.(net.Error); ok && netErr.Timeout() {

6
vendor/github.com/go-redis/redis/internal/hashtag/hashtag.go сгенерированный поставляемый
Просмотреть файл

@@ -5,7 +5,7 @@ import (
"strings"
)
const SlotNumber = 16384
const slotNumber = 16384
// CRC16 implementation according to CCITT standards.
// Copyright 2001-2010 Georges Menie (www.menie.org)
@@ -56,7 +56,7 @@ func Key(key string) string {
}
func RandomSlot() int {
return rand.Intn(SlotNumber)
return rand.Intn(slotNumber)
}
// hashSlot returns a consistent slot number between 0 and 16383
@@ -66,7 +66,7 @@ func Slot(key string) int {
return RandomSlot()
}
key = Key(key)
return int(crc16sum(key)) % SlotNumber
return int(crc16sum(key)) % slotNumber
}
func crc16sum(key string) (crc uint16) {

41
vendor/github.com/go-redis/redis/internal/pool/conn.go сгенерированный поставляемый
Просмотреть файл

@@ -13,19 +13,21 @@ var noDeadline = time.Time{}
type Conn struct {
netConn net.Conn
Rd *proto.Reader
Wb *proto.WriteBuffer
rd *proto.Reader
rdLocked bool
wr *proto.Writer
Inited bool
usedAt atomic.Value
InitedAt time.Time
pooled bool
usedAt atomic.Value
}
func NewConn(netConn net.Conn) *Conn {
cn := &Conn{
netConn: netConn,
Wb: proto.NewWriteBuffer(),
}
cn.Rd = proto.NewReader(cn.netConn)
cn.rd = proto.NewReader(netConn)
cn.wr = proto.NewWriter(netConn)
cn.SetUsedAt(time.Now())
return cn
}
@@ -40,14 +42,11 @@ func (cn *Conn) SetUsedAt(tm time.Time) {
func (cn *Conn) SetNetConn(netConn net.Conn) {
cn.netConn = netConn
cn.Rd.Reset(netConn)
cn.rd.Reset(netConn)
cn.wr.Reset(netConn)
}
func (cn *Conn) IsStale(timeout time.Duration) bool {
return timeout > 0 && time.Since(cn.UsedAt()) > timeout
}
func (cn *Conn) SetReadTimeout(timeout time.Duration) error {
func (cn *Conn) setReadTimeout(timeout time.Duration) error {
now := time.Now()
cn.SetUsedAt(now)
if timeout > 0 {
@@ -56,7 +55,7 @@ func (cn *Conn) SetReadTimeout(timeout time.Duration) error {
return cn.netConn.SetReadDeadline(noDeadline)
}
func (cn *Conn) SetWriteTimeout(timeout time.Duration) error {
func (cn *Conn) setWriteTimeout(timeout time.Duration) error {
now := time.Now()
cn.SetUsedAt(now)
if timeout > 0 {
@@ -73,6 +72,22 @@ func (cn *Conn) RemoteAddr() net.Addr {
return cn.netConn.RemoteAddr()
}
func (cn *Conn) WithReader(timeout time.Duration, fn func(rd *proto.Reader) error) error {
_ = cn.setReadTimeout(timeout)
return fn(cn.rd)
}
func (cn *Conn) WithWriter(timeout time.Duration, fn func(wr *proto.Writer) error) error {
_ = cn.setWriteTimeout(timeout)
firstErr := fn(cn.wr)
err := cn.wr.Flush()
if err != nil && firstErr == nil {
firstErr = err
}
return firstErr
}
func (cn *Conn) Close() error {
return cn.netConn.Close()
}

132
vendor/github.com/go-redis/redis/internal/pool/pool.go сгенерированный поставляемый
Просмотреть файл

@@ -28,7 +28,6 @@ type Stats struct {
Timeouts uint32 // number of times a wait timeout occurred
TotalConns uint32 // number of total connections in the pool
FreeConns uint32 // deprecated - use IdleConns
IdleConns uint32 // number of idle connections in the pool
StaleConns uint32 // number of stale connections removed from the pool
}
@@ -53,6 +52,8 @@ type Options struct {
OnClose func(*Conn) error
PoolSize int
MinIdleConns int
MaxConnAge time.Duration
PoolTimeout time.Duration
IdleTimeout time.Duration
IdleCheckFrequency time.Duration
@@ -63,16 +64,16 @@ type ConnPool struct {
dialErrorsNum uint32 // atomic
lastDialError error
lastDialErrorMu sync.RWMutex
lastDialError error
queue chan struct{}
connsMu sync.Mutex
conns []*Conn
idleConnsMu sync.RWMutex
idleConns []*Conn
connsMu sync.Mutex
conns []*Conn
idleConns []*Conn
poolSize int
idleConnsLen int
stats Stats
@@ -90,6 +91,10 @@ func NewConnPool(opt *Options) *ConnPool {
idleConns: make([]*Conn, 0, opt.PoolSize),
}
for i := 0; i < opt.MinIdleConns; i++ {
p.checkMinIdleConns()
}
if opt.IdleTimeout > 0 && opt.IdleCheckFrequency > 0 {
go p.reaper(opt.IdleCheckFrequency)
}
@@ -97,19 +102,53 @@ func NewConnPool(opt *Options) *ConnPool {
return p
}
func (p *ConnPool) checkMinIdleConns() {
if p.opt.MinIdleConns == 0 {
return
}
if p.poolSize < p.opt.PoolSize && p.idleConnsLen < p.opt.MinIdleConns {
p.poolSize++
p.idleConnsLen++
go p.addIdleConn()
}
}
func (p *ConnPool) addIdleConn() {
cn, err := p.newConn(true)
if err != nil {
return
}
p.connsMu.Lock()
p.conns = append(p.conns, cn)
p.idleConns = append(p.idleConns, cn)
p.connsMu.Unlock()
}
func (p *ConnPool) NewConn() (*Conn, error) {
cn, err := p.newConn()
return p._NewConn(false)
}
func (p *ConnPool) _NewConn(pooled bool) (*Conn, error) {
cn, err := p.newConn(pooled)
if err != nil {
return nil, err
}
p.connsMu.Lock()
p.conns = append(p.conns, cn)
if pooled {
if p.poolSize < p.opt.PoolSize {
p.poolSize++
} else {
cn.pooled = false
}
}
p.connsMu.Unlock()
return cn, nil
}
func (p *ConnPool) newConn() (*Conn, error) {
func (p *ConnPool) newConn(pooled bool) (*Conn, error) {
if p.closed() {
return nil, ErrClosed
}
@@ -127,7 +166,9 @@ func (p *ConnPool) newConn() (*Conn, error) {
return nil, err
}
return NewConn(netConn), nil
cn := NewConn(netConn)
cn.pooled = pooled
return cn, nil
}
func (p *ConnPool) tryDial() {
@@ -174,16 +215,16 @@ func (p *ConnPool) Get() (*Conn, error) {
}
for {
p.idleConnsMu.Lock()
p.connsMu.Lock()
cn := p.popIdle()
p.idleConnsMu.Unlock()
p.connsMu.Unlock()
if cn == nil {
break
}
if cn.IsStale(p.opt.IdleTimeout) {
p.CloseConn(cn)
if p.isStaleConn(cn) {
_ = p.CloseConn(cn)
continue
}
@@ -193,7 +234,7 @@ func (p *ConnPool) Get() (*Conn, error) {
atomic.AddUint32(&p.stats.Misses, 1)
newcn, err := p.NewConn()
newcn, err := p._NewConn(true)
if err != nil {
p.freeTurn()
return nil, err
@@ -241,21 +282,21 @@ func (p *ConnPool) popIdle() *Conn {
idx := len(p.idleConns) - 1
cn := p.idleConns[idx]
p.idleConns = p.idleConns[:idx]
p.idleConnsLen--
p.checkMinIdleConns()
return cn
}
func (p *ConnPool) Put(cn *Conn) {
buf := cn.Rd.PeekBuffered()
if buf != nil {
internal.Logf("connection has unread data: %.100q", buf)
if !cn.pooled {
p.Remove(cn)
return
}
p.idleConnsMu.Lock()
p.connsMu.Lock()
p.idleConns = append(p.idleConns, cn)
p.idleConnsMu.Unlock()
p.idleConnsLen++
p.connsMu.Unlock()
p.freeTurn()
}
@@ -275,6 +316,10 @@ func (p *ConnPool) removeConn(cn *Conn) {
for i, c := range p.conns {
if c == cn {
p.conns = append(p.conns[:i], p.conns[i+1:]...)
if cn.pooled {
p.poolSize--
p.checkMinIdleConns()
}
break
}
}
@@ -291,17 +336,17 @@ func (p *ConnPool) closeConn(cn *Conn) error {
// Len returns total number of connections.
func (p *ConnPool) Len() int {
p.connsMu.Lock()
l := len(p.conns)
n := len(p.conns)
p.connsMu.Unlock()
return l
return n
}
// FreeLen returns number of idle connections.
// IdleLen returns number of idle connections.
func (p *ConnPool) IdleLen() int {
p.idleConnsMu.RLock()
l := len(p.idleConns)
p.idleConnsMu.RUnlock()
return l
p.connsMu.Lock()
n := p.idleConnsLen
p.connsMu.Unlock()
return n
}
func (p *ConnPool) Stats() *Stats {
@@ -312,7 +357,6 @@ func (p *ConnPool) Stats() *Stats {
Timeouts: atomic.LoadUint32(&p.stats.Timeouts),
TotalConns: uint32(p.Len()),
FreeConns: uint32(idleLen),
IdleConns: uint32(idleLen),
StaleConns: atomic.LoadUint32(&p.stats.StaleConns),
}
@@ -349,11 +393,10 @@ func (p *ConnPool) Close() error {
}
}
p.conns = nil
p.connsMu.Unlock()
p.idleConnsMu.Lock()
p.poolSize = 0
p.idleConns = nil
p.idleConnsMu.Unlock()
p.idleConnsLen = 0
p.connsMu.Unlock()
return firstErr
}
@@ -364,11 +407,12 @@ func (p *ConnPool) reapStaleConn() *Conn {
}
cn := p.idleConns[0]
if !cn.IsStale(p.opt.IdleTimeout) {
if !p.isStaleConn(cn) {
return nil
}
p.idleConns = append(p.idleConns[:0], p.idleConns[1:]...)
p.idleConnsLen--
return cn
}
@@ -378,9 +422,9 @@ func (p *ConnPool) ReapStaleConns() (int, error) {
for {
p.getTurn()
p.idleConnsMu.Lock()
p.connsMu.Lock()
cn := p.reapStaleConn()
p.idleConnsMu.Unlock()
p.connsMu.Unlock()
if cn != nil {
p.removeConn(cn)
@@ -414,3 +458,19 @@ func (p *ConnPool) reaper(frequency time.Duration) {
atomic.AddUint32(&p.stats.StaleConns, uint32(n))
}
}
func (p *ConnPool) isStaleConn(cn *Conn) bool {
if p.opt.IdleTimeout == 0 && p.opt.MaxConnAge == 0 {
return false
}
now := time.Now()
if p.opt.IdleTimeout > 0 && now.Sub(cn.UsedAt()) >= p.opt.IdleTimeout {
return true
}
if p.opt.MaxConnAge > 0 && now.Sub(cn.InitedAt) >= p.opt.MaxConnAge {
return true
}
return false
}

214
vendor/github.com/go-redis/redis/internal/proto/reader.go сгенерированный поставляемый
Просмотреть файл

@@ -9,8 +9,6 @@ import (
"github.com/go-redis/redis/internal/util"
)
const bytesAllocLimit = 1024 * 1024 // 1mb
const (
ErrorReply = '-'
StatusReply = '+'
@@ -32,40 +30,23 @@ func (e RedisError) Error() string { return string(e) }
type MultiBulkParse func(*Reader, int64) (interface{}, error)
type Reader struct {
src *bufio.Reader
buf []byte
rd *bufio.Reader
_buf []byte
}
func NewReader(rd io.Reader) *Reader {
return &Reader{
src: bufio.NewReader(rd),
buf: make([]byte, 4096),
rd: bufio.NewReader(rd),
_buf: make([]byte, 64),
}
}
func (r *Reader) Reset(rd io.Reader) {
r.src.Reset(rd)
}
func (r *Reader) PeekBuffered() []byte {
if n := r.src.Buffered(); n != 0 {
b, _ := r.src.Peek(n)
return b
}
return nil
}
func (r *Reader) ReadN(n int) ([]byte, error) {
b, err := readN(r.src, r.buf, n)
if err != nil {
return nil, err
}
r.buf = b
return b, nil
r.rd.Reset(rd)
}
func (r *Reader) ReadLine() ([]byte, error) {
line, isPrefix, err := r.src.ReadLine()
line, isPrefix, err := r.rd.ReadLine()
if err != nil {
return nil, err
}
@@ -91,11 +72,11 @@ func (r *Reader) ReadReply(m MultiBulkParse) (interface{}, error) {
case ErrorReply:
return nil, ParseErrorReply(line)
case StatusReply:
return parseStatusValue(line), nil
return string(line[1:]), nil
case IntReply:
return util.ParseInt(line[1:], 10, 64)
case StringReply:
return r.readTmpBytesValue(line)
return r.readStringReply(line)
case ArrayReply:
n, err := parseArrayLen(line)
if err != nil {
@@ -121,47 +102,42 @@ func (r *Reader) ReadIntReply() (int64, error) {
}
}
func (r *Reader) ReadTmpBytesReply() ([]byte, error) {
func (r *Reader) ReadString() (string, error) {
line, err := r.ReadLine()
if err != nil {
return nil, err
}
switch line[0] {
case ErrorReply:
return nil, ParseErrorReply(line)
case StringReply:
return r.readTmpBytesValue(line)
case StatusReply:
return parseStatusValue(line), nil
default:
return nil, fmt.Errorf("redis: can't parse string reply: %.100q", line)
}
}
func (r *Reader) ReadBytesReply() ([]byte, error) {
b, err := r.ReadTmpBytesReply()
if err != nil {
return nil, err
}
cp := make([]byte, len(b))
copy(cp, b)
return cp, nil
}
func (r *Reader) ReadStringReply() (string, error) {
b, err := r.ReadTmpBytesReply()
if err != nil {
return "", err
}
return string(b), nil
switch line[0] {
case ErrorReply:
return "", ParseErrorReply(line)
case StringReply:
return r.readStringReply(line)
case StatusReply:
return string(line[1:]), nil
case IntReply:
return string(line[1:]), nil
default:
return "", fmt.Errorf("redis: can't parse reply=%.100q reading string", line)
}
}
func (r *Reader) ReadFloatReply() (float64, error) {
b, err := r.ReadTmpBytesReply()
if err != nil {
return 0, err
func (r *Reader) readStringReply(line []byte) (string, error) {
if isNilReply(line) {
return "", Nil
}
return util.ParseFloat(b, 64)
replyLen, err := strconv.Atoi(string(line[1:]))
if err != nil {
return "", err
}
b := make([]byte, replyLen+2)
_, err = io.ReadFull(r.rd, b)
if err != nil {
return "", err
}
return util.BytesToString(b[:replyLen]), nil
}
func (r *Reader) ReadArrayReply(m MultiBulkParse) (interface{}, error) {
@@ -219,7 +195,7 @@ func (r *Reader) ReadScanReply() ([]string, uint64, error) {
keys := make([]string, n)
for i := int64(0); i < n; i++ {
key, err := r.ReadStringReply()
key, err := r.ReadString()
if err != nil {
return nil, 0, err
}
@@ -229,7 +205,48 @@ func (r *Reader) ReadScanReply() ([]string, uint64, error) {
return keys, cursor, err
}
func (r *Reader) readTmpBytesValue(line []byte) ([]byte, error) {
func (r *Reader) ReadInt() (int64, error) {
b, err := r.readTmpBytesReply()
if err != nil {
return 0, err
}
return util.ParseInt(b, 10, 64)
}
func (r *Reader) ReadUint() (uint64, error) {
b, err := r.readTmpBytesReply()
if err != nil {
return 0, err
}
return util.ParseUint(b, 10, 64)
}
func (r *Reader) ReadFloatReply() (float64, error) {
b, err := r.readTmpBytesReply()
if err != nil {
return 0, err
}
return util.ParseFloat(b, 64)
}
func (r *Reader) readTmpBytesReply() ([]byte, error) {
line, err := r.ReadLine()
if err != nil {
return nil, err
}
switch line[0] {
case ErrorReply:
return nil, ParseErrorReply(line)
case StringReply:
return r._readTmpBytesReply(line)
case StatusReply:
return line[1:], nil
default:
return nil, fmt.Errorf("redis: can't parse string reply: %.100q", line)
}
}
func (r *Reader) _readTmpBytesReply(line []byte) ([]byte, error) {
if isNilReply(line) {
return nil, Nil
}
@@ -239,71 +256,20 @@ func (r *Reader) readTmpBytesValue(line []byte) ([]byte, error) {
return nil, err
}
b, err := r.ReadN(replyLen + 2)
buf := r.buf(replyLen + 2)
_, err = io.ReadFull(r.rd, buf)
if err != nil {
return nil, err
}
return b[:replyLen], nil
return buf[:replyLen], nil
}
func (r *Reader) ReadInt() (int64, error) {
b, err := r.ReadTmpBytesReply()
if err != nil {
return 0, err
func (r *Reader) buf(n int) []byte {
if d := n - cap(r._buf); d > 0 {
r._buf = append(r._buf, make([]byte, d)...)
}
return util.ParseInt(b, 10, 64)
}
func (r *Reader) ReadUint() (uint64, error) {
b, err := r.ReadTmpBytesReply()
if err != nil {
return 0, err
}
return util.ParseUint(b, 10, 64)
}
// --------------------------------------------------------------------
func readN(r io.Reader, b []byte, n int) ([]byte, error) {
if n == 0 && b == nil {
return make([]byte, 0), nil
}
if cap(b) >= n {
b = b[:n]
_, err := io.ReadFull(r, b)
return b, err
}
b = b[:cap(b)]
pos := 0
for pos < n {
diff := n - len(b)
if diff > bytesAllocLimit {
diff = bytesAllocLimit
}
b = append(b, make([]byte, diff)...)
nn, err := io.ReadFull(r, b[pos:])
if err != nil {
return nil, err
}
pos += nn
}
return b, nil
}
func formatInt(n int64) string {
return strconv.FormatInt(n, 10)
}
func formatUint(u uint64) string {
return strconv.FormatUint(u, 10)
}
func formatFloat(f float64) string {
return strconv.FormatFloat(f, 'f', -1, 64)
return r._buf[:n]
}
func isNilReply(b []byte) bool {
@@ -316,10 +282,6 @@ func ParseErrorReply(line []byte) error {
return RedisError(string(line[1:]))
}
func parseStatusValue(line []byte) []byte {
return line[1:]
}
func parseArrayLen(line []byte) (int64, error) {
if isNilReply(line) {
return 0, Nil

101
vendor/github.com/go-redis/redis/internal/proto/write_buffer.go сгенерированный поставляемый
Просмотреть файл

@@ -1,101 +0,0 @@
package proto
import (
"encoding"
"fmt"
"strconv"
)
type WriteBuffer struct {
b []byte
}
func NewWriteBuffer() *WriteBuffer {
return &WriteBuffer{
b: make([]byte, 0, 4096),
}
}
func (w *WriteBuffer) Len() int { return len(w.b) }
func (w *WriteBuffer) Bytes() []byte { return w.b }
func (w *WriteBuffer) Reset() { w.b = w.b[:0] }
func (w *WriteBuffer) Append(args []interface{}) error {
w.b = append(w.b, ArrayReply)
w.b = strconv.AppendUint(w.b, uint64(len(args)), 10)
w.b = append(w.b, '\r', '\n')
for _, arg := range args {
if err := w.append(arg); err != nil {
return err
}
}
return nil
}
func (w *WriteBuffer) append(val interface{}) error {
switch v := val.(type) {
case nil:
w.AppendString("")
case string:
w.AppendString(v)
case []byte:
w.AppendBytes(v)
case int:
w.AppendString(formatInt(int64(v)))
case int8:
w.AppendString(formatInt(int64(v)))
case int16:
w.AppendString(formatInt(int64(v)))
case int32:
w.AppendString(formatInt(int64(v)))
case int64:
w.AppendString(formatInt(v))
case uint:
w.AppendString(formatUint(uint64(v)))
case uint8:
w.AppendString(formatUint(uint64(v)))
case uint16:
w.AppendString(formatUint(uint64(v)))
case uint32:
w.AppendString(formatUint(uint64(v)))
case uint64:
w.AppendString(formatUint(v))
case float32:
w.AppendString(formatFloat(float64(v)))
case float64:
w.AppendString(formatFloat(v))
case bool:
if v {
w.AppendString("1")
} else {
w.AppendString("0")
}
case encoding.BinaryMarshaler:
b, err := v.MarshalBinary()
if err != nil {
return err
}
w.AppendBytes(b)
default:
return fmt.Errorf(
"redis: can't marshal %T (consider implementing encoding.BinaryMarshaler)", val)
}
return nil
}
func (w *WriteBuffer) AppendString(s string) {
w.b = append(w.b, StringReply)
w.b = strconv.AppendUint(w.b, uint64(len(s)), 10)
w.b = append(w.b, '\r', '\n')
w.b = append(w.b, s...)
w.b = append(w.b, '\r', '\n')
}
func (w *WriteBuffer) AppendBytes(p []byte) {
w.b = append(w.b, StringReply)
w.b = strconv.AppendUint(w.b, uint64(len(p)), 10)
w.b = append(w.b, '\r', '\n')
w.b = append(w.b, p...)
w.b = append(w.b, '\r', '\n')
}

159
vendor/github.com/go-redis/redis/internal/proto/writer.go сгенерированный поставляемый Обычный файл
Просмотреть файл

@@ -0,0 +1,159 @@
package proto
import (
"bufio"
"encoding"
"fmt"
"io"
"strconv"
"github.com/go-redis/redis/internal/util"
)
type Writer struct {
wr *bufio.Writer
lenBuf []byte
numBuf []byte
}
func NewWriter(wr io.Writer) *Writer {
return &Writer{
wr: bufio.NewWriter(wr),
lenBuf: make([]byte, 64),
numBuf: make([]byte, 64),
}
}
func (w *Writer) WriteArgs(args []interface{}) error {
err := w.wr.WriteByte(ArrayReply)
if err != nil {
return err
}
err = w.writeLen(len(args))
if err != nil {
return err
}
for _, arg := range args {
err := w.writeArg(arg)
if err != nil {
return err
}
}
return nil
}
func (w *Writer) writeLen(n int) error {
w.lenBuf = strconv.AppendUint(w.lenBuf[:0], uint64(n), 10)
w.lenBuf = append(w.lenBuf, '\r', '\n')
_, err := w.wr.Write(w.lenBuf)
return err
}
func (w *Writer) writeArg(v interface{}) error {
switch v := v.(type) {
case nil:
return w.string("")
case string:
return w.string(v)
case []byte:
return w.bytes(v)
case int:
return w.int(int64(v))
case int8:
return w.int(int64(v))
case int16:
return w.int(int64(v))
case int32:
return w.int(int64(v))
case int64:
return w.int(v)
case uint:
return w.uint(uint64(v))
case uint8:
return w.uint(uint64(v))
case uint16:
return w.uint(uint64(v))
case uint32:
return w.uint(uint64(v))
case uint64:
return w.uint(v)
case float32:
return w.float(float64(v))
case float64:
return w.float(v)
case bool:
if v {
return w.int(1)
} else {
return w.int(0)
}
case encoding.BinaryMarshaler:
b, err := v.MarshalBinary()
if err != nil {
return err
}
return w.bytes(b)
default:
return fmt.Errorf(
"redis: can't marshal %T (implement encoding.BinaryMarshaler)", v)
}
}
func (w *Writer) bytes(b []byte) error {
err := w.wr.WriteByte(StringReply)
if err != nil {
return err
}
err = w.writeLen(len(b))
if err != nil {
return err
}
_, err = w.wr.Write(b)
if err != nil {
return err
}
return w.crlf()
}
func (w *Writer) string(s string) error {
return w.bytes(util.StringToBytes(s))
}
func (w *Writer) uint(n uint64) error {
w.numBuf = strconv.AppendUint(w.numBuf[:0], n, 10)
return w.bytes(w.numBuf)
}
func (w *Writer) int(n int64) error {
w.numBuf = strconv.AppendInt(w.numBuf[:0], n, 10)
return w.bytes(w.numBuf)
}
func (w *Writer) float(f float64) error {
w.numBuf = strconv.AppendFloat(w.numBuf[:0], f, 'f', -1, 64)
return w.bytes(w.numBuf)
}
func (w *Writer) crlf() error {
err := w.wr.WriteByte('\r')
if err != nil {
return err
}
return w.wr.WriteByte('\n')
}
func (w *Writer) Reset(wr io.Writer) {
w.wr.Reset(wr)
}
func (w *Writer) Flush() error {
return w.wr.Flush()
}

4
vendor/github.com/go-redis/redis/internal/util/safe.go сгенерированный поставляемый
Просмотреть файл

@@ -5,3 +5,7 @@ package util
func BytesToString(b []byte) string {
return string(b)
}
func StringToBytes(s string) []byte {
return []byte(s)
}

10
vendor/github.com/go-redis/redis/internal/util/unsafe.go сгенерированный поставляемый
Просмотреть файл

@@ -10,3 +10,13 @@ import (
func BytesToString(b []byte) string {
return *(*string)(unsafe.Pointer(&b))
}
// StringToBytes converts string to byte slice.
func StringToBytes(s string) []byte {
return *(*[]byte)(unsafe.Pointer(
&struct {
string
Cap int
}{s, len(s)},
))
}

29
vendor/github.com/go-redis/redis/options.go сгенерированный поставляемый
Просмотреть файл

@@ -59,16 +59,24 @@ type Options struct {
// Maximum number of socket connections.
// Default is 10 connections per every CPU as reported by runtime.NumCPU.
PoolSize int
// Minimum number of idle connections which is useful when establishing
// new connection is slow.
MinIdleConns int
// Connection age at which client retires (closes) the connection.
// Default is to not close aged connections.
MaxConnAge time.Duration
// Amount of time client waits for connection if all connections
// are busy before returning an error.
// Default is ReadTimeout + 1 second.
PoolTimeout time.Duration
// Amount of time after which client closes idle connections.
// Should be less than server's timeout.
// Default is 5 minutes.
// Default is 5 minutes. -1 disables idle timeout check.
IdleTimeout time.Duration
// Frequency of idle checks.
// Default is 1 minute. -1 disables idle check.
// Frequency of idle checks made by idle connections reaper.
// Default is 1 minute. -1 disables idle connections reaper,
// but idle connections are still discarded by the client
// if IdleTimeout is set.
IdleCheckFrequency time.Duration
// Enables read only queries on slave nodes.
@@ -84,12 +92,15 @@ func (opt *Options) init() {
}
if opt.Dialer == nil {
opt.Dialer = func() (net.Conn, error) {
conn, err := net.DialTimeout(opt.Network, opt.Addr, opt.DialTimeout)
if opt.TLSConfig == nil || err != nil {
return conn, err
netDialer := &net.Dialer{
Timeout: opt.DialTimeout,
KeepAlive: 5 * time.Minute,
}
if opt.TLSConfig == nil {
return netDialer.Dial(opt.Network, opt.Addr)
} else {
return tls.DialWithDialer(netDialer, opt.Network, opt.Addr, opt.TLSConfig)
}
t := tls.Client(conn, opt.TLSConfig)
return t, t.Handshake()
}
}
if opt.PoolSize == 0 {
@@ -192,6 +203,8 @@ func newConnPool(opt *Options) *pool.ConnPool {
return pool.NewConnPool(&pool.Options{
Dialer: opt.Dialer,
PoolSize: opt.PoolSize,
MinIdleConns: opt.MinIdleConns,
MaxConnAge: opt.MaxConnAge,
PoolTimeout: opt.PoolTimeout,
IdleTimeout: opt.IdleTimeout,
IdleCheckFrequency: opt.IdleCheckFrequency,

394
vendor/github.com/go-redis/redis/parser.go сгенерированный поставляемый
Просмотреть файл

@@ -1,394 +0,0 @@
package redis
import (
"fmt"
"net"
"strconv"
"time"
"github.com/go-redis/redis/internal/proto"
)
// Implements proto.MultiBulkParse
func sliceParser(rd *proto.Reader, n int64) (interface{}, error) {
vals := make([]interface{}, 0, n)
for i := int64(0); i < n; i++ {
v, err := rd.ReadReply(sliceParser)
if err != nil {
if err == Nil {
vals = append(vals, nil)
continue
}
if err, ok := err.(proto.RedisError); ok {
vals = append(vals, err)
continue
}
return nil, err
}
switch v := v.(type) {
case []byte:
vals = append(vals, string(v))
default:
vals = append(vals, v)
}
}
return vals, nil
}
// Implements proto.MultiBulkParse
func boolSliceParser(rd *proto.Reader, n int64) (interface{}, error) {
bools := make([]bool, 0, n)
for i := int64(0); i < n; i++ {
n, err := rd.ReadIntReply()
if err != nil {
return nil, err
}
bools = append(bools, n == 1)
}
return bools, nil
}
// Implements proto.MultiBulkParse
func stringSliceParser(rd *proto.Reader, n int64) (interface{}, error) {
ss := make([]string, 0, n)
for i := int64(0); i < n; i++ {
s, err := rd.ReadStringReply()
if err == Nil {
ss = append(ss, "")
} else if err != nil {
return nil, err
} else {
ss = append(ss, s)
}
}
return ss, nil
}
// Implements proto.MultiBulkParse
func stringStringMapParser(rd *proto.Reader, n int64) (interface{}, error) {
m := make(map[string]string, n/2)
for i := int64(0); i < n; i += 2 {
key, err := rd.ReadStringReply()
if err != nil {
return nil, err
}
value, err := rd.ReadStringReply()
if err != nil {
return nil, err
}
m[key] = value
}
return m, nil
}
// Implements proto.MultiBulkParse
func stringIntMapParser(rd *proto.Reader, n int64) (interface{}, error) {
m := make(map[string]int64, n/2)
for i := int64(0); i < n; i += 2 {
key, err := rd.ReadStringReply()
if err != nil {
return nil, err
}
n, err := rd.ReadIntReply()
if err != nil {
return nil, err
}
m[key] = n
}
return m, nil
}
// Implements proto.MultiBulkParse
func stringStructMapParser(rd *proto.Reader, n int64) (interface{}, error) {
m := make(map[string]struct{}, n)
for i := int64(0); i < n; i++ {
key, err := rd.ReadStringReply()
if err != nil {
return nil, err
}
m[key] = struct{}{}
}
return m, nil
}
// Implements proto.MultiBulkParse
func zSliceParser(rd *proto.Reader, n int64) (interface{}, error) {
zz := make([]Z, n/2)
for i := int64(0); i < n; i += 2 {
var err error
z := &zz[i/2]
z.Member, err = rd.ReadStringReply()
if err != nil {
return nil, err
}
z.Score, err = rd.ReadFloatReply()
if err != nil {
return nil, err
}
}
return zz, nil
}
// Implements proto.MultiBulkParse
func clusterSlotsParser(rd *proto.Reader, n int64) (interface{}, error) {
slots := make([]ClusterSlot, n)
for i := 0; i < len(slots); i++ {
n, err := rd.ReadArrayLen()
if err != nil {
return nil, err
}
if n < 2 {
err := fmt.Errorf("redis: got %d elements in cluster info, expected at least 2", n)
return nil, err
}
start, err := rd.ReadIntReply()
if err != nil {
return nil, err
}
end, err := rd.ReadIntReply()
if err != nil {
return nil, err
}
nodes := make([]ClusterNode, n-2)
for j := 0; j < len(nodes); j++ {
n, err := rd.ReadArrayLen()
if err != nil {
return nil, err
}
if n != 2 && n != 3 {
err := fmt.Errorf("got %d elements in cluster info address, expected 2 or 3", n)
return nil, err
}
ip, err := rd.ReadStringReply()
if err != nil {
return nil, err
}
port, err := rd.ReadIntReply()
if err != nil {
return nil, err
}
nodes[j].Addr = net.JoinHostPort(ip, strconv.FormatInt(port, 10))
if n == 3 {
id, err := rd.ReadStringReply()
if err != nil {
return nil, err
}
nodes[j].Id = id
}
}
slots[i] = ClusterSlot{
Start: int(start),
End: int(end),
Nodes: nodes,
}
}
return slots, nil
}
func newGeoLocationParser(q *GeoRadiusQuery) proto.MultiBulkParse {
return func(rd *proto.Reader, n int64) (interface{}, error) {
var loc GeoLocation
var err error
loc.Name, err = rd.ReadStringReply()
if err != nil {
return nil, err
}
if q.WithDist {
loc.Dist, err = rd.ReadFloatReply()
if err != nil {
return nil, err
}
}
if q.WithGeoHash {
loc.GeoHash, err = rd.ReadIntReply()
if err != nil {
return nil, err
}
}
if q.WithCoord {
n, err := rd.ReadArrayLen()
if err != nil {
return nil, err
}
if n != 2 {
return nil, fmt.Errorf("got %d coordinates, expected 2", n)
}
loc.Longitude, err = rd.ReadFloatReply()
if err != nil {
return nil, err
}
loc.Latitude, err = rd.ReadFloatReply()
if err != nil {
return nil, err
}
}
return &loc, nil
}
}
func newGeoLocationSliceParser(q *GeoRadiusQuery) proto.MultiBulkParse {
return func(rd *proto.Reader, n int64) (interface{}, error) {
locs := make([]GeoLocation, 0, n)
for i := int64(0); i < n; i++ {
v, err := rd.ReadReply(newGeoLocationParser(q))
if err != nil {
return nil, err
}
switch vv := v.(type) {
case []byte:
locs = append(locs, GeoLocation{
Name: string(vv),
})
case *GeoLocation:
locs = append(locs, *vv)
default:
return nil, fmt.Errorf("got %T, expected string or *GeoLocation", v)
}
}
return locs, nil
}
}
func geoPosParser(rd *proto.Reader, n int64) (interface{}, error) {
var pos GeoPos
var err error
pos.Longitude, err = rd.ReadFloatReply()
if err != nil {
return nil, err
}
pos.Latitude, err = rd.ReadFloatReply()
if err != nil {
return nil, err
}
return &pos, nil
}
func geoPosSliceParser(rd *proto.Reader, n int64) (interface{}, error) {
positions := make([]*GeoPos, 0, n)
for i := int64(0); i < n; i++ {
v, err := rd.ReadReply(geoPosParser)
if err != nil {
if err == Nil {
positions = append(positions, nil)
continue
}
return nil, err
}
switch v := v.(type) {
case *GeoPos:
positions = append(positions, v)
default:
return nil, fmt.Errorf("got %T, expected *GeoPos", v)
}
}
return positions, nil
}
func commandInfoParser(rd *proto.Reader, n int64) (interface{}, error) {
var cmd CommandInfo
var err error
if n != 6 {
return nil, fmt.Errorf("redis: got %d elements in COMMAND reply, wanted 6", n)
}
cmd.Name, err = rd.ReadStringReply()
if err != nil {
return nil, err
}
arity, err := rd.ReadIntReply()
if err != nil {
return nil, err
}
cmd.Arity = int8(arity)
flags, err := rd.ReadReply(stringSliceParser)
if err != nil {
return nil, err
}
cmd.Flags = flags.([]string)
firstKeyPos, err := rd.ReadIntReply()
if err != nil {
return nil, err
}
cmd.FirstKeyPos = int8(firstKeyPos)
lastKeyPos, err := rd.ReadIntReply()
if err != nil {
return nil, err
}
cmd.LastKeyPos = int8(lastKeyPos)
stepCount, err := rd.ReadIntReply()
if err != nil {
return nil, err
}
cmd.StepCount = int8(stepCount)
for _, flag := range cmd.Flags {
if flag == "readonly" {
cmd.ReadOnly = true
break
}
}
return &cmd, nil
}
// Implements proto.MultiBulkParse
func commandInfoSliceParser(rd *proto.Reader, n int64) (interface{}, error) {
m := make(map[string]*CommandInfo, n)
for i := int64(0); i < n; i++ {
v, err := rd.ReadReply(commandInfoParser)
if err != nil {
return nil, err
}
vv := v.(*CommandInfo)
m[vv.Name] = vv
}
return m, nil
}
// Implements proto.MultiBulkParse
func timeParser(rd *proto.Reader, n int64) (interface{}, error) {
if n != 2 {
return nil, fmt.Errorf("got %d elements, expected 2", n)
}
sec, err := rd.ReadInt()
if err != nil {
return nil, err
}
microsec, err := rd.ReadInt()
if err != nil {
return nil, err
}
return time.Unix(sec, microsec*1000), nil
}

1
vendor/github.com/go-redis/redis/pipeline.go сгенерированный поставляемый
Просмотреть файл

@@ -31,6 +31,7 @@ type Pipeline struct {
closed bool
}
// Process queues the cmd for later execution.
func (c *Pipeline) Process(cmd Cmder) error {
c.mu.Lock()
c.cmds = append(c.cmds, cmd)

296
vendor/github.com/go-redis/redis/pubsub.go сгенерированный поставляемый
Просмотреть файл

@@ -2,20 +2,20 @@ package redis
import (
"fmt"
"net"
"sync"
"time"
"github.com/go-redis/redis/internal"
"github.com/go-redis/redis/internal/pool"
"github.com/go-redis/redis/internal/proto"
)
// PubSub implements Pub/Sub commands as described in
// http://redis.io/topics/pubsub. It's NOT safe for concurrent use by
// multiple goroutines.
// PubSub implements Pub/Sub commands bas described in
// http://redis.io/topics/pubsub. Message receiving is NOT safe
// for concurrent use by multiple goroutines.
//
// PubSub automatically resubscribes to the channels and patterns
// when Redis becomes unavailable.
// PubSub automatically reconnects to Redis Server and resubscribes
// to the channels in case of network errors.
type PubSub struct {
opt *Options
@@ -27,11 +27,17 @@ type PubSub struct {
channels map[string]struct{}
patterns map[string]struct{}
closed bool
exit chan struct{}
cmd *Cmd
chOnce sync.Once
ch chan *Message
ping chan struct{}
}
func (c *PubSub) init() {
c.exit = make(chan struct{})
}
func (c *PubSub) conn() (*pool.Conn, error) {
@@ -41,7 +47,7 @@ func (c *PubSub) conn() (*pool.Conn, error) {
return cn, err
}
func (c *PubSub) _conn(channels []string) (*pool.Conn, error) {
func (c *PubSub) _conn(newChannels []string) (*pool.Conn, error) {
if c.closed {
return nil, pool.ErrClosed
}
@@ -50,6 +56,9 @@ func (c *PubSub) _conn(channels []string) (*pool.Conn, error) {
return c.cn, nil
}
channels := mapKeys(c.channels)
channels = append(channels, newChannels...)
cn, err := c.newConn(channels)
if err != nil {
return nil, err
@@ -64,61 +73,81 @@ func (c *PubSub) _conn(channels []string) (*pool.Conn, error) {
return cn, nil
}
func (c *PubSub) writeCmd(cn *pool.Conn, cmd Cmder) error {
return cn.WithWriter(c.opt.WriteTimeout, func(wr *proto.Writer) error {
return writeCmd(wr, cmd)
})
}
func (c *PubSub) resubscribe(cn *pool.Conn) error {
var firstErr error
if len(c.channels) > 0 {
channels := make([]string, len(c.channels))
i := 0
for channel := range c.channels {
channels[i] = channel
i++
}
if err := c._subscribe(cn, "subscribe", channels...); err != nil && firstErr == nil {
err := c._subscribe(cn, "subscribe", mapKeys(c.channels))
if err != nil && firstErr == nil {
firstErr = err
}
}
if len(c.patterns) > 0 {
patterns := make([]string, len(c.patterns))
i := 0
for pattern := range c.patterns {
patterns[i] = pattern
i++
}
if err := c._subscribe(cn, "psubscribe", patterns...); err != nil && firstErr == nil {
err := c._subscribe(cn, "psubscribe", mapKeys(c.patterns))
if err != nil && firstErr == nil {
firstErr = err
}
}
return firstErr
}
func (c *PubSub) _subscribe(cn *pool.Conn, redisCmd string, channels ...string) error {
args := make([]interface{}, 1+len(channels))
args[0] = redisCmd
for i, channel := range channels {
args[1+i] = channel
func mapKeys(m map[string]struct{}) []string {
s := make([]string, len(m))
i := 0
for k := range m {
s[i] = k
i++
}
cmd := NewSliceCmd(args...)
cn.SetWriteTimeout(c.opt.WriteTimeout)
return writeCmd(cn, cmd)
return s
}
func (c *PubSub) releaseConn(cn *pool.Conn, err error) {
func (c *PubSub) _subscribe(
cn *pool.Conn, redisCmd string, channels []string,
) error {
args := make([]interface{}, 0, 1+len(channels))
args = append(args, redisCmd)
for _, channel := range channels {
args = append(args, channel)
}
cmd := NewSliceCmd(args...)
return c.writeCmd(cn, cmd)
}
func (c *PubSub) releaseConn(cn *pool.Conn, err error, allowTimeout bool) {
c.mu.Lock()
c._releaseConn(cn, err)
c._releaseConn(cn, err, allowTimeout)
c.mu.Unlock()
}
func (c *PubSub) _releaseConn(cn *pool.Conn, err error) {
func (c *PubSub) _releaseConn(cn *pool.Conn, err error, allowTimeout bool) {
if c.cn != cn {
return
}
if internal.IsBadConn(err, true) {
_ = c.closeTheCn()
if internal.IsBadConn(err, allowTimeout) {
c._reconnect(err)
}
}
func (c *PubSub) closeTheCn() error {
func (c *PubSub) _reconnect(reason error) {
_ = c._closeTheCn(reason)
_, _ = c._conn(nil)
}
func (c *PubSub) _closeTheCn(reason error) error {
if c.cn == nil {
return nil
}
if !c.closed {
internal.Logf("redis: discarding bad PubSub connection: %s", reason)
}
err := c.closeConn(c.cn)
c.cn = nil
return err
@@ -132,25 +161,25 @@ func (c *PubSub) Close() error {
return pool.ErrClosed
}
c.closed = true
close(c.exit)
if c.cn != nil {
return c.closeTheCn()
}
return nil
err := c._closeTheCn(pool.ErrClosed)
return err
}
// Subscribe the client to the specified channels. It returns
// empty subscription if there are no channels.
func (c *PubSub) Subscribe(channels ...string) error {
c.mu.Lock()
defer c.mu.Unlock()
err := c.subscribe("subscribe", channels...)
if c.channels == nil {
c.channels = make(map[string]struct{})
}
for _, channel := range channels {
c.channels[channel] = struct{}{}
for _, s := range channels {
c.channels[s] = struct{}{}
}
c.mu.Unlock()
return err
}
@@ -158,14 +187,15 @@ func (c *PubSub) Subscribe(channels ...string) error {
// empty subscription if there are no patterns.
func (c *PubSub) PSubscribe(patterns ...string) error {
c.mu.Lock()
defer c.mu.Unlock()
err := c.subscribe("psubscribe", patterns...)
if c.patterns == nil {
c.patterns = make(map[string]struct{})
}
for _, pattern := range patterns {
c.patterns[pattern] = struct{}{}
for _, s := range patterns {
c.patterns[s] = struct{}{}
}
c.mu.Unlock()
return err
}
@@ -173,11 +203,12 @@ func (c *PubSub) PSubscribe(patterns ...string) error {
// them if none is given.
func (c *PubSub) Unsubscribe(channels ...string) error {
c.mu.Lock()
err := c.subscribe("unsubscribe", channels...)
defer c.mu.Unlock()
for _, channel := range channels {
delete(c.channels, channel)
}
c.mu.Unlock()
err := c.subscribe("unsubscribe", channels...)
return err
}
@@ -185,11 +216,12 @@ func (c *PubSub) Unsubscribe(channels ...string) error {
// them if none is given.
func (c *PubSub) PUnsubscribe(patterns ...string) error {
c.mu.Lock()
err := c.subscribe("punsubscribe", patterns...)
defer c.mu.Unlock()
for _, pattern := range patterns {
delete(c.patterns, pattern)
}
c.mu.Unlock()
err := c.subscribe("punsubscribe", patterns...)
return err
}
@@ -199,8 +231,8 @@ func (c *PubSub) subscribe(redisCmd string, channels ...string) error {
return err
}
err = c._subscribe(cn, redisCmd, channels...)
c._releaseConn(cn, err)
err = c._subscribe(cn, redisCmd, channels)
c._releaseConn(cn, err, false)
return err
}
@@ -216,9 +248,8 @@ func (c *PubSub) Ping(payload ...string) error {
return err
}
cn.SetWriteTimeout(c.opt.WriteTimeout)
err = writeCmd(cn, cmd)
c.releaseConn(cn, err)
err = c.writeCmd(cn, cmd)
c.releaseConn(cn, err, false)
return err
}
@@ -297,8 +328,8 @@ func (c *PubSub) newMessage(reply interface{}) (interface{}, error) {
}
// ReceiveTimeout acts like Receive but returns an error if message
// is not received in time. This is low-level API and most clients
// should use ReceiveMessage.
// is not received in time. This is low-level API and in most cases
// Channel should be used instead.
func (c *PubSub) ReceiveTimeout(timeout time.Duration) (interface{}, error) {
if c.cmd == nil {
c.cmd = NewCmd()
@@ -309,9 +340,11 @@ func (c *PubSub) ReceiveTimeout(timeout time.Duration) (interface{}, error) {
return nil, err
}
cn.SetReadTimeout(timeout)
err = c.cmd.readReply(cn)
c.releaseConn(cn, err)
err = cn.WithReader(timeout, func(rd *proto.Reader) error {
return c.cmd.readReply(rd)
})
c.releaseConn(cn, err, timeout > 0)
if err != nil {
return nil, err
}
@@ -320,49 +353,23 @@ func (c *PubSub) ReceiveTimeout(timeout time.Duration) (interface{}, error) {
}
// Receive returns a message as a Subscription, Message, Pong or error.
// See PubSub example for details. This is low-level API and most clients
// should use ReceiveMessage.
// See PubSub example for details. This is low-level API and in most cases
// Channel should be used instead.
func (c *PubSub) Receive() (interface{}, error) {
return c.ReceiveTimeout(0)
}
// ReceiveMessage returns a Message or error ignoring Subscription or Pong
// messages. It automatically reconnects to Redis Server and resubscribes
// to channels in case of network errors.
// ReceiveMessage returns a Message or error ignoring Subscription and Pong
// messages. This is low-level API and in most cases Channel should be used
// instead.
func (c *PubSub) ReceiveMessage() (*Message, error) {
return c.receiveMessage(5 * time.Second)
}
func (c *PubSub) receiveMessage(timeout time.Duration) (*Message, error) {
var errNum uint
for {
msgi, err := c.ReceiveTimeout(timeout)
msg, err := c.Receive()
if err != nil {
if !internal.IsNetworkError(err) {
return nil, err
}
errNum++
if errNum < 3 {
if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
err := c.Ping()
if err != nil {
internal.Logf("PubSub.Ping failed: %s", err)
}
}
} else {
// 3 consequent errors - connection is broken or
// Redis Server is down.
// Sleep to not exceed max number of open connections.
time.Sleep(time.Second)
}
continue
return nil, err
}
// Reset error number, because we received a message.
errNum = 0
switch msg := msgi.(type) {
switch msg := msg.(type) {
case *Subscription:
// Ignore.
case *Pong:
@@ -370,30 +377,93 @@ func (c *PubSub) receiveMessage(timeout time.Duration) (*Message, error) {
case *Message:
return msg, nil
default:
return nil, fmt.Errorf("redis: unknown message: %T", msgi)
err := fmt.Errorf("redis: unknown message: %T", msg)
return nil, err
}
}
}
// Channel returns a Go channel for concurrently receiving messages.
// The channel is closed with PubSub. Receive or ReceiveMessage APIs
// can not be used after channel is created.
// It periodically sends Ping messages to test connection health.
// The channel is closed with PubSub. Receive* APIs can not be used
// after channel is created.
func (c *PubSub) Channel() <-chan *Message {
c.chOnce.Do(func() {
c.ch = make(chan *Message, 100)
go func() {
for {
msg, err := c.ReceiveMessage()
if err != nil {
if err == pool.ErrClosed {
break
}
continue
}
c.ch <- msg
}
close(c.ch)
}()
})
c.chOnce.Do(c.initChannel)
return c.ch
}
func (c *PubSub) initChannel() {
c.ch = make(chan *Message, 100)
c.ping = make(chan struct{}, 10)
go func() {
var errCount int
for {
msg, err := c.Receive()
if err != nil {
if err == pool.ErrClosed {
close(c.ch)
return
}
if errCount > 0 {
time.Sleep(c.retryBackoff(errCount))
}
errCount++
continue
}
errCount = 0
// Any message is as good as a ping.
select {
case c.ping <- struct{}{}:
default:
}
switch msg := msg.(type) {
case *Subscription:
// Ignore.
case *Pong:
// Ignore.
case *Message:
c.ch <- msg
default:
internal.Logf("redis: unknown message: %T", msg)
}
}
}()
go func() {
const timeout = 5 * time.Second
timer := time.NewTimer(timeout)
timer.Stop()
healthy := true
var pingErr error
for {
timer.Reset(timeout)
select {
case <-c.ping:
healthy = true
if !timer.Stop() {
<-timer.C
}
case <-timer.C:
pingErr = c.Ping()
if healthy {
healthy = false
} else {
c.mu.Lock()
c._reconnect(pingErr)
c.mu.Unlock()
}
case <-c.exit:
return
}
}
}()
}
func (c *PubSub) retryBackoff(attempt int) time.Duration {
return internal.RetryBackoff(attempt, c.opt.MinRetryBackoff, c.opt.MaxRetryBackoff)
}

91
vendor/github.com/go-redis/redis/redis.go сгенерированный поставляемый
Просмотреть файл

@@ -50,7 +50,7 @@ func (c *baseClient) newConn() (*pool.Conn, error) {
return nil, err
}
if !cn.Inited {
if cn.InitedAt.IsZero() {
if err := c.initConn(cn); err != nil {
_ = c.connPool.CloseConn(cn)
return nil, err
@@ -66,7 +66,7 @@ func (c *baseClient) getConn() (*pool.Conn, error) {
return nil, err
}
if !cn.Inited {
if cn.InitedAt.IsZero() {
err := c.initConn(cn)
if err != nil {
c.connPool.Remove(cn)
@@ -88,7 +88,7 @@ func (c *baseClient) releaseConn(cn *pool.Conn, err error) bool {
}
func (c *baseClient) initConn(cn *pool.Conn) error {
cn.Inited = true
cn.InitedAt = time.Now()
if c.opt.Password == "" &&
c.opt.DB == 0 &&
@@ -123,8 +123,17 @@ func (c *baseClient) initConn(cn *pool.Conn) error {
return nil
}
// Do creates a Cmd from the args and processes the cmd.
func (c *baseClient) Do(args ...interface{}) *Cmd {
cmd := NewCmd(args...)
c.Process(cmd)
return cmd
}
// WrapProcess wraps function that processes Redis commands.
func (c *baseClient) WrapProcess(fn func(oldProcess func(cmd Cmder) error) func(cmd Cmder) error) {
func (c *baseClient) WrapProcess(
fn func(oldProcess func(cmd Cmder) error) func(cmd Cmder) error,
) {
c.process = fn(c.process)
}
@@ -147,8 +156,10 @@ func (c *baseClient) defaultProcess(cmd Cmder) error {
return err
}
cn.SetWriteTimeout(c.opt.WriteTimeout)
if err := writeCmd(cn, cmd); err != nil {
err = cn.WithWriter(c.opt.WriteTimeout, func(wr *proto.Writer) error {
return writeCmd(wr, cmd)
})
if err != nil {
c.releaseConn(cn, err)
cmd.setErr(err)
if internal.IsRetryableError(err, true) {
@@ -157,8 +168,9 @@ func (c *baseClient) defaultProcess(cmd Cmder) error {
return err
}
cn.SetReadTimeout(c.cmdTimeout(cmd))
err = cmd.readReply(cn)
err = cn.WithReader(c.cmdTimeout(cmd), func(rd *proto.Reader) error {
return cmd.readReply(rd)
})
c.releaseConn(cn, err)
if err != nil && internal.IsRetryableError(err, cmd.readTimeout() == nil) {
continue
@@ -176,9 +188,8 @@ func (c *baseClient) retryBackoff(attempt int) time.Duration {
func (c *baseClient) cmdTimeout(cmd Cmder) time.Duration {
if timeout := cmd.readTimeout(); timeout != nil {
return *timeout
return readTimeout(*timeout)
}
return c.opt.ReadTimeout
}
@@ -244,24 +255,27 @@ func (c *baseClient) generalProcessPipeline(cmds []Cmder, p pipelineProcessor) e
break
}
}
return firstCmdsErr(cmds)
return cmdsFirstErr(cmds)
}
func (c *baseClient) pipelineProcessCmds(cn *pool.Conn, cmds []Cmder) (bool, error) {
cn.SetWriteTimeout(c.opt.WriteTimeout)
if err := writeCmd(cn, cmds...); err != nil {
err := cn.WithWriter(c.opt.WriteTimeout, func(wr *proto.Writer) error {
return writeCmd(wr, cmds...)
})
if err != nil {
setCmdsErr(cmds, err)
return true, err
}
// Set read timeout for all commands.
cn.SetReadTimeout(c.opt.ReadTimeout)
return true, pipelineReadCmds(cn, cmds)
err = cn.WithReader(c.opt.ReadTimeout, func(rd *proto.Reader) error {
return pipelineReadCmds(rd, cmds)
})
return true, err
}
func pipelineReadCmds(cn *pool.Conn, cmds []Cmder) error {
func pipelineReadCmds(rd *proto.Reader, cmds []Cmder) error {
for _, cmd := range cmds {
err := cmd.readReply(cn)
err := cmd.readReply(rd)
if err != nil && !internal.IsRedisError(err) {
return err
}
@@ -270,47 +284,50 @@ func pipelineReadCmds(cn *pool.Conn, cmds []Cmder) error {
}
func (c *baseClient) txPipelineProcessCmds(cn *pool.Conn, cmds []Cmder) (bool, error) {
cn.SetWriteTimeout(c.opt.WriteTimeout)
if err := txPipelineWriteMulti(cn, cmds); err != nil {
err := cn.WithWriter(c.opt.WriteTimeout, func(wr *proto.Writer) error {
return txPipelineWriteMulti(wr, cmds)
})
if err != nil {
setCmdsErr(cmds, err)
return true, err
}
// Set read timeout for all commands.
cn.SetReadTimeout(c.opt.ReadTimeout)
if err := c.txPipelineReadQueued(cn, cmds); err != nil {
setCmdsErr(cmds, err)
return false, err
}
return false, pipelineReadCmds(cn, cmds)
err = cn.WithReader(c.opt.ReadTimeout, func(rd *proto.Reader) error {
err := txPipelineReadQueued(rd, cmds)
if err != nil {
setCmdsErr(cmds, err)
return err
}
return pipelineReadCmds(rd, cmds)
})
return false, err
}
func txPipelineWriteMulti(cn *pool.Conn, cmds []Cmder) error {
func txPipelineWriteMulti(wr *proto.Writer, cmds []Cmder) error {
multiExec := make([]Cmder, 0, len(cmds)+2)
multiExec = append(multiExec, NewStatusCmd("MULTI"))
multiExec = append(multiExec, cmds...)
multiExec = append(multiExec, NewSliceCmd("EXEC"))
return writeCmd(cn, multiExec...)
return writeCmd(wr, multiExec...)
}
func (c *baseClient) txPipelineReadQueued(cn *pool.Conn, cmds []Cmder) error {
func txPipelineReadQueued(rd *proto.Reader, cmds []Cmder) error {
// Parse queued replies.
var statusCmd StatusCmd
if err := statusCmd.readReply(cn); err != nil {
err := statusCmd.readReply(rd)
if err != nil {
return err
}
for _ = range cmds {
err := statusCmd.readReply(cn)
err = statusCmd.readReply(rd)
if err != nil && !internal.IsRedisError(err) {
return err
}
}
// Parse number of replies.
line, err := cn.Rd.ReadLine()
line, err := rd.ReadLine()
if err != nil {
if err == Nil {
err = TxFailedErr
@@ -424,7 +441,7 @@ func (c *Client) TxPipeline() Pipeliner {
}
func (c *Client) pubSub() *PubSub {
return &PubSub{
pubsub := &PubSub{
opt: c.opt,
newConn: func(channels []string) (*pool.Conn, error) {
@@ -432,6 +449,8 @@ func (c *Client) pubSub() *PubSub {
},
closeConn: c.connPool.CloseConn,
}
pubsub.init()
return pubsub
}
// Subscribe subscribes the client to the specified channels.

2
vendor/github.com/go-redis/redis/result.go сгенерированный поставляемый
Просмотреть файл

@@ -53,7 +53,7 @@ func NewBoolResult(val bool, err error) *BoolCmd {
// NewStringResult returns a StringCmd initialised with val and err for testing
func NewStringResult(val string, err error) *StringCmd {
var cmd StringCmd
cmd.val = []byte(val)
cmd.val = val
cmd.setErr(err)
return &cmd
}

79
vendor/github.com/go-redis/redis/ring.go сгенерированный поставляемый
Просмотреть файл

@@ -16,7 +16,8 @@ import (
"github.com/go-redis/redis/internal/pool"
)
const nreplicas = 100
// Hash is type of hash function used in consistent hash.
type Hash consistenthash.Hash
var errRingShardsDown = errors.New("redis: all ring shards are down")
@@ -30,6 +31,27 @@ type RingOptions struct {
// Shard is considered down after 3 subsequent failed checks.
HeartbeatFrequency time.Duration
// Hash function used in consistent hash.
// Default is crc32.ChecksumIEEE.
Hash Hash
// Number of replicas in consistent hash.
// Default is 100 replicas.
//
// Higher number of replicas will provide less deviation, that is keys will be
// distributed to nodes more evenly.
//
// Following is deviation for common nreplicas:
// --------------------------------------------------------
// | nreplicas | standard error | 99% confidence interval |
// | 10 | 0.3152 | (0.37, 1.98) |
// | 100 | 0.0997 | (0.76, 1.28) |
// | 1000 | 0.0316 | (0.92, 1.09) |
// --------------------------------------------------------
//
// See https://arxiv.org/abs/1406.2294 for reference
HashReplicas int
// Following options are copied from Options struct.
OnConnect func(*Conn) error
@@ -46,6 +68,8 @@ type RingOptions struct {
WriteTimeout time.Duration
PoolSize int
MinIdleConns int
MaxConnAge time.Duration
PoolTimeout time.Duration
IdleTimeout time.Duration
IdleCheckFrequency time.Duration
@@ -56,6 +80,10 @@ func (opt *RingOptions) init() {
opt.HeartbeatFrequency = 500 * time.Millisecond
}
if opt.HashReplicas == 0 {
opt.HashReplicas = 100
}
switch opt.MinRetryBackoff {
case -1:
opt.MinRetryBackoff = 0
@@ -82,6 +110,8 @@ func (opt *RingOptions) clientOptions() *Options {
WriteTimeout: opt.WriteTimeout,
PoolSize: opt.PoolSize,
MinIdleConns: opt.MinIdleConns,
MaxConnAge: opt.MaxConnAge,
PoolTimeout: opt.PoolTimeout,
IdleTimeout: opt.IdleTimeout,
IdleCheckFrequency: opt.IdleCheckFrequency,
@@ -133,16 +163,21 @@ func (shard *ringShard) Vote(up bool) bool {
//------------------------------------------------------------------------------
type ringShards struct {
opt *RingOptions
mu sync.RWMutex
hash *consistenthash.Map
shards map[string]*ringShard // read only
list []*ringShard // read only
len int
closed bool
}
func newRingShards() *ringShards {
func newRingShards(opt *RingOptions) *ringShards {
return &ringShards{
hash: consistenthash.New(nreplicas, nil),
opt: opt,
hash: newConsistentHash(opt),
shards: make(map[string]*ringShard),
}
}
@@ -238,18 +273,28 @@ func (c *ringShards) Heartbeat(frequency time.Duration) {
// rebalance removes dead shards from the Ring.
func (c *ringShards) rebalance() {
hash := consistenthash.New(nreplicas, nil)
hash := newConsistentHash(c.opt)
var shardsNum int
for name, shard := range c.shards {
if shard.IsUp() {
hash.Add(name)
shardsNum++
}
}
c.mu.Lock()
c.hash = hash
c.len = shardsNum
c.mu.Unlock()
}
func (c *ringShards) Len() int {
c.mu.RLock()
l := c.len
c.mu.RUnlock()
return l
}
func (c *ringShards) Close() error {
c.mu.Lock()
defer c.mu.Unlock()
@@ -305,7 +350,7 @@ func NewRing(opt *RingOptions) *Ring {
ring := &Ring{
opt: opt,
shards: newRingShards(),
shards: newRingShards(opt),
}
ring.cmdsInfoCache = newCmdsInfoCache(ring.cmdsInfo)
@@ -363,11 +408,16 @@ func (c *Ring) PoolStats() *PoolStats {
acc.Misses += s.Misses
acc.Timeouts += s.Timeouts
acc.TotalConns += s.TotalConns
acc.FreeConns += s.FreeConns
acc.IdleConns += s.IdleConns
}
return &acc
}
// Len returns the current number of shards in the ring.
func (c *Ring) Len() int {
return c.shards.Len()
}
// Subscribe subscribes the client to the specified channels.
func (c *Ring) Subscribe(channels ...string) *PubSub {
if len(channels) == 0 {
@@ -466,7 +516,16 @@ func (c *Ring) cmdShard(cmd Cmder) (*ringShard, error) {
return c.shards.GetByKey(firstKey)
}
func (c *Ring) WrapProcess(fn func(oldProcess func(cmd Cmder) error) func(cmd Cmder) error) {
// Do creates a Cmd from the args and processes the cmd.
func (c *Ring) Do(args ...interface{}) *Cmd {
cmd := NewCmd(args...)
c.Process(cmd)
return cmd
}
func (c *Ring) WrapProcess(
fn func(oldProcess func(cmd Cmder) error) func(cmd Cmder) error,
) {
c.ForEachShard(func(c *Client) error {
c.WrapProcess(fn)
return nil
@@ -552,7 +611,7 @@ func (c *Ring) defaultProcessPipeline(cmds []Cmder) error {
cmdsMap = failedCmdsMap
}
return firstCmdsErr(cmds)
return cmdsFirstErr(cmds)
}
func (c *Ring) TxPipeline() Pipeliner {
@@ -570,3 +629,7 @@ func (c *Ring) TxPipelined(fn func(Pipeliner) error) ([]Cmder, error) {
func (c *Ring) Close() error {
return c.shards.Close()
}
func newConsistentHash(opt *RingOptions) *consistenthash.Map {
return consistenthash.New(opt.HashReplicas, consistenthash.Hash(opt.Hash))
}

84
vendor/github.com/go-redis/redis/sentinel.go сгенерированный поставляемый
Просмотреть файл

@@ -29,13 +29,17 @@ type FailoverOptions struct {
Password string
DB int
MaxRetries int
MaxRetries int
MinRetryBackoff time.Duration
MaxRetryBackoff time.Duration
DialTimeout time.Duration
ReadTimeout time.Duration
WriteTimeout time.Duration
PoolSize int
MinIdleConns int
MaxConnAge time.Duration
PoolTimeout time.Duration
IdleTimeout time.Duration
IdleCheckFrequency time.Duration
@@ -92,7 +96,7 @@ func NewFailoverClient(failoverOpt *FailoverOptions) *Client {
},
}
c.baseClient.init()
c.setProcessor(c.Process)
c.cmdable.setProcessor(c.Process)
return &c
}
@@ -116,7 +120,7 @@ func NewSentinelClient(opt *Options) *SentinelClient {
}
func (c *SentinelClient) PubSub() *PubSub {
return &PubSub{
pubsub := &PubSub{
opt: c.opt,
newConn: func(channels []string) (*pool.Conn, error) {
@@ -124,6 +128,8 @@ func (c *SentinelClient) PubSub() *PubSub {
},
closeConn: c.connPool.CloseConn,
}
pubsub.init()
return pubsub
}
func (c *SentinelClient) GetMasterAddrByName(name string) *StringSliceCmd {
@@ -180,10 +186,7 @@ func (d *sentinelFailover) MasterAddr() (string, error) {
if err != nil {
return "", err
}
if d._masterAddr != addr {
d.switchMaster(addr)
}
d._switchMaster(addr)
return addr, nil
}
@@ -194,11 +197,11 @@ func (d *sentinelFailover) masterAddr() (string, error) {
addr, err := d.sentinel.GetMasterAddrByName(d.masterName).Result()
if err == nil {
addr := net.JoinHostPort(addr[0], addr[1])
internal.Logf("sentinel: master=%q addr=%q", d.masterName, addr)
return addr, nil
}
internal.Logf("sentinel: GetMasterAddrByName name=%q failed: %s", d.masterName, err)
internal.Logf("sentinel: GetMasterAddrByName name=%q failed: %s",
d.masterName, err)
d._resetSentinel()
}
@@ -234,15 +237,23 @@ func (d *sentinelFailover) masterAddr() (string, error) {
return "", errors.New("redis: all sentinels are unreachable")
}
func (d *sentinelFailover) switchMaster(masterAddr string) {
internal.Logf(
"sentinel: new master=%q addr=%q",
d.masterName, masterAddr,
)
_ = d.Pool().Filter(func(cn *pool.Conn) bool {
return cn.RemoteAddr().String() != masterAddr
func (c *sentinelFailover) switchMaster(addr string) {
c.mu.Lock()
c._switchMaster(addr)
c.mu.Unlock()
}
func (c *sentinelFailover) _switchMaster(addr string) {
if c._masterAddr == addr {
return
}
internal.Logf("sentinel: new master=%q addr=%q",
c.masterName, addr)
_ = c.Pool().Filter(func(cn *pool.Conn) bool {
return cn.RemoteAddr().String() != addr
})
d._masterAddr = masterAddr
c._masterAddr = addr
}
func (d *sentinelFailover) setSentinel(sentinel *SentinelClient) {
@@ -292,27 +303,25 @@ func (d *sentinelFailover) discoverSentinels(sentinel *SentinelClient) {
}
func (d *sentinelFailover) listen(sentinel *SentinelClient) {
var pubsub *PubSub
for {
if pubsub == nil {
pubsub = sentinel.PubSub()
pubsub := sentinel.PubSub()
defer pubsub.Close()
if err := pubsub.Subscribe("+switch-master"); err != nil {
internal.Logf("sentinel: Subscribe failed: %s", err)
pubsub.Close()
err := pubsub.Subscribe("+switch-master")
if err != nil {
internal.Logf("sentinel: Subscribe failed: %s", err)
d.resetSentinel()
return
}
for {
msg, err := pubsub.ReceiveMessage()
if err != nil {
if err == pool.ErrClosed {
d.resetSentinel()
return
}
}
msg, err := pubsub.ReceiveMessage()
if err != nil {
if err != pool.ErrClosed {
internal.Logf("sentinel: ReceiveMessage failed: %s", err)
pubsub.Close()
}
d.resetSentinel()
return
internal.Logf("sentinel: ReceiveMessage failed: %s", err)
continue
}
switch msg.Channel {
@@ -323,12 +332,7 @@ func (d *sentinelFailover) listen(sentinel *SentinelClient) {
continue
}
addr := net.JoinHostPort(parts[3], parts[4])
d.mu.Lock()
if d._masterAddr != addr {
d.switchMaster(addr)
}
d.mu.Unlock()
d.switchMaster(addr)
}
}
}

12
vendor/github.com/go-redis/redis/tx.go сгенерированный поставляемый
Просмотреть файл

@@ -29,6 +29,10 @@ func (c *Client) newTx() *Tx {
return &tx
}
// Watch prepares a transcaction and marks the keys to be watched
// for conditional execution if there are any keys.
//
// The transaction is automatically closed when the fn exits.
func (c *Client) Watch(fn func(*Tx) error, keys ...string) error {
tx := c.newTx()
if len(keys) > 0 {
@@ -74,6 +78,7 @@ func (c *Tx) Unwatch(keys ...string) *StatusCmd {
return cmd
}
// Pipeline creates a new pipeline. It is more convenient to use Pipelined.
func (c *Tx) Pipeline() Pipeliner {
pipe := Pipeline{
exec: c.processTxPipeline,
@@ -82,23 +87,24 @@ func (c *Tx) Pipeline() Pipeliner {
return &pipe
}
// Pipelined executes commands queued in the fn in a transaction
// and restores the connection state to normal.
// Pipelined executes commands queued in the fn in a transaction.
//
// When using WATCH, EXEC will execute commands only if the watched keys
// were not modified, allowing for a check-and-set mechanism.
//
// Exec always returns list of commands. If transaction fails
// TxFailedErr is returned. Otherwise Exec returns error of the first
// TxFailedErr is returned. Otherwise Exec returns an error of the first
// failed command or nil.
func (c *Tx) Pipelined(fn func(Pipeliner) error) ([]Cmder, error) {
return c.Pipeline().Pipelined(fn)
}
// TxPipelined is an alias for Pipelined.
func (c *Tx) TxPipelined(fn func(Pipeliner) error) ([]Cmder, error) {
return c.Pipelined(fn)
}
// TxPipeline is an alias for Pipeline.
func (c *Tx) TxPipeline() Pipeliner {
return c.Pipeline()
}

106
vendor/github.com/go-redis/redis/universal.go сгенерированный поставляемый
Просмотреть файл

@@ -12,35 +12,38 @@ type UniversalOptions struct {
// of cluster/sentinel nodes.
Addrs []string
// The sentinel master name.
// Only failover clients.
MasterName string
// Database to be selected after connecting to the server.
// Only single-node and failover clients.
DB int
// Only cluster clients.
// Enables read only queries on slave nodes.
ReadOnly bool
MaxRedirects int
RouteByLatency bool
// Common options
// Common options.
OnConnect func(*Conn) error
MaxRetries int
Password string
MaxRetries int
MinRetryBackoff time.Duration
MaxRetryBackoff time.Duration
DialTimeout time.Duration
ReadTimeout time.Duration
WriteTimeout time.Duration
PoolSize int
MinIdleConns int
MaxConnAge time.Duration
PoolTimeout time.Duration
IdleTimeout time.Duration
IdleCheckFrequency time.Duration
TLSConfig *tls.Config
// Only cluster clients.
MaxRedirects int
ReadOnly bool
RouteByLatency bool
RouteRandomly bool
// The sentinel master name.
// Only failover clients.
MasterName string
}
func (o *UniversalOptions) cluster() *ClusterOptions {
@@ -49,22 +52,31 @@ func (o *UniversalOptions) cluster() *ClusterOptions {
}
return &ClusterOptions{
Addrs: o.Addrs,
MaxRedirects: o.MaxRedirects,
RouteByLatency: o.RouteByLatency,
ReadOnly: o.ReadOnly,
Addrs: o.Addrs,
OnConnect: o.OnConnect,
Password: o.Password,
MaxRedirects: o.MaxRedirects,
ReadOnly: o.ReadOnly,
RouteByLatency: o.RouteByLatency,
RouteRandomly: o.RouteRandomly,
MaxRetries: o.MaxRetries,
MinRetryBackoff: o.MinRetryBackoff,
MaxRetryBackoff: o.MaxRetryBackoff,
OnConnect: o.OnConnect,
MaxRetries: o.MaxRetries,
Password: o.Password,
DialTimeout: o.DialTimeout,
ReadTimeout: o.ReadTimeout,
WriteTimeout: o.WriteTimeout,
PoolSize: o.PoolSize,
MinIdleConns: o.MinIdleConns,
MaxConnAge: o.MaxConnAge,
PoolTimeout: o.PoolTimeout,
IdleTimeout: o.IdleTimeout,
IdleCheckFrequency: o.IdleCheckFrequency,
TLSConfig: o.TLSConfig,
TLSConfig: o.TLSConfig,
}
}
@@ -76,19 +88,27 @@ func (o *UniversalOptions) failover() *FailoverOptions {
return &FailoverOptions{
SentinelAddrs: o.Addrs,
MasterName: o.MasterName,
DB: o.DB,
OnConnect: o.OnConnect,
DB: o.DB,
Password: o.Password,
MaxRetries: o.MaxRetries,
MinRetryBackoff: o.MinRetryBackoff,
MaxRetryBackoff: o.MaxRetryBackoff,
DialTimeout: o.DialTimeout,
ReadTimeout: o.ReadTimeout,
WriteTimeout: o.WriteTimeout,
OnConnect: o.OnConnect,
MaxRetries: o.MaxRetries,
Password: o.Password,
DialTimeout: o.DialTimeout,
ReadTimeout: o.ReadTimeout,
WriteTimeout: o.WriteTimeout,
PoolSize: o.PoolSize,
MinIdleConns: o.MinIdleConns,
MaxConnAge: o.MaxConnAge,
PoolTimeout: o.PoolTimeout,
IdleTimeout: o.IdleTimeout,
IdleCheckFrequency: o.IdleCheckFrequency,
TLSConfig: o.TLSConfig,
TLSConfig: o.TLSConfig,
}
}
@@ -99,20 +119,28 @@ func (o *UniversalOptions) simple() *Options {
}
return &Options{
Addr: addr,
DB: o.DB,
Addr: addr,
OnConnect: o.OnConnect,
DB: o.DB,
Password: o.Password,
MaxRetries: o.MaxRetries,
MinRetryBackoff: o.MinRetryBackoff,
MaxRetryBackoff: o.MaxRetryBackoff,
DialTimeout: o.DialTimeout,
ReadTimeout: o.ReadTimeout,
WriteTimeout: o.WriteTimeout,
OnConnect: o.OnConnect,
MaxRetries: o.MaxRetries,
Password: o.Password,
DialTimeout: o.DialTimeout,
ReadTimeout: o.ReadTimeout,
WriteTimeout: o.WriteTimeout,
PoolSize: o.PoolSize,
MinIdleConns: o.MinIdleConns,
MaxConnAge: o.MaxConnAge,
PoolTimeout: o.PoolTimeout,
IdleTimeout: o.IdleTimeout,
IdleCheckFrequency: o.IdleCheckFrequency,
TLSConfig: o.TLSConfig,
TLSConfig: o.TLSConfig,
}
}