redis/go-redis
1
0
Fork 0
mirror of https://github.com/redis/go-redis.git synced 2025-07-28 06:42:00 +03:00
Code Activity

Simplify resubscribing in PubSub.

Browse Source
This commit is contained in:
Vladimir Mihailenco
2016-09-29 12:07:04 +00:00
parent 833b0c68df
commit e57ac63b6e
14 changed files with 90 additions and 93 deletions
Download Patch File Download Diff File Expand all files Collapse all files

93
pubsub.go
View File

@ -18,12 +18,25 @@ type PubSub struct {
channels []string
patterns []string
}
nsub int // number of active subscriptions
func (c *PubSub) conn() (*pool.Conn, bool, error) {
cn, isNew, err := c.base.conn()
if err != nil {
return nil, false, err
}
if isNew {
c.resubscribe()
}
return cn, isNew, nil
}
func (c *PubSub) putConn(cn *pool.Conn, err error) {
c.base.putConn(cn, err, true)
}
func (c *PubSub) subscribe(redisCmd string, channels ...string) error {
cn, err := c.base.conn()
cn, _, err := c.conn()
if err != nil {
return err
}
@ -44,7 +57,6 @@ func (c *PubSub) Subscribe(channels ...string) error {
err := c.subscribe("SUBSCRIBE", channels...)
if err == nil {
c.channels = appendIfNotExists(c.channels, channels...)
c.nsub += len(channels)
}
return err
}
@ -54,43 +66,10 @@ func (c *PubSub) PSubscribe(patterns ...string) error {
err := c.subscribe("PSUBSCRIBE", patterns...)
if err == nil {
c.patterns = appendIfNotExists(c.patterns, patterns...)
c.nsub += len(patterns)
}
return err
}
func remove(ss []string, es ...string) []string {
if len(es) == 0 {
return ss[:0]
}
for _, e := range es {
for i, s := range ss {
if s == e {
ss = append(ss[:i], ss[i+1:]...)
break
}
}
}
return ss
}
func appendIfNotExists(ss []string, es ...string) []string {
for _, e := range es {
found := false
for _, s := range ss {
if s == e {
found = true
break
}
}
if !found {
ss = append(ss, e)
}
}
return ss
}
// Unsubscribes the client from the given channels, or from all of
// them if none is given.
func (c *PubSub) Unsubscribe(channels ...string) error {
@ -116,7 +95,7 @@ func (c *PubSub) Close() error {
}
func (c *PubSub) Ping(payload string) error {
cn, err := c.base.conn()
cn, _, err := c.conn()
if err != nil {
return err
}
@ -198,11 +177,7 @@ func (c *PubSub) newMessage(reply []interface{}) (interface{}, error) {
// is not received in time. This is low-level API and most clients
// should use ReceiveMessage.
func (c *PubSub) ReceiveTimeout(timeout time.Duration) (interface{}, error) {
if c.nsub == 0 {
c.resubscribe()
}
cn, err := c.base.conn()
cn, _, err := c.conn()
if err != nil {
return nil, err
}
@ -274,12 +249,6 @@ func (c *PubSub) receiveMessage(timeout time.Duration) (*Message, error) {
}
}
func (c *PubSub) putConn(cn *pool.Conn, err error) {
if !c.base.putConn(cn, err, true) {
c.nsub = 0
}
}
func (c *PubSub) resubscribe() {
if c.base.closed() {
return
@ -295,3 +264,31 @@ func (c *PubSub) resubscribe() {
}
}
}
func remove(ss []string, es ...string) []string {
if len(es) == 0 {
return ss[:0]
}
for _, e := range es {
for i, s := range ss {
if s == e {
ss = append(ss[:i], ss[i+1:]...)
break
}
}
}
return ss
}
func appendIfNotExists(ss []string, es ...string) []string {
loop:
for _, e := range es {
for _, s := range ss {
if s == e {
continue loop
}
}
ss = append(ss, e)
}
return ss
}