diff --git a/core/tx_pool.go b/core/tx_pool.go index 58922f12f..ca16c1ba3 100644 --- a/core/tx_pool.go +++ b/core/tx_pool.go @@ -89,7 +89,7 @@ type TxPool struct { gasLimit func() *big.Int // The current gas limit function callback minGasPrice *big.Int eventMux *event.TypeMux - events event.Subscription + events *event.TypeMuxSubscription localTx *txSet signer types.Signer mu sync.RWMutex diff --git a/eth/filters/filter_system.go b/eth/filters/filter_system.go index e0ee2ff51..3adf8111a 100644 --- a/eth/filters/filter_system.go +++ b/eth/filters/filter_system.go @@ -74,7 +74,7 @@ type subscription struct { // subscription which match the subscription criteria. type EventSystem struct { mux *event.TypeMux - sub event.Subscription + sub *event.TypeMuxSubscription backend Backend lightMode bool lastHead *types.Header @@ -277,7 +277,7 @@ func (es *EventSystem) SubscribePendingTxEvents(hashes chan common.Hash) *Subscr type filterIndex map[Type]map[rpc.ID]*subscription // broadcast event to filters that match criteria. -func (es *EventSystem) broadcast(filters filterIndex, ev *event.Event) { +func (es *EventSystem) broadcast(filters filterIndex, ev *event.TypeMuxEvent) { if ev == nil { return } diff --git a/eth/handler.go b/eth/handler.go index 63ba0821f..691fc0677 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -78,8 +78,8 @@ type ProtocolManager struct { SubProtocols []p2p.Protocol eventMux *event.TypeMux - txSub event.Subscription - minedBlockSub event.Subscription + txSub *event.TypeMuxSubscription + minedBlockSub *event.TypeMuxSubscription // channels for fetcher, syncer, txsyncLoop newPeerCh chan *peer diff --git a/event/event.go b/event/event.go index fd0bcfbd4..f8a2eb013 100644 --- a/event/event.go +++ b/event/event.go @@ -25,33 +25,22 @@ import ( "time" ) -// Event is a time-tagged notification pushed to subscribers. -type Event struct { +// TypeMuxEvent is a time-tagged notification pushed to subscribers. +type TypeMuxEvent struct { Time time.Time Data interface{} } -// Subscription is implemented by event subscriptions. -type Subscription interface { - // Chan returns a channel that carries events. - // Implementations should return the same channel - // for any subsequent calls to Chan. - Chan() <-chan *Event - - // Unsubscribe stops delivery of events to a subscription. - // The event channel is closed. - // Unsubscribe can be called more than once. - Unsubscribe() -} - // 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][]*muxsub + subm map[reflect.Type][]*TypeMuxSubscription stopped bool } @@ -61,7 +50,7 @@ var ErrMuxClosed = errors.New("event: mux closed") // 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{}) Subscription { +func (mux *TypeMux) Subscribe(types ...interface{}) *TypeMuxSubscription { sub := newsub(mux) mux.mutex.Lock() defer mux.mutex.Unlock() @@ -72,7 +61,7 @@ func (mux *TypeMux) Subscribe(types ...interface{}) Subscription { close(sub.postC) } else { if mux.subm == nil { - mux.subm = make(map[reflect.Type][]*muxsub) + mux.subm = make(map[reflect.Type][]*TypeMuxSubscription) } for _, t := range types { rtyp := reflect.TypeOf(t) @@ -80,7 +69,7 @@ func (mux *TypeMux) Subscribe(types ...interface{}) Subscription { if find(oldsubs, sub) != -1 { panic(fmt.Sprintf("event: duplicate type %s in Subscribe", rtyp)) } - subs := make([]*muxsub, len(oldsubs)+1) + subs := make([]*TypeMuxSubscription, len(oldsubs)+1) copy(subs, oldsubs) subs[len(oldsubs)] = sub mux.subm[rtyp] = subs @@ -92,7 +81,7 @@ func (mux *TypeMux) Subscribe(types ...interface{}) Subscription { // 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 := &Event{ + event := &TypeMuxEvent{ Time: time.Now(), Data: ev, } @@ -125,7 +114,7 @@ func (mux *TypeMux) Stop() { mux.mutex.Unlock() } -func (mux *TypeMux) del(s *muxsub) { +func (mux *TypeMux) del(s *TypeMuxSubscription) { mux.mutex.Lock() for typ, subs := range mux.subm { if pos := find(subs, s); pos >= 0 { @@ -139,7 +128,7 @@ func (mux *TypeMux) del(s *muxsub) { s.mux.mutex.Unlock() } -func find(slice []*muxsub, item *muxsub) int { +func find(slice []*TypeMuxSubscription, item *TypeMuxSubscription) int { for i, v := range slice { if v == item { return i @@ -148,14 +137,15 @@ func find(slice []*muxsub, item *muxsub) int { return -1 } -func posdelete(slice []*muxsub, pos int) []*muxsub { - news := make([]*muxsub, len(slice)-1) +func posdelete(slice []*TypeMuxSubscription, pos int) []*TypeMuxSubscription { + news := make([]*TypeMuxSubscription, len(slice)-1) copy(news[:pos], slice[:pos]) copy(news[pos:], slice[pos+1:]) return news } -type muxsub struct { +// TypeMuxSubscription is a subscription established through TypeMux. +type TypeMuxSubscription struct { mux *TypeMux created time.Time closeMu sync.Mutex @@ -166,13 +156,13 @@ type muxsub struct { // postC can be set to nil without affecting the return value of // Chan. postMu sync.RWMutex - readC <-chan *Event - postC chan<- *Event + readC <-chan *TypeMuxEvent + postC chan<- *TypeMuxEvent } -func newsub(mux *TypeMux) *muxsub { - c := make(chan *Event) - return &muxsub{ +func newsub(mux *TypeMux) *TypeMuxSubscription { + c := make(chan *TypeMuxEvent) + return &TypeMuxSubscription{ mux: mux, created: time.Now(), readC: c, @@ -181,16 +171,16 @@ func newsub(mux *TypeMux) *muxsub { } } -func (s *muxsub) Chan() <-chan *Event { +func (s *TypeMuxSubscription) Chan() <-chan *TypeMuxEvent { return s.readC } -func (s *muxsub) Unsubscribe() { +func (s *TypeMuxSubscription) Unsubscribe() { s.mux.del(s) s.closewait() } -func (s *muxsub) closewait() { +func (s *TypeMuxSubscription) closewait() { s.closeMu.Lock() defer s.closeMu.Unlock() if s.closed { @@ -205,7 +195,7 @@ func (s *muxsub) closewait() { s.postMu.Unlock() } -func (s *muxsub) deliver(event *Event) { +func (s *TypeMuxSubscription) deliver(event *TypeMuxEvent) { // Short circuit delivery if stale event if s.created.After(event.Time) { return diff --git a/light/txpool.go b/light/txpool.go index d0781593b..bcdb6123d 100644 --- a/light/txpool.go +++ b/light/txpool.go @@ -47,7 +47,7 @@ type TxPool struct { signer types.Signer quit chan bool eventMux *event.TypeMux - events event.Subscription + events *event.TypeMuxSubscription mu sync.RWMutex chain *LightChain odr OdrBackend diff --git a/miner/worker.go b/miner/worker.go index 77e4e0205..49ac60253 100644 --- a/miner/worker.go +++ b/miner/worker.go @@ -90,7 +90,7 @@ type worker struct { // update loop mux *event.TypeMux - events event.Subscription + events *event.TypeMuxSubscription wg sync.WaitGroup agents map[Agent]struct{}