40 Star 146 Fork 3

Gitee 极速下载/grafana

加入 Gitee
与超过 1200万 开发者一起发现、参与优秀开源项目,私有仓库也完全免费 :)
免费加入
文件
此仓库是为了提升国内下载速度的镜像仓库,每日同步一次。 原始仓库: https://github.com/grafana/grafana
克隆/下载
pubsub.go 2.64 KB
一键复制 编辑 原始数据 按行查看 历史
package redis
import (
"fmt"
"time"
)
// Not thread-safe.
type PubSub struct {
*baseClient
}
func (c *Client) PubSub() *PubSub {
return &PubSub{
baseClient: &baseClient{
opt: c.opt,
connPool: newSingleConnPool(c.connPool, false),
},
}
}
func (c *Client) Publish(channel, message string) *IntCmd {
req := NewIntCmd("PUBLISH", channel, message)
c.Process(req)
return req
}
type Message struct {
Channel string
Payload string
}
func (m *Message) String() string {
return fmt.Sprintf("Message<%s: %s>", m.Channel, m.Payload)
}
type PMessage struct {
Channel string
Pattern string
Payload string
}
func (m *PMessage) String() string {
return fmt.Sprintf("PMessage<%s: %s>", m.Channel, m.Payload)
}
type Subscription struct {
Kind string
Channel string
Count int
}
func (m *Subscription) String() string {
return fmt.Sprintf("%s: %s", m.Kind, m.Channel)
}
func (c *PubSub) Receive() (interface{}, error) {
return c.ReceiveTimeout(0)
}
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
}
reply := cmd.Val()
msgName := reply[0].(string)
switch msgName {
case "subscribe", "unsubscribe", "psubscribe", "punsubscribe":
return &Subscription{
Kind: msgName,
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
}
return nil, fmt.Errorf("redis: unsupported message name: %q", msgName)
}
func (c *PubSub) subscribe(cmd string, channels ...string) error {
cn, err := c.conn()
if err != nil {
return err
}
args := append([]string{cmd}, channels...)
req := NewSliceCmd(args...)
return c.writeCmd(cn, req)
}
func (c *PubSub) Subscribe(channels ...string) error {
return c.subscribe("SUBSCRIBE", channels...)
}
func (c *PubSub) PSubscribe(patterns ...string) error {
return c.subscribe("PSUBSCRIBE", patterns...)
}
func (c *PubSub) unsubscribe(cmd string, channels ...string) error {
cn, err := c.conn()
if err != nil {
return err
}
args := append([]string{cmd}, channels...)
req := NewSliceCmd(args...)
return c.writeCmd(cn, req)
}
func (c *PubSub) Unsubscribe(channels ...string) error {
return c.unsubscribe("UNSUBSCRIBE", channels...)
}
func (c *PubSub) PUnsubscribe(patterns ...string) error {
return c.unsubscribe("PUNSUBSCRIBE", patterns...)
}
马建仓 AI 助手
尝试更多
代码解读
代码找茬
代码优化
JavaScript
1
https://gitee.com/mirrors/grafana.git
git@gitee.com:mirrors/grafana.git
mirrors
grafana
grafana
v2.1.2

搜索帮助

0d507c66 1850385 C8b1a773 1850385