Skip to content

Commit

Permalink
Revert "event: move type fixation logic into Feed.init (ethereum#27249)"
Browse files Browse the repository at this point in the history
This reverts commit b2a95a2.
  • Loading branch information
devopsbo3 authored Nov 10, 2023
1 parent 7cc5d3c commit 7410a9c
Showing 1 changed file with 22 additions and 12 deletions.
34 changes: 22 additions & 12 deletions event/feed.go
Original file line number Diff line number Diff line change
Expand Up @@ -57,8 +57,7 @@ func (e feedTypeError) Error() string {
return "event: wrong type in " + e.op + " got " + e.got.String() + ", want " + e.want.String()
}

func (f *Feed) init(etype reflect.Type) {
f.etype = etype
func (f *Feed) init() {
f.removeSub = make(chan interface{})
f.sendLock = make(chan struct{}, 1)
f.sendLock <- struct{}{}
Expand All @@ -71,27 +70,36 @@ func (f *Feed) init(etype reflect.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 {
panic(errBadChannel)
}
sub := &feedSub{feed: f, channel: chanval, err: make(chan error, 1)}

f.once.Do(func() { f.init(chantyp.Elem()) })
if f.etype != chantyp.Elem() {
panic(feedTypeError{op: "Subscribe", got: chantyp, want: reflect.ChanOf(reflect.SendDir, f.etype)})
}

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
}

// note: callers must hold f.mu
func (f *Feed) typecheck(typ reflect.Type) bool {
if f.etype == nil {
f.etype = typ
return true
}
return f.etype == typ
}

func (f *Feed) remove(sub *feedSub) {
// Delete from inbox first, which covers channels
// that have not been added to f.sendCases yet.
Expand Down Expand Up @@ -120,17 +128,19 @@ func (f *Feed) remove(sub *feedSub) {
func (f *Feed) Send(value interface{}) (nsent int) {
rvalue := reflect.ValueOf(value)

f.once.Do(func() { f.init(rvalue.Type()) })
if f.etype != rvalue.Type() {
panic(feedTypeError{op: "Send", got: rvalue.Type(), want: f.etype})
}

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

if !f.typecheck(rvalue.Type()) {
f.sendLock <- struct{}{}
f.mu.Unlock()
panic(feedTypeError{op: "Send", got: rvalue.Type(), want: f.etype})
}
f.mu.Unlock()

// Set the sent value on all channels.
Expand Down

0 comments on commit 7410a9c

Please sign in to comment.