added event bus for managing websocket events
This commit is contained in:
parent
374e6b20b6
commit
9e8aa205e4
2 changed files with 63 additions and 0 deletions
15
eventbus/bus.go
Normal file
15
eventbus/bus.go
Normal file
|
@ -0,0 +1,15 @@
|
||||||
|
package eventbus
|
||||||
|
|
||||||
|
import "github.com/bradrydzewski/drone/common"
|
||||||
|
|
||||||
|
type Event struct {
|
||||||
|
Build *common.Build
|
||||||
|
Repo *common.Repo
|
||||||
|
Task *common.Task
|
||||||
|
}
|
||||||
|
|
||||||
|
type Bus interface {
|
||||||
|
Subscribe(chan *Event)
|
||||||
|
Unsubscribe(chan *Event)
|
||||||
|
Send(*Event)
|
||||||
|
}
|
48
eventbus/eventbus.go
Normal file
48
eventbus/eventbus.go
Normal file
|
@ -0,0 +1,48 @@
|
||||||
|
package eventbus
|
||||||
|
|
||||||
|
import (
|
||||||
|
"sync"
|
||||||
|
)
|
||||||
|
|
||||||
|
type EventBus struct {
|
||||||
|
sync.Mutex
|
||||||
|
subs map[chan *Event]bool
|
||||||
|
}
|
||||||
|
|
||||||
|
// New creates a new EventBus that manages a list of
|
||||||
|
// subscribers to which events are published.
|
||||||
|
func New() *EventBus {
|
||||||
|
return &EventBus{
|
||||||
|
subs: make(map[chan *Event]bool),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Subscribe adds the channel to the list of
|
||||||
|
// subscribers. Each subscriber in the list will
|
||||||
|
// receive broadcast events.
|
||||||
|
func (b *EventBus) Subscribe(c chan *Event) {
|
||||||
|
b.Lock()
|
||||||
|
b.subs[c] = true
|
||||||
|
b.Unlock()
|
||||||
|
}
|
||||||
|
|
||||||
|
// Unsubscribe removes the channel from the
|
||||||
|
// list of subscribers.
|
||||||
|
func (b *EventBus) Unsubscribe(c chan *Event) {
|
||||||
|
b.Lock()
|
||||||
|
delete(b.subs, c)
|
||||||
|
b.Unlock()
|
||||||
|
}
|
||||||
|
|
||||||
|
// Send dispatches a message to all subscribers.
|
||||||
|
func (b *EventBus) Send(event *Event) {
|
||||||
|
b.Lock()
|
||||||
|
defer b.Unlock()
|
||||||
|
|
||||||
|
for s, _ := range b.subs {
|
||||||
|
go func(c chan *Event) {
|
||||||
|
defer recover()
|
||||||
|
c <- event
|
||||||
|
}(s)
|
||||||
|
}
|
||||||
|
}
|
Loading…
Reference in a new issue