3 Star 0 Fork 0

Gitee 极速下载/go-redis-redissource

加入 Gitee
与超过 1200万 开发者一起发现、参与优秀开源项目,私有仓库也完全免费 :)
免费加入
此仓库是为了提升国内下载速度的镜像仓库,每日同步一次。 原始仓库: https://github.com/go-redis/redis
克隆/下载
pubsub.go 4.50 KB
一键复制 编辑 原始数据 按行查看 历史
Vladimir Mihailenco 提交于 2015-07-11 13:42 . pubsub: improve API and docs.
package redis
import (
"fmt"
"time"
)
// Posts a message to the given channel.
func (c *Client) Publish(channel, message string) *IntCmd {
req := NewIntCmd("PUBLISH", channel, message)
c.Process(req)
return req
}
// PubSub implements Pub/Sub commands as described in
// http://redis.io/topics/pubsub.
type PubSub struct {
*baseClient
}
// Deprecated. Use Subscribe/PSubscribe instead.
func (c *Client) PubSub() *PubSub {
return &PubSub{
baseClient: &baseClient{
opt: c.opt,
connPool: newSingleConnPool(c.connPool, false),
},
}
}
// Subscribes the client to the specified channels.
func (c *Client) Subscribe(channels ...string) (*PubSub, error) {
pubsub := c.PubSub()
return pubsub, pubsub.Subscribe(channels...)
}
// Subscribes the client to the given patterns.
func (c *Client) PSubscribe(channels ...string) (*PubSub, error) {
pubsub := c.PubSub()
return pubsub, pubsub.PSubscribe(channels...)
}
func (c *PubSub) Ping(payload string) error {
cn, err := c.conn()
if err != nil {
return err
}
args := []interface{}{"PING"}
if payload != "" {
args = append(args, payload)
}
cmd := NewCmd(args...)
return cn.writeCmds(cmd)
}
// Message 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 {
Channel string
Payload string
}
func (m *Message) String() string {
return fmt.Sprintf("Message<%s: %s>", m.Channel, m.Payload)
}
// Message matching a pattern-matching subscription received as result
// of a PUBLISH command issued by another client.
type PMessage struct {
Channel string
Pattern string
Payload string
}
func (m *PMessage) String() string {
return fmt.Sprintf("PMessage<%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"
}
// Returns a message as a Subscription, Message, PMessage, Pong or
// error. See PubSub example for details.
func (c *PubSub) Receive() (interface{}, error) {
return c.ReceiveTimeout(0)
}
func newMessage(reply []interface{}) (interface{}, error) {
switch kind := reply[0].(string); kind {
case "subscribe", "unsubscribe", "psubscribe", "punsubscribe":
return &Subscription{
Kind: kind,
Channel: reply[1].(string),
Count: int(reply[2].(int64)),
}, nil
case "message":
return &Message{
Channel: reply[1].(string),
Payload: reply[2].(string),
}, nil
case "pmessage":
return &PMessage{
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 notification: %q", kind)
}
}
// ReceiveTimeout acts like Receive but returns an error if message
// is not received in time.
func (c *PubSub) ReceiveTimeout(timeout time.Duration) (interface{}, error) {
cn, err := c.conn()
if err != nil {
return nil, err
}
cn.ReadTimeout = timeout
cmd := NewSliceCmd()
if err := cmd.parseReply(cn.rd); err != nil {
return nil, err
}
return newMessage(cmd.Val())
}
func (c *PubSub) subscribe(cmd string, channels ...string) error {
cn, err := c.conn()
if err != nil {
return err
}
args := make([]interface{}, 1+len(channels))
args[0] = cmd
for i, channel := range channels {
args[1+i] = channel
}
req := NewSliceCmd(args...)
return cn.writeCmds(req)
}
// Subscribes the client to the specified channels.
func (c *PubSub) Subscribe(channels ...string) error {
return c.subscribe("SUBSCRIBE", channels...)
}
// Subscribes the client to the given patterns.
func (c *PubSub) PSubscribe(patterns ...string) error {
return c.subscribe("PSUBSCRIBE", patterns...)
}
// Unsubscribes the client from the given channels, or from all of
// them if none is given.
func (c *PubSub) Unsubscribe(channels ...string) error {
return c.subscribe("UNSUBSCRIBE", channels...)
}
// Unsubscribes the client from the given patterns, or from all of
// them if none is given.
func (c *PubSub) PUnsubscribe(patterns ...string) error {
return c.subscribe("PUNSUBSCRIBE", patterns...)
}
马建仓 AI 助手
尝试更多
代码解读
代码找茬
代码优化
1
https://gitee.com/mirrors/redissource.git
git@gitee.com:mirrors/redissource.git
mirrors
redissource
go-redis-redissource
v3.2.4

搜索帮助