mirror of
https://github.com/MHSanaei/3x-ui.git
synced 2026-07-15 17:16:07 +00:00
fix(eventbus): dispatch each subscriber in its own goroutine
The fan-out loop called every subscriber's handler sequentially on the single dispatch goroutine. The email and Telegram notifiers block on network I/O for tens of seconds (or minutes when the remote is slow), so one slow subscriber stalled the whole loop: the 256-slot channel then filled and Publish silently dropped later events — including high-value xray.crash and node.down notifications unrelated to the slow handler. Hand each delivered event to every handler in its own goroutine so a blocking subscriber can no longer stall delivery to the others. safeCall already recovers panics, so a detached handler cannot take down the bus.
This commit is contained in:
@@ -81,9 +81,11 @@ func (b *Bus) Publish(e Event) {
|
||||
}
|
||||
}
|
||||
|
||||
// dispatch is the fan-out loop. It reads events from the channel and calls
|
||||
// every subscriber's handler sequentially. Handlers run on the dispatch
|
||||
// goroutine — they must not block.
|
||||
// dispatch is the fan-out loop. It reads events from the channel and hands each
|
||||
// one to every subscriber's handler in its own goroutine, so a subscriber whose
|
||||
// handler blocks on network I/O (the email and Telegram notifiers can block for
|
||||
// tens of seconds) cannot stall delivery of unrelated, higher-value events such
|
||||
// as xray.crash or node.down.
|
||||
func (b *Bus) dispatch() {
|
||||
defer b.wg.Done()
|
||||
for {
|
||||
@@ -97,7 +99,7 @@ func (b *Bus) dispatch() {
|
||||
copy(subs, b.subs)
|
||||
b.mu.RUnlock()
|
||||
for _, s := range subs {
|
||||
safeCall(s.handler, e)
|
||||
go safeCall(s.handler, e)
|
||||
}
|
||||
case <-b.done:
|
||||
return
|
||||
@@ -115,8 +117,9 @@ func safeCall(fn func(Event), e Event) {
|
||||
fn(e)
|
||||
}
|
||||
|
||||
// Stop shuts down the bus: the dispatch goroutine exits, in-flight handlers
|
||||
// finish, and any events still buffered may be dropped. Safe to call once.
|
||||
// Stop shuts down the bus: the dispatch goroutine exits and any events still
|
||||
// buffered may be dropped. Handler goroutines already spawned for delivered
|
||||
// events run to completion on their own. Safe to call once.
|
||||
func (b *Bus) Stop() {
|
||||
close(b.done)
|
||||
b.wg.Wait()
|
||||
|
||||
@@ -149,6 +149,31 @@ func TestBusPanicRecovery(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestBusBlockingSubscriberDoesNotStallOthers(t *testing.T) {
|
||||
b := New(16)
|
||||
defer b.Stop()
|
||||
|
||||
release := make(chan struct{})
|
||||
b.Subscribe("blocking", func(e Event) {
|
||||
<-release
|
||||
})
|
||||
|
||||
fast := make(chan struct{}, 1)
|
||||
b.Subscribe("fast", func(e Event) {
|
||||
fast <- struct{}{}
|
||||
})
|
||||
|
||||
b.Publish(Event{Type: EventXrayCrash})
|
||||
|
||||
select {
|
||||
case <-fast:
|
||||
case <-time.After(time.Second):
|
||||
close(release)
|
||||
t.Fatal("a blocking subscriber stalled event delivery to another subscriber")
|
||||
}
|
||||
close(release)
|
||||
}
|
||||
|
||||
func TestBusBufferFull(t *testing.T) {
|
||||
b := New(2)
|
||||
defer b.Stop()
|
||||
|
||||
Reference in New Issue
Block a user