You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
630 lines
14 KiB
Go
630 lines
14 KiB
Go
5 years ago
|
package redis
|
||
|
|
||
|
import (
|
||
4 years ago
|
"context"
|
||
5 years ago
|
"errors"
|
||
|
"fmt"
|
||
4 years ago
|
"strings"
|
||
5 years ago
|
"sync"
|
||
|
"time"
|
||
|
|
||
4 years ago
|
"github.com/go-redis/redis/v8/internal"
|
||
|
"github.com/go-redis/redis/v8/internal/pool"
|
||
|
"github.com/go-redis/redis/v8/internal/proto"
|
||
5 years ago
|
)
|
||
|
|
||
4 years ago
|
const (
|
||
|
pingTimeout = time.Second
|
||
|
chanSendTimeout = time.Minute
|
||
|
)
|
||
4 years ago
|
|
||
5 years ago
|
var errPingTimeout = errors.New("redis: ping timeout")
|
||
|
|
||
4 years ago
|
// PubSub implements Pub/Sub commands as described in
|
||
5 years ago
|
// http://redis.io/topics/pubsub. Message receiving is NOT safe
|
||
|
// for concurrent use by multiple goroutines.
|
||
|
//
|
||
|
// PubSub automatically reconnects to Redis Server and resubscribes
|
||
|
// to the channels in case of network errors.
|
||
|
type PubSub struct {
|
||
|
opt *Options
|
||
|
|
||
4 years ago
|
newConn func(ctx context.Context, channels []string) (*pool.Conn, error)
|
||
5 years ago
|
closeConn func(*pool.Conn) error
|
||
|
|
||
|
mu sync.Mutex
|
||
|
cn *pool.Conn
|
||
|
channels map[string]struct{}
|
||
|
patterns map[string]struct{}
|
||
4 years ago
|
|
||
|
closed bool
|
||
|
exit chan struct{}
|
||
5 years ago
|
|
||
|
cmd *Cmd
|
||
|
|
||
|
chOnce sync.Once
|
||
4 years ago
|
msgCh chan *Message
|
||
|
allCh chan interface{}
|
||
5 years ago
|
ping chan struct{}
|
||
|
}
|
||
|
|
||
4 years ago
|
func (c *PubSub) String() string {
|
||
|
channels := mapKeys(c.channels)
|
||
|
channels = append(channels, mapKeys(c.patterns)...)
|
||
|
return fmt.Sprintf("PubSub(%s)", strings.Join(channels, ", "))
|
||
|
}
|
||
|
|
||
5 years ago
|
func (c *PubSub) init() {
|
||
|
c.exit = make(chan struct{})
|
||
|
}
|
||
|
|
||
4 years ago
|
func (c *PubSub) connWithLock(ctx context.Context) (*pool.Conn, error) {
|
||
5 years ago
|
c.mu.Lock()
|
||
4 years ago
|
cn, err := c.conn(ctx, nil)
|
||
5 years ago
|
c.mu.Unlock()
|
||
|
return cn, err
|
||
|
}
|
||
|
|
||
4 years ago
|
func (c *PubSub) conn(ctx context.Context, newChannels []string) (*pool.Conn, error) {
|
||
5 years ago
|
if c.closed {
|
||
|
return nil, pool.ErrClosed
|
||
|
}
|
||
|
if c.cn != nil {
|
||
|
return c.cn, nil
|
||
|
}
|
||
|
|
||
|
channels := mapKeys(c.channels)
|
||
|
channels = append(channels, newChannels...)
|
||
|
|
||
4 years ago
|
cn, err := c.newConn(ctx, channels)
|
||
5 years ago
|
if err != nil {
|
||
|
return nil, err
|
||
|
}
|
||
|
|
||
4 years ago
|
if err := c.resubscribe(ctx, cn); err != nil {
|
||
5 years ago
|
_ = c.closeConn(cn)
|
||
|
return nil, err
|
||
|
}
|
||
|
|
||
|
c.cn = cn
|
||
|
return cn, nil
|
||
|
}
|
||
|
|
||
4 years ago
|
func (c *PubSub) writeCmd(ctx context.Context, cn *pool.Conn, cmd Cmder) error {
|
||
|
return cn.WithWriter(ctx, c.opt.WriteTimeout, func(wr *proto.Writer) error {
|
||
5 years ago
|
return writeCmd(wr, cmd)
|
||
|
})
|
||
|
}
|
||
|
|
||
4 years ago
|
func (c *PubSub) resubscribe(ctx context.Context, cn *pool.Conn) error {
|
||
5 years ago
|
var firstErr error
|
||
|
|
||
|
if len(c.channels) > 0 {
|
||
4 years ago
|
firstErr = c._subscribe(ctx, cn, "subscribe", mapKeys(c.channels))
|
||
5 years ago
|
}
|
||
|
|
||
|
if len(c.patterns) > 0 {
|
||
4 years ago
|
err := c._subscribe(ctx, cn, "psubscribe", mapKeys(c.patterns))
|
||
5 years ago
|
if err != nil && firstErr == nil {
|
||
|
firstErr = err
|
||
|
}
|
||
|
}
|
||
|
|
||
|
return firstErr
|
||
|
}
|
||
|
|
||
|
func mapKeys(m map[string]struct{}) []string {
|
||
|
s := make([]string, len(m))
|
||
|
i := 0
|
||
|
for k := range m {
|
||
|
s[i] = k
|
||
|
i++
|
||
|
}
|
||
|
return s
|
||
|
}
|
||
|
|
||
|
func (c *PubSub) _subscribe(
|
||
4 years ago
|
ctx context.Context, cn *pool.Conn, redisCmd string, channels []string,
|
||
5 years ago
|
) error {
|
||
|
args := make([]interface{}, 0, 1+len(channels))
|
||
|
args = append(args, redisCmd)
|
||
|
for _, channel := range channels {
|
||
|
args = append(args, channel)
|
||
|
}
|
||
4 years ago
|
cmd := NewSliceCmd(ctx, args...)
|
||
|
return c.writeCmd(ctx, cn, cmd)
|
||
5 years ago
|
}
|
||
|
|
||
4 years ago
|
func (c *PubSub) releaseConnWithLock(
|
||
|
ctx context.Context,
|
||
|
cn *pool.Conn,
|
||
|
err error,
|
||
|
allowTimeout bool,
|
||
|
) {
|
||
5 years ago
|
c.mu.Lock()
|
||
4 years ago
|
c.releaseConn(ctx, cn, err, allowTimeout)
|
||
5 years ago
|
c.mu.Unlock()
|
||
|
}
|
||
|
|
||
4 years ago
|
func (c *PubSub) releaseConn(ctx context.Context, cn *pool.Conn, err error, allowTimeout bool) {
|
||
5 years ago
|
if c.cn != cn {
|
||
|
return
|
||
|
}
|
||
4 years ago
|
if isBadConn(err, allowTimeout) {
|
||
4 years ago
|
c.reconnect(ctx, err)
|
||
5 years ago
|
}
|
||
|
}
|
||
|
|
||
4 years ago
|
func (c *PubSub) reconnect(ctx context.Context, reason error) {
|
||
4 years ago
|
_ = c.closeTheCn(reason)
|
||
4 years ago
|
_, _ = c.conn(ctx, nil)
|
||
5 years ago
|
}
|
||
|
|
||
4 years ago
|
func (c *PubSub) closeTheCn(reason error) error {
|
||
5 years ago
|
if c.cn == nil {
|
||
|
return nil
|
||
|
}
|
||
|
if !c.closed {
|
||
4 years ago
|
internal.Logger.Printf(c.getContext(), "redis: discarding bad PubSub connection: %s", reason)
|
||
5 years ago
|
}
|
||
|
err := c.closeConn(c.cn)
|
||
|
c.cn = nil
|
||
|
return err
|
||
|
}
|
||
|
|
||
|
func (c *PubSub) Close() error {
|
||
|
c.mu.Lock()
|
||
|
defer c.mu.Unlock()
|
||
|
|
||
|
if c.closed {
|
||
|
return pool.ErrClosed
|
||
|
}
|
||
|
c.closed = true
|
||
|
close(c.exit)
|
||
|
|
||
4 years ago
|
return c.closeTheCn(pool.ErrClosed)
|
||
5 years ago
|
}
|
||
|
|
||
|
// Subscribe the client to the specified channels. It returns
|
||
|
// empty subscription if there are no channels.
|
||
4 years ago
|
func (c *PubSub) Subscribe(ctx context.Context, channels ...string) error {
|
||
5 years ago
|
c.mu.Lock()
|
||
|
defer c.mu.Unlock()
|
||
|
|
||
4 years ago
|
err := c.subscribe(ctx, "subscribe", channels...)
|
||
5 years ago
|
if c.channels == nil {
|
||
|
c.channels = make(map[string]struct{})
|
||
|
}
|
||
|
for _, s := range channels {
|
||
|
c.channels[s] = struct{}{}
|
||
|
}
|
||
|
return err
|
||
|
}
|
||
|
|
||
|
// PSubscribe the client to the given patterns. It returns
|
||
|
// empty subscription if there are no patterns.
|
||
4 years ago
|
func (c *PubSub) PSubscribe(ctx context.Context, patterns ...string) error {
|
||
5 years ago
|
c.mu.Lock()
|
||
|
defer c.mu.Unlock()
|
||
|
|
||
4 years ago
|
err := c.subscribe(ctx, "psubscribe", patterns...)
|
||
5 years ago
|
if c.patterns == nil {
|
||
|
c.patterns = make(map[string]struct{})
|
||
|
}
|
||
|
for _, s := range patterns {
|
||
|
c.patterns[s] = struct{}{}
|
||
|
}
|
||
|
return err
|
||
|
}
|
||
|
|
||
|
// Unsubscribe the client from the given channels, or from all of
|
||
|
// them if none is given.
|
||
4 years ago
|
func (c *PubSub) Unsubscribe(ctx context.Context, channels ...string) error {
|
||
5 years ago
|
c.mu.Lock()
|
||
|
defer c.mu.Unlock()
|
||
|
|
||
|
for _, channel := range channels {
|
||
|
delete(c.channels, channel)
|
||
|
}
|
||
4 years ago
|
err := c.subscribe(ctx, "unsubscribe", channels...)
|
||
5 years ago
|
return err
|
||
|
}
|
||
|
|
||
|
// PUnsubscribe the client from the given patterns, or from all of
|
||
|
// them if none is given.
|
||
4 years ago
|
func (c *PubSub) PUnsubscribe(ctx context.Context, patterns ...string) error {
|
||
5 years ago
|
c.mu.Lock()
|
||
|
defer c.mu.Unlock()
|
||
|
|
||
|
for _, pattern := range patterns {
|
||
|
delete(c.patterns, pattern)
|
||
|
}
|
||
4 years ago
|
err := c.subscribe(ctx, "punsubscribe", patterns...)
|
||
5 years ago
|
return err
|
||
|
}
|
||
|
|
||
4 years ago
|
func (c *PubSub) subscribe(ctx context.Context, redisCmd string, channels ...string) error {
|
||
|
cn, err := c.conn(ctx, channels)
|
||
5 years ago
|
if err != nil {
|
||
|
return err
|
||
|
}
|
||
|
|
||
4 years ago
|
err = c._subscribe(ctx, cn, redisCmd, channels)
|
||
|
c.releaseConn(ctx, cn, err, false)
|
||
5 years ago
|
return err
|
||
|
}
|
||
|
|
||
4 years ago
|
func (c *PubSub) Ping(ctx context.Context, payload ...string) error {
|
||
5 years ago
|
args := []interface{}{"ping"}
|
||
|
if len(payload) == 1 {
|
||
|
args = append(args, payload[0])
|
||
|
}
|
||
4 years ago
|
cmd := NewCmd(ctx, args...)
|
||
5 years ago
|
|
||
4 years ago
|
cn, err := c.connWithLock(ctx)
|
||
5 years ago
|
if err != nil {
|
||
|
return err
|
||
|
}
|
||
|
|
||
4 years ago
|
err = c.writeCmd(ctx, cn, cmd)
|
||
|
c.releaseConnWithLock(ctx, cn, err, false)
|
||
5 years ago
|
return err
|
||
|
}
|
||
|
|
||
|
// Subscription received after a successful subscription to channel.
|
||
|
type Subscription struct {
|
||
|
// Can be "subscribe", "unsubscribe", "psubscribe" or "punsubscribe".
|
||
|
Kind string
|
||
|
// Channel name we have subscribed to.
|
||
|
Channel string
|
||
|
// Number of channels we are currently subscribed to.
|
||
|
Count int
|
||
|
}
|
||
|
|
||
|
func (m *Subscription) String() string {
|
||
|
return fmt.Sprintf("%s: %s", m.Kind, m.Channel)
|
||
|
}
|
||
|
|
||
|
// Message received as result of a PUBLISH command issued by another client.
|
||
|
type Message struct {
|
||
4 years ago
|
Channel string
|
||
|
Pattern string
|
||
|
Payload string
|
||
|
PayloadSlice []string
|
||
5 years ago
|
}
|
||
|
|
||
|
func (m *Message) String() string {
|
||
|
return fmt.Sprintf("Message<%s: %s>", m.Channel, m.Payload)
|
||
|
}
|
||
|
|
||
|
// Pong received as result of a PING command issued by another client.
|
||
|
type Pong struct {
|
||
|
Payload string
|
||
|
}
|
||
|
|
||
|
func (p *Pong) String() string {
|
||
|
if p.Payload != "" {
|
||
|
return fmt.Sprintf("Pong<%s>", p.Payload)
|
||
|
}
|
||
|
return "Pong"
|
||
|
}
|
||
|
|
||
|
func (c *PubSub) newMessage(reply interface{}) (interface{}, error) {
|
||
|
switch reply := reply.(type) {
|
||
|
case string:
|
||
|
return &Pong{
|
||
|
Payload: reply,
|
||
|
}, nil
|
||
|
case []interface{}:
|
||
|
switch kind := reply[0].(string); kind {
|
||
|
case "subscribe", "unsubscribe", "psubscribe", "punsubscribe":
|
||
4 years ago
|
// Can be nil in case of "unsubscribe".
|
||
|
channel, _ := reply[1].(string)
|
||
5 years ago
|
return &Subscription{
|
||
|
Kind: kind,
|
||
4 years ago
|
Channel: channel,
|
||
5 years ago
|
Count: int(reply[2].(int64)),
|
||
|
}, nil
|
||
|
case "message":
|
||
4 years ago
|
switch payload := reply[2].(type) {
|
||
|
case string:
|
||
|
return &Message{
|
||
|
Channel: reply[1].(string),
|
||
|
Payload: payload,
|
||
|
}, nil
|
||
|
case []interface{}:
|
||
|
ss := make([]string, len(payload))
|
||
|
for i, s := range payload {
|
||
|
ss[i] = s.(string)
|
||
|
}
|
||
|
return &Message{
|
||
|
Channel: reply[1].(string),
|
||
|
PayloadSlice: ss,
|
||
|
}, nil
|
||
|
default:
|
||
|
return nil, fmt.Errorf("redis: unsupported pubsub message payload: %T", payload)
|
||
|
}
|
||
5 years ago
|
case "pmessage":
|
||
|
return &Message{
|
||
|
Pattern: reply[1].(string),
|
||
|
Channel: reply[2].(string),
|
||
|
Payload: reply[3].(string),
|
||
|
}, nil
|
||
|
case "pong":
|
||
|
return &Pong{
|
||
|
Payload: reply[1].(string),
|
||
|
}, nil
|
||
|
default:
|
||
|
return nil, fmt.Errorf("redis: unsupported pubsub message: %q", kind)
|
||
|
}
|
||
|
default:
|
||
|
return nil, fmt.Errorf("redis: unsupported pubsub message: %#v", reply)
|
||
|
}
|
||
|
}
|
||
|
|
||
|
// ReceiveTimeout acts like Receive but returns an error if message
|
||
|
// is not received in time. This is low-level API and in most cases
|
||
|
// Channel should be used instead.
|
||
4 years ago
|
func (c *PubSub) ReceiveTimeout(ctx context.Context, timeout time.Duration) (interface{}, error) {
|
||
5 years ago
|
if c.cmd == nil {
|
||
4 years ago
|
c.cmd = NewCmd(ctx)
|
||
5 years ago
|
}
|
||
|
|
||
4 years ago
|
cn, err := c.connWithLock(ctx)
|
||
5 years ago
|
if err != nil {
|
||
|
return nil, err
|
||
|
}
|
||
|
|
||
4 years ago
|
err = cn.WithReader(ctx, timeout, func(rd *proto.Reader) error {
|
||
5 years ago
|
return c.cmd.readReply(rd)
|
||
|
})
|
||
|
|
||
4 years ago
|
c.releaseConnWithLock(ctx, cn, err, timeout > 0)
|
||
5 years ago
|
if err != nil {
|
||
|
return nil, err
|
||
|
}
|
||
|
|
||
|
return c.newMessage(c.cmd.Val())
|
||
|
}
|
||
|
|
||
|
// Receive returns a message as a Subscription, Message, Pong or error.
|
||
|
// See PubSub example for details. This is low-level API and in most cases
|
||
|
// Channel should be used instead.
|
||
4 years ago
|
func (c *PubSub) Receive(ctx context.Context) (interface{}, error) {
|
||
|
return c.ReceiveTimeout(ctx, 0)
|
||
5 years ago
|
}
|
||
|
|
||
|
// 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.
|
||
4 years ago
|
func (c *PubSub) ReceiveMessage(ctx context.Context) (*Message, error) {
|
||
5 years ago
|
for {
|
||
4 years ago
|
msg, err := c.Receive(ctx)
|
||
5 years ago
|
if err != nil {
|
||
|
return nil, err
|
||
|
}
|
||
|
|
||
|
switch msg := msg.(type) {
|
||
|
case *Subscription:
|
||
|
// Ignore.
|
||
|
case *Pong:
|
||
|
// Ignore.
|
||
|
case *Message:
|
||
|
return msg, nil
|
||
|
default:
|
||
|
err := fmt.Errorf("redis: unknown message: %T", msg)
|
||
|
return nil, err
|
||
|
}
|
||
|
}
|
||
|
}
|
||
|
|
||
|
// Channel returns a Go channel for concurrently receiving messages.
|
||
4 years ago
|
// The channel is closed together with the PubSub. If the Go channel
|
||
|
// is blocked full for 30 seconds the message is dropped.
|
||
|
// Receive* APIs can not be used after channel is created.
|
||
|
//
|
||
|
// go-redis periodically sends ping messages to test connection health
|
||
|
// and re-subscribes if ping can not not received for 30 seconds.
|
||
5 years ago
|
func (c *PubSub) Channel() <-chan *Message {
|
||
4 years ago
|
return c.ChannelSize(100)
|
||
5 years ago
|
}
|
||
|
|
||
4 years ago
|
// ChannelSize is like Channel, but creates a Go channel
|
||
|
// with specified buffer size.
|
||
|
func (c *PubSub) ChannelSize(size int) <-chan *Message {
|
||
|
c.chOnce.Do(func() {
|
||
|
c.initPing()
|
||
|
c.initMsgChan(size)
|
||
|
})
|
||
|
if c.msgCh == nil {
|
||
|
err := fmt.Errorf("redis: Channel can't be called after ChannelWithSubscriptions")
|
||
|
panic(err)
|
||
|
}
|
||
|
if cap(c.msgCh) != size {
|
||
|
err := fmt.Errorf("redis: PubSub.Channel size can not be changed once created")
|
||
|
panic(err)
|
||
|
}
|
||
|
return c.msgCh
|
||
|
}
|
||
|
|
||
|
// ChannelWithSubscriptions is like Channel, but message type can be either
|
||
|
// *Subscription or *Message. Subscription messages can be used to detect
|
||
|
// reconnections.
|
||
|
//
|
||
|
// ChannelWithSubscriptions can not be used together with Channel or ChannelSize.
|
||
4 years ago
|
func (c *PubSub) ChannelWithSubscriptions(ctx context.Context, size int) <-chan interface{} {
|
||
4 years ago
|
c.chOnce.Do(func() {
|
||
|
c.initPing()
|
||
|
c.initAllChan(size)
|
||
|
})
|
||
|
if c.allCh == nil {
|
||
|
err := fmt.Errorf("redis: ChannelWithSubscriptions can't be called after Channel")
|
||
|
panic(err)
|
||
|
}
|
||
|
if cap(c.allCh) != size {
|
||
|
err := fmt.Errorf("redis: PubSub.Channel size can not be changed once created")
|
||
|
panic(err)
|
||
|
}
|
||
|
return c.allCh
|
||
|
}
|
||
|
|
||
4 years ago
|
func (c *PubSub) getContext() context.Context {
|
||
|
if c.cmd != nil {
|
||
|
return c.cmd.ctx
|
||
|
}
|
||
|
return context.Background()
|
||
|
}
|
||
|
|
||
4 years ago
|
func (c *PubSub) initPing() {
|
||
4 years ago
|
ctx := context.TODO()
|
||
4 years ago
|
c.ping = make(chan struct{}, 1)
|
||
|
go func() {
|
||
4 years ago
|
timer := time.NewTimer(time.Minute)
|
||
4 years ago
|
timer.Stop()
|
||
|
|
||
|
healthy := true
|
||
|
for {
|
||
|
timer.Reset(pingTimeout)
|
||
|
select {
|
||
|
case <-c.ping:
|
||
|
healthy = true
|
||
|
if !timer.Stop() {
|
||
|
<-timer.C
|
||
|
}
|
||
|
case <-timer.C:
|
||
4 years ago
|
pingErr := c.Ping(ctx)
|
||
4 years ago
|
if healthy {
|
||
|
healthy = false
|
||
|
} else {
|
||
|
if pingErr == nil {
|
||
|
pingErr = errPingTimeout
|
||
|
}
|
||
|
c.mu.Lock()
|
||
4 years ago
|
c.reconnect(ctx, pingErr)
|
||
4 years ago
|
healthy = true
|
||
|
c.mu.Unlock()
|
||
|
}
|
||
|
case <-c.exit:
|
||
|
return
|
||
|
}
|
||
|
}
|
||
|
}()
|
||
|
}
|
||
5 years ago
|
|
||
4 years ago
|
// initMsgChan must be in sync with initAllChan.
|
||
|
func (c *PubSub) initMsgChan(size int) {
|
||
4 years ago
|
ctx := context.TODO()
|
||
4 years ago
|
c.msgCh = make(chan *Message, size)
|
||
5 years ago
|
go func() {
|
||
4 years ago
|
timer := time.NewTimer(time.Minute)
|
||
4 years ago
|
timer.Stop()
|
||
|
|
||
5 years ago
|
var errCount int
|
||
|
for {
|
||
4 years ago
|
msg, err := c.Receive(ctx)
|
||
5 years ago
|
if err != nil {
|
||
|
if err == pool.ErrClosed {
|
||
4 years ago
|
close(c.msgCh)
|
||
5 years ago
|
return
|
||
|
}
|
||
|
if errCount > 0 {
|
||
4 years ago
|
time.Sleep(100 * time.Millisecond)
|
||
5 years ago
|
}
|
||
|
errCount++
|
||
|
continue
|
||
|
}
|
||
4 years ago
|
|
||
5 years ago
|
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:
|
||
4 years ago
|
timer.Reset(chanSendTimeout)
|
||
4 years ago
|
select {
|
||
|
case c.msgCh <- msg:
|
||
|
if !timer.Stop() {
|
||
|
<-timer.C
|
||
|
}
|
||
|
case <-timer.C:
|
||
|
internal.Logger.Printf(
|
||
4 years ago
|
c.getContext(),
|
||
|
"redis: %s channel is full for %s (message is dropped)",
|
||
|
c,
|
||
|
chanSendTimeout,
|
||
|
)
|
||
4 years ago
|
}
|
||
5 years ago
|
default:
|
||
4 years ago
|
internal.Logger.Printf(c.getContext(), "redis: unknown message type: %T", msg)
|
||
5 years ago
|
}
|
||
|
}
|
||
|
}()
|
||
4 years ago
|
}
|
||
5 years ago
|
|
||
4 years ago
|
// initAllChan must be in sync with initMsgChan.
|
||
|
func (c *PubSub) initAllChan(size int) {
|
||
4 years ago
|
ctx := context.TODO()
|
||
4 years ago
|
c.allCh = make(chan interface{}, size)
|
||
5 years ago
|
go func() {
|
||
4 years ago
|
timer := time.NewTimer(pingTimeout)
|
||
5 years ago
|
timer.Stop()
|
||
|
|
||
4 years ago
|
var errCount int
|
||
5 years ago
|
for {
|
||
4 years ago
|
msg, err := c.Receive(ctx)
|
||
4 years ago
|
if err != nil {
|
||
|
if err == pool.ErrClosed {
|
||
|
close(c.allCh)
|
||
|
return
|
||
5 years ago
|
}
|
||
4 years ago
|
if errCount > 0 {
|
||
4 years ago
|
time.Sleep(100 * time.Millisecond)
|
||
5 years ago
|
}
|
||
4 years ago
|
errCount++
|
||
|
continue
|
||
|
}
|
||
|
|
||
|
errCount = 0
|
||
|
|
||
|
// Any message is as good as a ping.
|
||
|
select {
|
||
|
case c.ping <- struct{}{}:
|
||
|
default:
|
||
|
}
|
||
|
|
||
|
switch msg := msg.(type) {
|
||
|
case *Subscription:
|
||
|
c.sendMessage(msg, timer)
|
||
|
case *Pong:
|
||
|
// Ignore.
|
||
|
case *Message:
|
||
|
c.sendMessage(msg, timer)
|
||
|
default:
|
||
4 years ago
|
internal.Logger.Printf(c.getContext(), "redis: unknown message type: %T", msg)
|
||
5 years ago
|
}
|
||
|
}
|
||
|
}()
|
||
|
}
|
||
|
|
||
4 years ago
|
func (c *PubSub) sendMessage(msg interface{}, timer *time.Timer) {
|
||
|
timer.Reset(pingTimeout)
|
||
|
select {
|
||
|
case c.allCh <- msg:
|
||
|
if !timer.Stop() {
|
||
|
<-timer.C
|
||
|
}
|
||
|
case <-timer.C:
|
||
|
internal.Logger.Printf(
|
||
4 years ago
|
c.getContext(),
|
||
4 years ago
|
"redis: %s channel is full for %s (message is dropped)", c, pingTimeout)
|
||
|
}
|
||
|
}
|