代码拉取完成,页面将自动刷新
package bus
import "sync"
type EventChan chan Event
var _ Publisher = (*Bus)(nil)
var _ Subscriber = (*Bus)(nil)
type Bus struct {
subscribers map[string][]EventChan
rm sync.RWMutex
}
func New() *Bus {
return &Bus{
subscribers: make(map[string][]EventChan, 0),
}
}
func (bus *Bus) Subscribe(topic string, ch EventChan) {
bus.rm.Lock()
defer bus.rm.Unlock()
if prev, found := bus.subscribers[topic]; found {
bus.subscribers[topic] = append(prev, ch)
} else {
bus.subscribers[topic] = append([]EventChan{}, ch)
}
}
func (bus *Bus) UnSubscribe(topic string, ch EventChan) {
bus.rm.Lock()
defer bus.rm.Unlock()
channels, found := bus.subscribers[topic]
if !found {
return
}
for i, c := range channels {
if c == ch {
channels = append(channels[:i], channels[i+1:]...)
}
}
}
func (bus *Bus) Start() error {
return nil
}
func (bus *Bus) Stop() {
}
func (bus *Bus) Publish(topic string, ev Event) {
bus.rm.RLock()
defer bus.rm.RUnlock()
ch, found := bus.subscribers[topic]
if !found {
return
}
channels := append([]EventChan{}, ch...)
go func(ev Event, channel []EventChan) {
for _, ch1 := range channel {
ch1 <- ev
}
}(ev, channels)
}
此处可能存在不合适展示的内容,页面不予展示。您可通过相关编辑功能自查并修改。
如您确认内容无涉及 不当用语 / 纯广告导流 / 暴力 / 低俗色情 / 侵权 / 盗版 / 虚假 / 无价值内容或违法国家有关法律法规的内容,可点击提交进行申诉,我们将尽快为您处理。