You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
201 lines
4.2 KiB
201 lines
4.2 KiB
package events
|
|
|
|
import (
|
|
"sync"
|
|
|
|
. "github.com/tendermint/tendermint/common"
|
|
)
|
|
|
|
// reactors and other modules should export
|
|
// this interface to become eventable
|
|
type Eventable interface {
|
|
SetFireable(Fireable)
|
|
}
|
|
|
|
// an event switch or cache implements fireable
|
|
type Fireable interface {
|
|
FireEvent(event string, msg interface{})
|
|
}
|
|
|
|
type EventSwitch struct {
|
|
BaseService
|
|
|
|
mtx sync.RWMutex
|
|
eventCells map[string]*eventCell
|
|
listeners map[string]*eventListener
|
|
}
|
|
|
|
func NewEventSwitch() *EventSwitch {
|
|
evsw := &EventSwitch{}
|
|
evsw.BaseService = *NewBaseService(log, "EventSwitch", evsw)
|
|
return evsw
|
|
}
|
|
|
|
func (evsw *EventSwitch) AfterStart() {
|
|
evsw.eventCells = make(map[string]*eventCell)
|
|
evsw.listeners = make(map[string]*eventListener)
|
|
}
|
|
|
|
func (evsw *EventSwitch) AfterStop() {
|
|
evsw.eventCells = nil
|
|
evsw.listeners = nil
|
|
}
|
|
|
|
func (evsw *EventSwitch) AddListenerForEvent(listenerId, event string, cb eventCallback) {
|
|
// Get/Create eventCell and listener
|
|
evsw.mtx.Lock()
|
|
eventCell := evsw.eventCells[event]
|
|
if eventCell == nil {
|
|
eventCell = newEventCell()
|
|
evsw.eventCells[event] = eventCell
|
|
}
|
|
listener := evsw.listeners[listenerId]
|
|
if listener == nil {
|
|
listener = newEventListener(listenerId)
|
|
evsw.listeners[listenerId] = listener
|
|
}
|
|
evsw.mtx.Unlock()
|
|
|
|
// Add event and listener
|
|
eventCell.AddListener(listenerId, cb)
|
|
listener.AddEvent(event)
|
|
}
|
|
|
|
func (evsw *EventSwitch) RemoveListener(listenerId string) {
|
|
// Get and remove listener
|
|
evsw.mtx.RLock()
|
|
listener := evsw.listeners[listenerId]
|
|
delete(evsw.listeners, listenerId)
|
|
evsw.mtx.RUnlock()
|
|
|
|
if listener == nil {
|
|
return
|
|
}
|
|
|
|
// Remove callback for each event.
|
|
listener.SetRemoved()
|
|
for _, event := range listener.GetEvents() {
|
|
evsw.RemoveListenerForEvent(event, listenerId)
|
|
}
|
|
}
|
|
|
|
func (evsw *EventSwitch) RemoveListenerForEvent(event string, listenerId string) {
|
|
// Get eventCell
|
|
evsw.mtx.Lock()
|
|
eventCell := evsw.eventCells[event]
|
|
evsw.mtx.Unlock()
|
|
|
|
if eventCell == nil {
|
|
return
|
|
}
|
|
|
|
// Remove listenerId from eventCell
|
|
numListeners := eventCell.RemoveListener(listenerId)
|
|
|
|
// Maybe garbage collect eventCell.
|
|
if numListeners == 0 {
|
|
// Lock again and double check.
|
|
evsw.mtx.Lock() // OUTER LOCK
|
|
eventCell.mtx.Lock() // INNER LOCK
|
|
if len(eventCell.listeners) == 0 {
|
|
delete(evsw.eventCells, event)
|
|
}
|
|
eventCell.mtx.Unlock() // INNER LOCK
|
|
evsw.mtx.Unlock() // OUTER LOCK
|
|
}
|
|
}
|
|
|
|
func (evsw *EventSwitch) FireEvent(event string, msg interface{}) {
|
|
// Get the eventCell
|
|
evsw.mtx.RLock()
|
|
eventCell := evsw.eventCells[event]
|
|
evsw.mtx.RUnlock()
|
|
|
|
if eventCell == nil {
|
|
return
|
|
}
|
|
|
|
// Fire event for all listeners in eventCell
|
|
eventCell.FireEvent(msg)
|
|
}
|
|
|
|
//-----------------------------------------------------------------------------
|
|
|
|
// eventCell handles keeping track of listener callbacks for a given event.
|
|
type eventCell struct {
|
|
mtx sync.RWMutex
|
|
listeners map[string]eventCallback
|
|
}
|
|
|
|
func newEventCell() *eventCell {
|
|
return &eventCell{
|
|
listeners: make(map[string]eventCallback),
|
|
}
|
|
}
|
|
|
|
func (cell *eventCell) AddListener(listenerId string, cb eventCallback) {
|
|
cell.mtx.Lock()
|
|
cell.listeners[listenerId] = cb
|
|
cell.mtx.Unlock()
|
|
}
|
|
|
|
func (cell *eventCell) RemoveListener(listenerId string) int {
|
|
cell.mtx.Lock()
|
|
delete(cell.listeners, listenerId)
|
|
numListeners := len(cell.listeners)
|
|
cell.mtx.Unlock()
|
|
return numListeners
|
|
}
|
|
|
|
func (cell *eventCell) FireEvent(msg interface{}) {
|
|
cell.mtx.RLock()
|
|
for _, listener := range cell.listeners {
|
|
listener(msg)
|
|
}
|
|
cell.mtx.RUnlock()
|
|
}
|
|
|
|
//-----------------------------------------------------------------------------
|
|
|
|
type eventCallback func(msg interface{})
|
|
|
|
type eventListener struct {
|
|
id string
|
|
|
|
mtx sync.RWMutex
|
|
removed bool
|
|
events []string
|
|
}
|
|
|
|
func newEventListener(id string) *eventListener {
|
|
return &eventListener{
|
|
id: id,
|
|
removed: false,
|
|
events: nil,
|
|
}
|
|
}
|
|
|
|
func (evl *eventListener) AddEvent(event string) {
|
|
evl.mtx.Lock()
|
|
defer evl.mtx.Unlock()
|
|
|
|
if evl.removed {
|
|
return
|
|
}
|
|
evl.events = append(evl.events, event)
|
|
}
|
|
|
|
func (evl *eventListener) GetEvents() []string {
|
|
evl.mtx.RLock()
|
|
defer evl.mtx.RUnlock()
|
|
|
|
events := make([]string, len(evl.events))
|
|
copy(events, evl.events)
|
|
return events
|
|
}
|
|
|
|
func (evl *eventListener) SetRemoved() {
|
|
evl.mtx.Lock()
|
|
defer evl.mtx.Unlock()
|
|
evl.removed = true
|
|
}
|