Skip to content

Commit

Permalink
event: deprecate TypeMux and related types
Browse files Browse the repository at this point in the history
The Subscription type is gone, all uses are replaced by
*TypeMuxSubscription. This change is prep-work for the
introduction of the new Subscription type in a later commit.

   gorename -from '"github.com/ethereum/go-ethereum/event"::Event' -to TypeMuxEvent
   gorename -from '"github.com/ethereum/go-ethereum/event"::muxsub' -to TypeMuxSubscription
   gofmt -w -r 'Subscription -> *TypeMuxSubscription' ./event/*.go
   find . -name '*.go' -and -not -regex '\./vendor/.*' \| xargs gofmt -w -r 'event.Subscription -> *event.TypeMuxSubscription'
  • Loading branch information
fjl committed Jan 25, 2017
1 parent f1069a3 commit 9b62fac
Show file tree
Hide file tree
Showing 6 changed files with 31 additions and 41 deletions.
2 changes: 1 addition & 1 deletion core/tx_pool.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
4 changes: 2 additions & 2 deletions eth/filters/filter_system.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
}
Expand Down
4 changes: 2 additions & 2 deletions eth/handler.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
58 changes: 24 additions & 34 deletions event/event.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
}

Expand All @@ -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()
Expand All @@ -72,15 +61,15 @@ 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)
oldsubs := mux.subm[rtyp]
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
Expand All @@ -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,
}
Expand Down Expand Up @@ -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 {
Expand All @@ -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
Expand All @@ -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
Expand All @@ -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,
Expand All @@ -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 {
Expand All @@ -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
Expand Down
2 changes: 1 addition & 1 deletion light/txpool.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
2 changes: 1 addition & 1 deletion miner/worker.go
Original file line number Diff line number Diff line change
Expand Up @@ -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{}
Expand Down

0 comments on commit 9b62fac

Please sign in to comment.