Skip to content

Commit

Permalink
pubsub: drop a message when the channel is full
Browse files Browse the repository at this point in the history
  • Loading branch information
vmihailenco committed Mar 12, 2019
1 parent 0d39ee8 commit 21913a8
Show file tree
Hide file tree
Showing 2 changed files with 38 additions and 7 deletions.
38 changes: 31 additions & 7 deletions pubsub.go
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ package redis
import (
"errors"
"fmt"
"strings"
"sync"
"time"

Expand All @@ -29,8 +30,9 @@ type PubSub struct {
cn *pool.Conn
channels map[string]struct{}
patterns map[string]struct{}
closed bool
exit chan struct{}

closed bool
exit chan struct{}

cmd *Cmd

Expand All @@ -39,6 +41,12 @@ type PubSub struct {
ping chan struct{}
}

func (c *PubSub) String() string {
channels := mapKeys(c.channels)
channels = append(channels, mapKeys(c.patterns)...)
return fmt.Sprintf("PubSub(%s)", strings.Join(channels, ", "))
}

func (c *PubSub) init() {
c.exit = make(chan struct{})
}
Expand Down Expand Up @@ -389,16 +397,23 @@ func (c *PubSub) ReceiveMessage() (*Message, error) {
// 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.
//
// If the Go channel is full for 30 seconds the message is dropped.
func (c *PubSub) Channel() <-chan *Message {
c.chOnce.Do(c.initChannel)
return c.ch
}

func (c *PubSub) initChannel() {
const timeout = 30 * time.Second

c.ch = make(chan *Message, 100)
c.ping = make(chan struct{}, 10)
c.ping = make(chan struct{}, 1)

go func() {
timer := time.NewTimer(timeout)
timer.Stop()

var errCount int
for {
msg, err := c.Receive()
Expand All @@ -413,6 +428,7 @@ func (c *PubSub) initChannel() {
errCount++
continue
}

errCount = 0

// Any message is as good as a ping.
Expand All @@ -427,16 +443,24 @@ func (c *PubSub) initChannel() {
case *Pong:
// Ignore.
case *Message:
c.ch <- msg
timer.Reset(timeout)
select {
case c.ch <- msg:
if !timer.Stop() {
<-timer.C
}
case <-timer.C:
internal.Logf(
"redis: %s channel is full for %s (message is dropped)",
c, timeout)
}
default:
internal.Logf("redis: unknown message: %T", msg)
internal.Logf("redis: unknown message type: %T", msg)
}
}
}()

go func() {
const timeout = 5 * time.Second

timer := time.NewTimer(timeout)
timer.Stop()

Expand Down
7 changes: 7 additions & 0 deletions pubsub_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,13 @@ var _ = Describe("PubSub", func() {
Expect(client.Close()).NotTo(HaveOccurred())
})

It("implements Stringer", func() {
pubsub := client.PSubscribe("mychannel*")
defer pubsub.Close()

Expect(pubsub.String()).To(Equal("PubSub(mychannel*)"))
})

It("should support pattern matching", func() {
pubsub := client.PSubscribe("mychannel*")
defer pubsub.Close()
Expand Down

0 comments on commit 21913a8

Please sign in to comment.