go/ferret
go/ferret
1
0
Fork 0
mirror of https://github.com/MontFerret/ferret.git synced 2025-02-05 13:15:20 +02:00
Code Issues Releases Activity

Some tweaks in EventBroker

Browse Source
This commit is contained in:
Tim Voronov 2018-10-15 17:48:30 -04:00
parent 1d6a23fa96
commit 18b76be3fc
2 changed files with 21 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
pkg/html/dynamic/events/broker.go
View File

@ -184,7 +184,7 @@ func (broker *EventBroker) runLoop(ctx context.Context) {
case <-broker.onLoad.Ready(): case <-broker.onLoad.Ready():
reply, err := broker.onLoad.Recv() reply, err := broker.onLoad.Recv()
broker.emit(EventLoad, reply, err) broker.emit(ctx, EventLoad, reply, err)
default: default:
} }
@ -194,7 +194,7 @@ func (broker *EventBroker) runLoop(ctx context.Context) {
case <-broker.onReload.Ready(): case <-broker.onReload.Ready():
reply, err := broker.onReload.Recv() reply, err := broker.onReload.Recv()
broker.emit(EventReload, reply, err) broker.emit(ctx, EventReload, reply, err)
default: default:
} }
@ -204,7 +204,7 @@ func (broker *EventBroker) runLoop(ctx context.Context) {
case <-broker.onAttrModified.Ready(): case <-broker.onAttrModified.Ready():
reply, err := broker.onAttrModified.Recv() reply, err := broker.onAttrModified.Recv()
broker.emit(EventAttrModified, reply, err) broker.emit(ctx, EventAttrModified, reply, err)
default: default:
} }
@ -214,7 +214,7 @@ func (broker *EventBroker) runLoop(ctx context.Context) {
case <-broker.onAttrRemoved.Ready(): case <-broker.onAttrRemoved.Ready():
reply, err := broker.onAttrRemoved.Recv() reply, err := broker.onAttrRemoved.Recv()
broker.emit(EventAttrRemoved, reply, err) broker.emit(ctx, EventAttrRemoved, reply, err)
default: default:
} }
@ -224,7 +224,7 @@ func (broker *EventBroker) runLoop(ctx context.Context) {
case <-broker.onChildNodeCountUpdated.Ready(): case <-broker.onChildNodeCountUpdated.Ready():
reply, err := broker.onChildNodeCountUpdated.Recv() reply, err := broker.onChildNodeCountUpdated.Recv()
broker.emit(EventChildNodeCountUpdated, reply, err) broker.emit(ctx, EventChildNodeCountUpdated, reply, err)
default: default:
} }
@ -235,7 +235,7 @@ func (broker *EventBroker) runLoop(ctx context.Context) {
case <-broker.onChildNodeInserted.Ready(): case <-broker.onChildNodeInserted.Ready():
reply, err := broker.onChildNodeInserted.Recv() reply, err := broker.onChildNodeInserted.Recv()
broker.emit(EventChildNodeInserted, reply, err) broker.emit(ctx, EventChildNodeInserted, reply, err)
default: default:
} }
@ -245,13 +245,13 @@ func (broker *EventBroker) runLoop(ctx context.Context) {
case <-broker.onChildNodeRemoved.Ready(): case <-broker.onChildNodeRemoved.Ready():
reply, err := broker.onChildNodeRemoved.Recv() reply, err := broker.onChildNodeRemoved.Recv()
broker.emit(EventChildNodeRemoved, reply, err) broker.emit(ctx, EventChildNodeRemoved, reply, err)
default: default:
} }
} }
} }
func (broker *EventBroker) emit(event Event, message interface{}, err error) { func (broker *EventBroker) emit(ctx context.Context, event Event, message interface{}, err error) {
if err != nil { if err != nil {
event = EventError event = EventError
message = err message = err
@ -271,7 +271,12 @@ func (broker *EventBroker) emit(event Event, message interface{}, err error) {
go func() { go func() {
for _, listener := range snapshot { for _, listener := range snapshot {
select {
case <-ctx.Done():
return
default:
listener(message) listener(message)
} }
}
}() }()
} }

11
pkg/html/dynamic/events/broker_test.go
View File

@ -263,15 +263,19 @@ func TestEventBroker(t *testing.T) {
defer b.Stop() defer b.Stop()
counter := 0 counter := 0
listener := func(message interface{}) {
var listener events.EventListener
listener = func(message interface{}) {
counter += 1 counter += 1
b.RemoveEventListener(events.EventLoad, listener)
} }
b.AddEventListener(events.EventLoad, listener) b.AddEventListener(events.EventLoad, listener)
b.OnLoad.Emit(&page.LoadEventFiredReply{}) b.OnLoad.Emit(&page.LoadEventFiredReply{})
time.Sleep(time.Duration(10) * time.Millisecond) time.Sleep(time.Duration(10) * time.Millisecond)
b.RemoveEventListener(events.EventLoad, listener)
StressTestAsync(func() error { StressTestAsync(func() error {
b.OnLoad.Emit(&page.LoadEventFiredReply{}) b.OnLoad.Emit(&page.LoadEventFiredReply{})
@ -292,14 +296,13 @@ func TestEventBroker(t *testing.T) {
counter := 0 counter := 0
b.AddEventListener(events.EventLoad, func(message interface{}) { b.AddEventListener(events.EventLoad, func(message interface{}) {
counter++ counter++
b.Stop()
}) })
b.OnLoad.EmitDefault() b.OnLoad.EmitDefault()
time.Sleep(time.Duration(5) * time.Millisecond) time.Sleep(time.Duration(5) * time.Millisecond)
b.Stop()
go func() { go func() {
b.OnLoad.EmitDefault() b.OnLoad.EmitDefault()
}() }()