mirrors/tendermint
1
0
Fork 0
mirror of https://github.com/tendermint/tendermint.git synced 2026-09-29 19:25:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

detele everything

Browse Source
This commit is contained in:
Zach Ramsay
2018-07-10 11:12:46 -04:00
parent 9d81a74429
commit d02c5d1e30
721 changed files with 0 additions and 99035 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libs/events/Makefile
-9
View File
@@ -1,9 +0,0 @@
.PHONY: docs
REPO:=github.com/tendermint/tendermint/libs/events
docs:
@go get github.com/davecheney/godoc2md
godoc2md $(REPO) > README.md
test:
go test -v ./...
libs/events/README.md
-175
View File
@@ -1,175 +0,0 @@
# events
`import "github.com/tendermint/tendermint/libs/events"`
* [Overview](#pkg-overview)
* [Index](#pkg-index)
## <a name="pkg-overview">Overview</a>
Pub-Sub in go with event caching
## <a name="pkg-index">Index</a>
* [type EventCache](#EventCache)
* [func NewEventCache(evsw Fireable) *EventCache](#NewEventCache)
* [func (evc *EventCache) FireEvent(event string, data EventData)](#EventCache.FireEvent)
* [func (evc *EventCache) Flush()](#EventCache.Flush)
* [type EventCallback](#EventCallback)
* [type EventData](#EventData)
* [type EventSwitch](#EventSwitch)
* [func NewEventSwitch() EventSwitch](#NewEventSwitch)
* [type Eventable](#Eventable)
* [type Fireable](#Fireable)
#### <a name="pkg-files">Package files</a>
[event_cache.go](/src/github.com/tendermint/tendermint/libs/events/event_cache.go) [events.go](/src/github.com/tendermint/tendermint/libs/events/events.go)
## <a name="EventCache">type</a> [EventCache](/src/target/event_cache.go?s=116:179#L5)
``` go
type EventCache struct {
// contains filtered or unexported fields
}
```
An EventCache buffers events for a Fireable
All events are cached. Filtering happens on Flush
### <a name="NewEventCache">func</a> [NewEventCache](/src/target/event_cache.go?s=239:284#L11)
``` go
func NewEventCache(evsw Fireable) *EventCache
```
Create a new EventCache with an EventSwitch as backend
### <a name="EventCache.FireEvent">func</a> (\*EventCache) [FireEvent](/src/target/event_cache.go?s=449:511#L24)
``` go
func (evc *EventCache) FireEvent(event string, data EventData)
```
Cache an event to be fired upon finality.
### <a name="EventCache.Flush">func</a> (\*EventCache) [Flush](/src/target/event_cache.go?s=735:765#L31)
``` go
func (evc *EventCache) Flush()
```
Fire events by running evsw.FireEvent on all cached events. Blocks.
Clears cached events
## <a name="EventCallback">type</a> [EventCallback](/src/target/events.go?s=4201:4240#L185)
``` go
type EventCallback func(data EventData)
```
## <a name="EventData">type</a> [EventData](/src/target/events.go?s=243:294#L14)
``` go
type EventData interface {
}
```
Generic event data can be typed and registered with tendermint/go-amino
via concrete implementation of this interface
## <a name="EventSwitch">type</a> [EventSwitch](/src/target/events.go?s=560:771#L29)
``` go
type EventSwitch interface {
cmn.Service
Fireable
AddListenerForEvent(listenerID, event string, cb EventCallback)
RemoveListenerForEvent(event string, listenerID string)
RemoveListener(listenerID string)
}
```
### <a name="NewEventSwitch">func</a> [NewEventSwitch](/src/target/events.go?s=917:950#L46)
``` go
func NewEventSwitch() EventSwitch
```
## <a name="Eventable">type</a> [Eventable](/src/target/events.go?s=378:440#L20)
``` go
type Eventable interface {
SetEventSwitch(evsw EventSwitch)
}
```
reactors and other modules should export
this interface to become eventable
## <a name="Fireable">type</a> [Fireable](/src/target/events.go?s=490:558#L25)
``` go
type Fireable interface {
FireEvent(event string, data EventData)
}
```
an event switch or cache implements fireable
- - -
Generated by [godoc2md](http://godoc.org/github.com/davecheney/godoc2md)
libs/events/event_cache.go
-37
View File
@@ -1,37 +0,0 @@
package events
// An EventCache buffers events for a Fireable
// All events are cached. Filtering happens on Flush
type EventCache struct {
evsw Fireable
events []eventInfo
}
// Create a new EventCache with an EventSwitch as backend
func NewEventCache(evsw Fireable) *EventCache {
return &EventCache{
evsw: evsw,
}
}
// a cached event
type eventInfo struct {
event string
data EventData
}
// Cache an event to be fired upon finality.
func (evc *EventCache) FireEvent(event string, data EventData) {
// append to list (go will grow our backing array exponentially)
evc.events = append(evc.events, eventInfo{event, data})
}
// Fire events by running evsw.FireEvent on all cached events. Blocks.
// Clears cached events
func (evc *EventCache) Flush() {
for _, ei := range evc.events {
evc.evsw.FireEvent(ei.event, ei.data)
}
// Clear the buffer, since we only add to it with append it's safe to just set it to nil and maybe safe an allocation
evc.events = nil
}
libs/events/event_cache_test.go
-35
View File
@@ -1,35 +0,0 @@
package events
import (
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func TestEventCache_Flush(t *testing.T) {
evsw := NewEventSwitch()
evsw.Start()
evsw.AddListenerForEvent("nothingness", "", func(data EventData) {
// Check we are not initialising an empty buffer full of zeroed eventInfos in the EventCache
require.FailNow(t, "We should never receive a message on this switch since none are fired")
})
evc := NewEventCache(evsw)
evc.Flush()
// Check after reset
evc.Flush()
fail := true
pass := false
evsw.AddListenerForEvent("somethingness", "something", func(data EventData) {
if fail {
require.FailNow(t, "Shouldn't see a message until flushed")
}
pass = true
})
evc.FireEvent("something", struct{ int }{1})
evc.FireEvent("something", struct{ int }{2})
evc.FireEvent("something", struct{ int }{3})
fail = false
evc.Flush()
assert.True(t, pass)
}
libs/events/events.go
-220
View File
@@ -1,220 +0,0 @@
/*
Pub-Sub in go with event caching
*/
package events
import (
"sync"
cmn "github.com/tendermint/tendermint/libs/common"
)
// Generic event data can be typed and registered with tendermint/go-amino
// via concrete implementation of this interface
type EventData interface {
//AssertIsEventData()
}
// 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 EventData)
}
type EventSwitch interface {
cmn.Service
Fireable
AddListenerForEvent(listenerID, event string, cb EventCallback)
RemoveListenerForEvent(event string, listenerID string)
RemoveListener(listenerID string)
}
type eventSwitch struct {
cmn.BaseService
mtx sync.RWMutex
eventCells map[string]*eventCell
listeners map[string]*eventListener
}
func NewEventSwitch() EventSwitch {
evsw := &eventSwitch{
eventCells: make(map[string]*eventCell),
listeners: make(map[string]*eventListener),
}
evsw.BaseService = *cmn.NewBaseService(nil, "EventSwitch", evsw)
return evsw
}
func (evsw *eventSwitch) OnStart() error {
return nil
}
func (evsw *eventSwitch) OnStop() {}
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]
evsw.mtx.RUnlock()
if listener == nil {
return
}
evsw.mtx.Lock()
delete(evsw.listeners, listenerID)
evsw.mtx.Unlock()
// 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 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)
}
//-----------------------------------------------------------------------------
// 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 EventData) {
cell.mtx.RLock()
for _, listener := range cell.listeners {
listener(data)
}
cell.mtx.RUnlock()
}
//-----------------------------------------------------------------------------
type EventCallback func(data 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
}
libs/events/events_test.go
-380
View File
@@ -1,380 +0,0 @@
package events
import (
"fmt"
"math/rand"
"testing"
"time"
"github.com/stretchr/testify/assert"
)
// TestAddListenerForEventFireOnce sets up an EventSwitch, subscribes a single
// listener to an event, and sends a string "data".
func TestAddListenerForEventFireOnce(t *testing.T) {
evsw := NewEventSwitch()
err := evsw.Start()
if err != nil {
t.Errorf("Failed to start EventSwitch, error: %v", err)
}
messages := make(chan EventData)
evsw.AddListenerForEvent("listener", "event",
func(data EventData) {
messages <- data
})
go evsw.FireEvent("event", "data")
received := <-messages
if received != "data" {
t.Errorf("Message received does not match: %v", received)
}
}
// TestAddListenerForEventFireMany sets up an EventSwitch, subscribes a single
// listener to an event, and sends a thousand integers.
func TestAddListenerForEventFireMany(t *testing.T) {
evsw := NewEventSwitch()
err := evsw.Start()
if err != nil {
t.Errorf("Failed to start EventSwitch, error: %v", err)
}
doneSum := make(chan uint64)
doneSending := make(chan uint64)
numbers := make(chan uint64, 4)
// subscribe one listener for one event
evsw.AddListenerForEvent("listener", "event",
func(data EventData) {
numbers <- data.(uint64)
})
// collect received events
go sumReceivedNumbers(numbers, doneSum)
// go fire events
go fireEvents(evsw, "event", doneSending, uint64(1))
checkSum := <-doneSending
close(numbers)
eventSum := <-doneSum
if checkSum != eventSum {
t.Errorf("Not all messages sent were received.\n")
}
}
// TestAddListenerForDifferentEvents sets up an EventSwitch, subscribes a single
// listener to three different events and sends a thousand integers for each
// of the three events.
func TestAddListenerForDifferentEvents(t *testing.T) {
evsw := NewEventSwitch()
err := evsw.Start()
if err != nil {
t.Errorf("Failed to start EventSwitch, error: %v", err)
}
doneSum := make(chan uint64)
doneSending1 := make(chan uint64)
doneSending2 := make(chan uint64)
doneSending3 := make(chan uint64)
numbers := make(chan uint64, 4)
// subscribe one listener to three events
evsw.AddListenerForEvent("listener", "event1",
func(data EventData) {
numbers <- data.(uint64)
})
evsw.AddListenerForEvent("listener", "event2",
func(data EventData) {
numbers <- data.(uint64)
})
evsw.AddListenerForEvent("listener", "event3",
func(data EventData) {
numbers <- data.(uint64)
})
// collect received events
go sumReceivedNumbers(numbers, doneSum)
// go fire events
go fireEvents(evsw, "event1", doneSending1, uint64(1))
go fireEvents(evsw, "event2", doneSending2, uint64(1))
go fireEvents(evsw, "event3", doneSending3, uint64(1))
var checkSum uint64 = 0
checkSum += <-doneSending1
checkSum += <-doneSending2
checkSum += <-doneSending3
close(numbers)
eventSum := <-doneSum
if checkSum != eventSum {
t.Errorf("Not all messages sent were received.\n")
}
}
// TestAddDifferentListenerForDifferentEvents sets up an EventSwitch,
// subscribes a first listener to three events, and subscribes a second
// listener to two of those three events, and then sends a thousand integers
// for each of the three events.
func TestAddDifferentListenerForDifferentEvents(t *testing.T) {
evsw := NewEventSwitch()
err := evsw.Start()
if err != nil {
t.Errorf("Failed to start EventSwitch, error: %v", err)
}
doneSum1 := make(chan uint64)
doneSum2 := make(chan uint64)
doneSending1 := make(chan uint64)
doneSending2 := make(chan uint64)
doneSending3 := make(chan uint64)
numbers1 := make(chan uint64, 4)
numbers2 := make(chan uint64, 4)
// subscribe two listener to three events
evsw.AddListenerForEvent("listener1", "event1",
func(data EventData) {
numbers1 <- data.(uint64)
})
evsw.AddListenerForEvent("listener1", "event2",
func(data EventData) {
numbers1 <- data.(uint64)
})
evsw.AddListenerForEvent("listener1", "event3",
func(data EventData) {
numbers1 <- data.(uint64)
})
evsw.AddListenerForEvent("listener2", "event2",
func(data EventData) {
numbers2 <- data.(uint64)
})
evsw.AddListenerForEvent("listener2", "event3",
func(data EventData) {
numbers2 <- data.(uint64)
})
// collect received events for listener1
go sumReceivedNumbers(numbers1, doneSum1)
// collect received events for listener2
go sumReceivedNumbers(numbers2, doneSum2)
// go fire events
go fireEvents(evsw, "event1", doneSending1, uint64(1))
go fireEvents(evsw, "event2", doneSending2, uint64(1001))
go fireEvents(evsw, "event3", doneSending3, uint64(2001))
checkSumEvent1 := <-doneSending1
checkSumEvent2 := <-doneSending2
checkSumEvent3 := <-doneSending3
checkSum1 := checkSumEvent1 + checkSumEvent2 + checkSumEvent3
checkSum2 := checkSumEvent2 + checkSumEvent3
close(numbers1)
close(numbers2)
eventSum1 := <-doneSum1
eventSum2 := <-doneSum2
if checkSum1 != eventSum1 ||
checkSum2 != eventSum2 {
t.Errorf("Not all messages sent were received for different listeners to different events.\n")
}
}
// TestAddAndRemoveListener sets up an EventSwitch, subscribes a listener to
// two events, fires a thousand integers for the first event, then unsubscribes
// the listener and fires a thousand integers for the second event.
func TestAddAndRemoveListener(t *testing.T) {
evsw := NewEventSwitch()
err := evsw.Start()
if err != nil {
t.Errorf("Failed to start EventSwitch, error: %v", err)
}
doneSum1 := make(chan uint64)
doneSum2 := make(chan uint64)
doneSending1 := make(chan uint64)
doneSending2 := make(chan uint64)
numbers1 := make(chan uint64, 4)
numbers2 := make(chan uint64, 4)
// subscribe two listener to three events
evsw.AddListenerForEvent("listener", "event1",
func(data EventData) {
numbers1 <- data.(uint64)
})
evsw.AddListenerForEvent("listener", "event2",
func(data EventData) {
numbers2 <- data.(uint64)
})
// collect received events for event1
go sumReceivedNumbers(numbers1, doneSum1)
// collect received events for event2
go sumReceivedNumbers(numbers2, doneSum2)
// go fire events
go fireEvents(evsw, "event1", doneSending1, uint64(1))
checkSumEvent1 := <-doneSending1
// after sending all event1, unsubscribe for all events
evsw.RemoveListener("listener")
go fireEvents(evsw, "event2", doneSending2, uint64(1001))
checkSumEvent2 := <-doneSending2
close(numbers1)
close(numbers2)
eventSum1 := <-doneSum1
eventSum2 := <-doneSum2
if checkSumEvent1 != eventSum1 ||
// correct value asserted by preceding tests, suffices to be non-zero
checkSumEvent2 == uint64(0) ||
eventSum2 != uint64(0) {
t.Errorf("Not all messages sent were received or unsubscription did not register.\n")
}
}
// TestRemoveListener does basic tests on adding and removing
func TestRemoveListener(t *testing.T) {
evsw := NewEventSwitch()
err := evsw.Start()
if err != nil {
t.Errorf("Failed to start EventSwitch, error: %v", err)
}
count := 10
sum1, sum2 := 0, 0
// add some listeners and make sure they work
evsw.AddListenerForEvent("listener", "event1",
func(data EventData) {
sum1++
})
evsw.AddListenerForEvent("listener", "event2",
func(data EventData) {
sum2++
})
for i := 0; i < count; i++ {
evsw.FireEvent("event1", true)
evsw.FireEvent("event2", true)
}
assert.Equal(t, count, sum1)
assert.Equal(t, count, sum2)
// remove one by event and make sure it is gone
evsw.RemoveListenerForEvent("event2", "listener")
for i := 0; i < count; i++ {
evsw.FireEvent("event1", true)
evsw.FireEvent("event2", true)
}
assert.Equal(t, count*2, sum1)
assert.Equal(t, count, sum2)
// remove the listener entirely and make sure both gone
evsw.RemoveListener("listener")
for i := 0; i < count; i++ {
evsw.FireEvent("event1", true)
evsw.FireEvent("event2", true)
}
assert.Equal(t, count*2, sum1)
assert.Equal(t, count, sum2)
}
// TestAddAndRemoveListenersAsync sets up an EventSwitch, subscribes two
// listeners to three events, and fires a thousand integers for each event.
// These two listeners serve as the baseline validation while other listeners
// are randomly subscribed and unsubscribed.
// More precisely it randomly subscribes new listeners (different from the first
// two listeners) to one of these three events. At the same time it starts
// randomly unsubscribing these additional listeners from all events they are
// at that point subscribed to.
// NOTE: it is important to run this test with race conditions tracking on,
// `go test -race`, to examine for possible race conditions.
func TestRemoveListenersAsync(t *testing.T) {
evsw := NewEventSwitch()
err := evsw.Start()
if err != nil {
t.Errorf("Failed to start EventSwitch, error: %v", err)
}
doneSum1 := make(chan uint64)
doneSum2 := make(chan uint64)
doneSending1 := make(chan uint64)
doneSending2 := make(chan uint64)
doneSending3 := make(chan uint64)
numbers1 := make(chan uint64, 4)
numbers2 := make(chan uint64, 4)
// subscribe two listener to three events
evsw.AddListenerForEvent("listener1", "event1",
func(data EventData) {
numbers1 <- data.(uint64)
})
evsw.AddListenerForEvent("listener1", "event2",
func(data EventData) {
numbers1 <- data.(uint64)
})
evsw.AddListenerForEvent("listener1", "event3",
func(data EventData) {
numbers1 <- data.(uint64)
})
evsw.AddListenerForEvent("listener2", "event1",
func(data EventData) {
numbers2 <- data.(uint64)
})
evsw.AddListenerForEvent("listener2", "event2",
func(data EventData) {
numbers2 <- data.(uint64)
})
evsw.AddListenerForEvent("listener2", "event3",
func(data EventData) {
numbers2 <- data.(uint64)
})
// collect received events for event1
go sumReceivedNumbers(numbers1, doneSum1)
// collect received events for event2
go sumReceivedNumbers(numbers2, doneSum2)
addListenersStress := func() {
s1 := rand.NewSource(time.Now().UnixNano())
r1 := rand.New(s1)
for k := uint16(0); k < 400; k++ {
listenerNumber := r1.Intn(100) + 3
eventNumber := r1.Intn(3) + 1
go evsw.AddListenerForEvent(fmt.Sprintf("listener%v", listenerNumber),
fmt.Sprintf("event%v", eventNumber),
func(_ EventData) {})
}
}
removeListenersStress := func() {
s2 := rand.NewSource(time.Now().UnixNano())
r2 := rand.New(s2)
for k := uint16(0); k < 80; k++ {
listenerNumber := r2.Intn(100) + 3
go evsw.RemoveListener(fmt.Sprintf("listener%v", listenerNumber))
}
}
addListenersStress()
// go fire events
go fireEvents(evsw, "event1", doneSending1, uint64(1))
removeListenersStress()
go fireEvents(evsw, "event2", doneSending2, uint64(1001))
go fireEvents(evsw, "event3", doneSending3, uint64(2001))
checkSumEvent1 := <-doneSending1
checkSumEvent2 := <-doneSending2
checkSumEvent3 := <-doneSending3
checkSum := checkSumEvent1 + checkSumEvent2 + checkSumEvent3
close(numbers1)
close(numbers2)
eventSum1 := <-doneSum1
eventSum2 := <-doneSum2
if checkSum != eventSum1 ||
checkSum != eventSum2 {
t.Errorf("Not all messages sent were received.\n")
}
}
//------------------------------------------------------------------------------
// Helper functions
// sumReceivedNumbers takes two channels and adds all numbers received
// until the receiving channel `numbers` is closed; it then sends the sum
// on `doneSum` and closes that channel. Expected to be run in a go-routine.
func sumReceivedNumbers(numbers, doneSum chan uint64) {
var sum uint64 = 0
for {
j, more := <-numbers
sum += j
if !more {
doneSum <- sum
close(doneSum)
return
}
}
}
// fireEvents takes an EventSwitch and fires a thousand integers under
// a given `event` with the integers mootonically increasing from `offset`
// to `offset` + 999. It additionally returns the addition of all integers
// sent on `doneChan` for assertion that all events have been sent, and enabling
// the test to assert all events have also been received.
func fireEvents(evsw EventSwitch, event string, doneChan chan uint64,
offset uint64) {
var sentSum uint64 = 0
for i := offset; i <= offset+uint64(999); i++ {
sentSum += i
evsw.FireEvent(event, i)
}
doneChan <- sentSum
close(doneChan)
}