fea redis subscribe

cache/redis.go

@@ -149,7 +149,7 @@ func (c RedisCache) Info() (string, error) {
 }
 
 // Subscribe subscribe message
-func (c RedisCache) Subscribe(channels string, cb func(channel string, message string, err error)) {
+func (c RedisCache) Subscribe(channels string, cb func(channel string, message string, err error)) error {
 	pubsub := &redis.PubSub{}
 	if c.cc != nil {
 		pubsub = c.cc.Subscribe(channels)
@@ -169,20 +169,17 @@ func (c RedisCache) Subscribe(channels string, cb func(channel string, message s
 	}
 
 	cb("", "", err)
-
-	return
+	return err
 }
 
 // Publish publish message
 func (c RedisCache) Publish(channel, message string) error {
+	if c.cc != nil {
+		return c.cc.Publish(channel, message).Err()
+	}
 	return c.c.Publish(channel, message).Err()
 }
 
-// PublishCluster publish message
-func (c RedisCache) PublishCluster(channel, message string) error {
-	return c.cc.Publish(channel, message).Err()
-}
-
 func connect() (*RedisCache, error) {
 	if cache != nil {
 		return cache, nil
@@ -239,3 +236,22 @@ func Info() (string, error) {
 
 	return c.Info()
 }
+
+// Publish publish message
+func Publish(channel, message string) error {
+	c, err := connect()
+	if err != nil {
+		return err
+	}
+
+	return c.Publish(channel, message)
+}
+
+func Subscribe(channels string, cb func(channel string, message string, err error)) error {
+	c, err := connect()
+	if err != nil {
+		return err
+	}
+
+	return c.Subscribe(channels, cb)
+}