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.
 
 
 
 
 
 

215 lines
4.7 KiB

package events
import (
"sync"
. "github.com/tendermint/go-common"
"github.com/tendermint/tendermint/types"
)
// reactors and other modules should export
// this interface to become eventable
type Eventable interface {
SetEventSwitch(evsw *EventSwitch)
}
// an event switch or cache implements fireable
type Fireable interface {
FireEvent(event string, data types.EventData)
}
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) OnStart() error {
evsw.BaseService.OnStart()
evsw.eventCells = make(map[string]*eventCell)
evsw.listeners = make(map[string]*eventListener)
return nil
}
func (evsw *EventSwitch) OnStop() {
evsw.BaseService.OnStop()
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, data types.EventData) {
// 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(data)
}
func (evsw *EventSwitch) SubscribeToEvent(receiver, eventID string, chanCap int) chan interface{} {
// listen for new round
ch := make(chan interface{}, chanCap)
evsw.AddListenerForEvent(receiver, eventID, func(data types.EventData) {
// NOTE: in production, evsw callbacks should be nonblocking.
ch <- data
})
return ch
}
//-----------------------------------------------------------------------------
// 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(data types.EventData) {
cell.mtx.RLock()
for _, listener := range cell.listeners {
listener(data)
}
cell.mtx.RUnlock()
}
//-----------------------------------------------------------------------------
type eventCallback func(data types.EventData)
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
}