vendor/github.com/tendermint/tmlibs/events/events.go

   1 /*
   2 Pub-Sub in go with event caching
   3 */
   4 package events
   5
   6 import (
   7         "sync"
   8
   9         . "github.com/tendermint/tmlibs/common"
  10 )
  11
  12 // Generic event data can be typed and registered with tendermint/go-wire
  13 // via concrete implementation of this interface
  14 type EventData interface {
  15         //AssertIsEventData()
  16 }
  17
  18 // reactors and other modules should export
  19 // this interface to become eventable
  20 type Eventable interface {
  21         SetEventSwitch(evsw EventSwitch)
  22 }
  23
  24 // an event switch or cache implements fireable
  25 type Fireable interface {
  26         FireEvent(event string, data EventData)
  27 }
  28
  29 type EventSwitch interface {
  30         Service
  31         Fireable
  32
  33         AddListenerForEvent(listenerID, event string, cb EventCallback)
  34         RemoveListenerForEvent(event string, listenerID string)
  35         RemoveListener(listenerID string)
  36 }
  37
  38 type eventSwitch struct {
  39         BaseService
  40
  41         mtx        sync.RWMutex
  42         eventCells map[string]*eventCell
  43         listeners  map[string]*eventListener
  44 }
  45
  46 func NewEventSwitch() EventSwitch {
  47         evsw := &eventSwitch{}
  48         evsw.BaseService = *NewBaseService(nil, "EventSwitch", evsw)
  49         return evsw
  50 }
  51
  52 func (evsw *eventSwitch) OnStart() error {
  53         evsw.BaseService.OnStart()
  54         evsw.eventCells = make(map[string]*eventCell)
  55         evsw.listeners = make(map[string]*eventListener)
  56         return nil
  57 }
  58
  59 func (evsw *eventSwitch) OnStop() {
  60         evsw.mtx.Lock()
  61         defer evsw.mtx.Unlock()
  62         evsw.BaseService.OnStop()
  63         evsw.eventCells = nil
  64         evsw.listeners = nil
  65 }
  66
  67 func (evsw *eventSwitch) AddListenerForEvent(listenerID, event string, cb EventCallback) {
  68         // Get/Create eventCell and listener
  69         evsw.mtx.Lock()
  70         eventCell := evsw.eventCells[event]
  71         if eventCell == nil {
  72                 eventCell = newEventCell()
  73                 evsw.eventCells[event] = eventCell
  74         }
  75         listener := evsw.listeners[listenerID]
  76         if listener == nil {
  77                 listener = newEventListener(listenerID)
  78                 evsw.listeners[listenerID] = listener
  79         }
  80         evsw.mtx.Unlock()
  81
  82         // Add event and listener
  83         eventCell.AddListener(listenerID, cb)
  84         listener.AddEvent(event)
  85 }
  86
  87 func (evsw *eventSwitch) RemoveListener(listenerID string) {
  88         // Get and remove listener
  89         evsw.mtx.RLock()
  90         listener := evsw.listeners[listenerID]
  91         evsw.mtx.RUnlock()
  92         if listener == nil {
  93                 return
  94         }
  95
  96         evsw.mtx.Lock()
  97         delete(evsw.listeners, listenerID)
  98         evsw.mtx.Unlock()
  99
 100         // Remove callback for each event.
 101         listener.SetRemoved()
 102         for _, event := range listener.GetEvents() {
 103                 evsw.RemoveListenerForEvent(event, listenerID)
 104         }
 105 }
 106
 107 func (evsw *eventSwitch) RemoveListenerForEvent(event string, listenerID string) {
 108         // Get eventCell
 109         evsw.mtx.Lock()
 110         eventCell := evsw.eventCells[event]
 111         evsw.mtx.Unlock()
 112
 113         if eventCell == nil {
 114                 return
 115         }
 116
 117         // Remove listenerID from eventCell
 118         numListeners := eventCell.RemoveListener(listenerID)
 119
 120         // Maybe garbage collect eventCell.
 121         if numListeners == 0 {
 122                 // Lock again and double check.
 123                 evsw.mtx.Lock()      // OUTER LOCK
 124                 eventCell.mtx.Lock() // INNER LOCK
 125                 if len(eventCell.listeners) == 0 {
 126                         delete(evsw.eventCells, event)
 127                 }
 128                 eventCell.mtx.Unlock() // INNER LOCK
 129                 evsw.mtx.Unlock()      // OUTER LOCK
 130         }
 131 }
 132
 133 func (evsw *eventSwitch) FireEvent(event string, data EventData) {
 134         // Get the eventCell
 135         evsw.mtx.RLock()
 136         eventCell := evsw.eventCells[event]
 137         evsw.mtx.RUnlock()
 138
 139         if eventCell == nil {
 140                 return
 141         }
 142
 143         // Fire event for all listeners in eventCell
 144         eventCell.FireEvent(data)
 145 }
 146
 147 //-----------------------------------------------------------------------------
 148
 149 // eventCell handles keeping track of listener callbacks for a given event.
 150 type eventCell struct {
 151         mtx       sync.RWMutex
 152         listeners map[string]EventCallback
 153 }
 154
 155 func newEventCell() *eventCell {
 156         return &eventCell{
 157                 listeners: make(map[string]EventCallback),
 158         }
 159 }
 160
 161 func (cell *eventCell) AddListener(listenerID string, cb EventCallback) {
 162         cell.mtx.Lock()
 163         cell.listeners[listenerID] = cb
 164         cell.mtx.Unlock()
 165 }
 166
 167 func (cell *eventCell) RemoveListener(listenerID string) int {
 168         cell.mtx.Lock()
 169         delete(cell.listeners, listenerID)
 170         numListeners := len(cell.listeners)
 171         cell.mtx.Unlock()
 172         return numListeners
 173 }
 174
 175 func (cell *eventCell) FireEvent(data EventData) {
 176         cell.mtx.RLock()
 177         for _, listener := range cell.listeners {
 178                 listener(data)
 179         }
 180         cell.mtx.RUnlock()
 181 }
 182
 183 //-----------------------------------------------------------------------------
 184
 185 type EventCallback func(data EventData)
 186
 187 type eventListener struct {
 188         id string
 189
 190         mtx     sync.RWMutex
 191         removed bool
 192         events  []string
 193 }
 194
 195 func newEventListener(id string) *eventListener {
 196         return &eventListener{
 197                 id:      id,
 198                 removed: false,
 199                 events:  nil,
 200         }
 201 }
 202
 203 func (evl *eventListener) AddEvent(event string) {
 204         evl.mtx.Lock()
 205         defer evl.mtx.Unlock()
 206
 207         if evl.removed {
 208                 return
 209         }
 210         evl.events = append(evl.events, event)
 211 }
 212
 213 func (evl *eventListener) GetEvents() []string {
 214         evl.mtx.RLock()
 215         defer evl.mtx.RUnlock()
 216
 217         events := make([]string, len(evl.events))
 218         copy(events, evl.events)
 219         return events
 220 }
 221
 222 func (evl *eventListener) SetRemoved() {
 223         evl.mtx.Lock()
 224         defer evl.mtx.Unlock()
 225         evl.removed = true
 226 }