The event package implements the event publishing and subscription mode within the same process.
Currently this part of the code is marked as Deprecated, telling the user to use the feed object. However, it is still used in the code. And there is not much code in this part. Just a brief introduction.
The data structure TypeMux is the main use. Subm records all subscribers. You can see that there are many subscribers for each type.
Feed implements one-to-many subscriptions
// TypeMuxEvent is a time-tagged notification pushed to subscribers.
type TypeMuxEvent struct {
Time time.Time
Data interface{}
}
// A TypeMux dispatches events to registered receivers. Receivers can be
// registered to handle events of certain type. Any operation
// called after mux is stopped will return ErrMuxClosed.
//
// The zero value is ready to use.
//
// Deprecated: use Feed
type TypeMux struct {
mutex sync.RWMutex
subm map[reflect.Type][]*TypeMuxSubscription
stopped bool
}Create a subscription that can subscribe to multiple types at the same time.
// Subscribe creates a subscription for events of the given types. The
// subscription's channel is closed when it is unsubscribed
// or the mux is closed.
func (mux *TypeMux) Subscribe(types ...interface{}) *TypeMuxSubscription {
sub := newsub(mux)
mux.mutex.Lock()
defer mux.mutex.Unlock()
if mux.stopped {
// set the status to closed so that calling Unsubscribe after this
// call will short circuit.
sub.closed = true
close(sub.postC)
} else {
if mux.subm == nil {
mux.subm = make(map[reflect.Type][]*TypeMuxSubscription)
}
for _, t := range types {
rtyp := reflect.TypeOf(t)
oldsubs := mux.subm[rtyp]
if find(oldsubs, sub) != -1 {
panic(fmt.Sprintf("event: duplicate type %s in Subscribe", rtyp))
}
subs := make([]*TypeMuxSubscription, len(oldsubs)+1)
copy(subs, oldsubs)
subs[len(oldsubs)] = sub
mux.subm[rtyp] = subs
}
}
return sub
}
// TypeMuxSubscription is a subscription established through TypeMux.
type TypeMuxSubscription struct {
mux *TypeMux
created time.Time
closeMu sync.Mutex
closing chan struct{}
closed bool
// these two are the same channel. they are stored separately so
// postC can be set to nil without affecting the return value of
// Chan.
postMu sync.RWMutex
// readC and postC are actually the same channel. But one is reading from the channel, one is only written from the channel
// Single direction channel
readC <-chan *TypeMuxEvent
postC chan<- *TypeMuxEvent
}
func newsub(mux *TypeMux) *TypeMuxSubscription {
c := make(chan *TypeMuxEvent)
return &TypeMuxSubscription{
mux: mux,
created: time.Now(),
readC: c,
postC: c,
closing: make(chan struct{}),
}
}Publish an event to TypeMux. At this time, all messages that subscribe to this type will receive this message.
// Post sends an event to all receivers registered for the given type.
// It returns ErrMuxClosed if the mux has been stopped.
func (mux *TypeMux) Post(ev interface{}) error {
event := &TypeMuxEvent{
Time: time.Now(),
Data: ev,
}
rtyp := reflect.TypeOf(ev)
mux.mutex.RLock()
if mux.stopped {
mux.mutex.RUnlock()
return ErrMuxClosed
}
subs := mux.subm[rtyp]
mux.mutex.RUnlock()
for _, sub := range subs {
// Blocked delivery.
sub.deliver(event)
}
return nil
}
func (s *TypeMuxSubscription) deliver(event *TypeMuxEvent) {
// Short circuit delivery if stale event
if s.created.After(event.Time) {
return
}
// Otherwise deliver the event
s.postMu.RLock()
defer s.postMu.RUnlock()
select { // Blocking method
case s.postC <- event:
case <-s.closing:
}
}The main object currently used. Replaced the TypeMux inside the event.go.
feed data structure
// Feed implements one-to-many subscriptions where the carrier of events is a channel.
// Values sent to a Feed are delivered to all subscribed channels simultaneously.
// Feeds can only be used with a single type. The type is determined by the first Send or
// Subscribe operation. Subsequent calls to these methods panic if the type does not
// match.
// The zero value is ready to use.
type Feed struct {
once sync.Once // ensures that init only runs once
sendLock chan struct{} // sendLock has a one-element buffer and is empty when held.It protects sendCases.
removeSub chan interface{} // interrupts Send
sendCases caseList // the active set of select cases used by Send
// The inbox holds newly subscribed channels until they are added to sendCases.
mu sync.Mutex
inbox caseList
etype reflect.Type
closed bool
}Initialization will be guaranteed by once to ensure that it will only be executed once.
func (f *Feed) init() {
f.removeSub = make(chan interface{})
f.sendLock = make(chan struct{}, 1)
f.sendLock <- struct{}{}
f.sendCases = caseList{{Chan: reflect.ValueOf(f.removeSub), Dir: reflect.SelectRecv}}
}Subscribe to post a channel. Relative to the event. The event subscription is passed in the type that needs to be subscribed, and then the channel is built and returned in the event's subscription code. This mode of direct delivery of channels may be more flexible. Then a SelectCase is generated based on the incoming channel. Put it in the inbox.
// Subscribe adds a channel to the feed. Future sends will be delivered on the channel
// until the subscription is canceled. All channels added must have the same element type.
//
// The channel should have ample buffer space to avoid blocking other subscribers.
// Slow subscribers are not dropped.
func (f *Feed) Subscribe(channel interface{}) Subscription {
f.once.Do(f.init)
chanval := reflect.ValueOf(channel)
chantyp := chanval.Type()
if chantyp.Kind() != reflect.Chan || chantyp.ChanDir()&reflect.SendDir == 0 { // If the type is not a channel. Or the direction of the channel cannot send data. Then the error exits.
panic(errBadChannel)
}
sub := &feedSub{feed: f, channel: chanval, err: make(chan error, 1)}
f.mu.Lock()
defer f.mu.Unlock()
if !f.typecheck(chantyp.Elem()) {
panic(feedTypeError{op: "Subscribe", got: chantyp, want: reflect.ChanOf(reflect.SendDir, f.etype)})
}
// Add the select case to the inbox.
// The next Send will add it to f.sendCases.
cas := reflect.SelectCase{Dir: reflect.SelectSend, Chan: chanval}
f.inbox = append(f.inbox, cas)
return sub
}Send method, the Send method of the feed does not traverse all the channels and then send the blocking mode. This can result in slow clients affecting fast clients. Instead, use SelectCase in a reflective manner. First call the non-blocking TrySend to try to send. This way if there is no slow client. The data will be sent directly to completion. If the TrySend part of the client fails. Then the subsequent is sent in the form of a loop Select. I guess this is why the feed will replace the event.
// Send delivers to all subscribed channels simultaneously.
// It returns the number of subscribers that the value was sent to.
func (f *Feed) Send(value interface{}) (nsent int) {
f.once.Do(f.init)
<-f.sendLock
// Add new cases from the inbox after taking the send lock.
f.mu.Lock()
f.sendCases = append(f.sendCases, f.inbox...)
f.inbox = nil
f.mu.Unlock()
// Set the sent value on all channels.
rvalue := reflect.ValueOf(value)
if !f.typecheck(rvalue.Type()) {
f.sendLock <- struct{}{}
panic(feedTypeError{op: "Send", got: rvalue.Type(), want: f.etype})
}
for i := firstSubSendCase; i < len(f.sendCases); i++ {
f.sendCases[i].Send = rvalue
}
// Send until all channels except removeSub have been chosen.
cases := f.sendCases
for {
// Fast path: try sending without blocking before adding to the select set.
// This should usually succeed if subscribers are fast enough and have free
// buffer space.
for i := firstSubSendCase; i < len(cases); i++ {
if cases[i].Chan.TrySend(rvalue) {
nsent++
cases = cases.deactivate(i)
i--
}
}
if len(cases) == firstSubSendCase {
break
}
// Select on all the receivers, waiting for them to unblock.
chosen, recv, _ := reflect.Select(cases)
if chosen == 0 /* <-f.removeSub */ {
index := f.sendCases.find(recv.Interface())
f.sendCases = f.sendCases.delete(index)
if index >= 0 && index < len(cases) {
cases = f.sendCases[:len(cases)-1]
}
} else {
cases = cases.deactivate(chosen)
nsent++
}
}
// Forget about the sent value and hand off the send lock.
for i := firstSubSendCase; i < len(f.sendCases); i++ {
f.sendCases[i].Send = reflect.Value{}
}
f.sendLock <- struct{}{}
return nsent
}