以太坊源碼分析(38）event源碼分析

    // 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 和 postC 實際上是同一個channel。 不過一個是從channel讀 一個只從channel寫

        // 單方向的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{}),

        }

    }

發佈一個event到TypeMux上面，這個時候全部訂閱了這個類型的都會收到這個消息。

    // 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 {

            // 阻塞式的投遞。

            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 {    //阻塞方式的方法

        case s.postC <- event:

        case <-s.closing:

        }

    }

## feed.go

目前主要使用的對象。取代了前面說的event.go內部的TypeMux

feed數據結構

    // 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.

    // Feed 實現了 1對多的訂閱模式，使用了channel來傳遞事件。 發送給Feed的值會同時被傳遞給全部訂閱的channel。

    // 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.

    // Feed只能被單個類型使用。這個和以前的event不一樣，event能夠使用多個類型。 類型被第一個Send調用或者是Subscribe調用決定。 後續的調用若是類型和其不一致會panic

    // 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

    }

初始化 初始化會被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}}

    }

訂閱，訂閱投遞了一個channel。 相對與event的不一樣。event的訂閱是傳入了須要訂閱的類型，而後channel是在event的訂閱代碼裏面構建而後返回的。 這種直接投遞channel的模式可能會更加靈活。

而後根據傳入的channel生成了SelectCase。放入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 { // 若是類型不是channel。 或者是channel的方向不能發送數據。那麼錯誤退出。

            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方法,feed的Send方法不是遍歷全部的channel而後阻塞方式的發送。這樣可能致使慢的客戶端影響快的客戶端。 而是使用反射的方式使用SelectCase。 首先調用非阻塞方式的TrySend來嘗試發送。這樣若是沒有慢的客戶端。數據會直接所有發送完成。 若是TrySend部分客戶端失敗。 那麼後續在循環Select的方式發送。 我猜想這也是feed會取代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

    }