cerc-io/plugeth
18
0
Fork 1
Code Issues 4 Pull Requests Packages Projects Releases Wiki Activity

ethreact: remove package

Browse Source
This commit is contained in:
Felix Lange 2014-10-14 02:18:12 +02:00
parent 28570ef109
commit ade980912d
3 changed files with 0 additions and 274 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
ethreact/README.md
View File

@ -1,28 +0,0 @@
## Reactor
Reactor is the internal broadcast engine that allows components to be notified of ethereum stack events such as finding new blocks or change in state.
Event notification is handled via subscription:
var blockChan = make(chan ethreact.Event, 10)
reactor.Subscribe("newBlock", blockChan)
ethreact.Event broadcast on the channel are
type Event struct {
Resource interface{}
Name string
}
Resource is polimorphic depending on the event type and should be typecast before use, e.g:
b := <-blockChan:
block := b.Resource.(*ethchain.Block)
Events are guaranteed to be broadcast in order but the broadcast never blocks or leaks which means while the subscribing event channel is blocked (e.g., full if buffered) further messages will be skipped.
The engine allows arbitrary events to be posted and subscribed to.
ethereum.Reactor().Post("newBlock", newBlock)

183
ethreact/reactor.go
View File

@ -1,183 +0,0 @@
package ethreact
import (
"sync"
"github.com/ethereum/eth-go/ethlog"
)
var logger = ethlog.NewLogger("REACTOR")
const (
eventBufferSize int = 10
)
type EventHandler struct {
lock sync.RWMutex
name string
chans []chan Event
}
// Post the Event with the reactor resource on the channels
// currently subscribed to the event
func (e *EventHandler) Post(event Event) {
e.lock.RLock()
defer e.lock.RUnlock()
// if we want to preserve order pushing to subscibed channels
// dispatching should be syncrounous
// this means if subscribed event channel is blocked
// the reactor dispatch will be blocked, so we need to mitigate by skipping
// rogue blocking subscribers
for i, ch := range e.chans {
select {
case ch <- event:
default:
logger.Debugf("subscribing channel %d to event %s blocked. skipping\n", i, event.Name)
}
}
}
// Add a subscriber to this event
func (e *EventHandler) Add(ch chan Event) {
e.lock.Lock()
defer e.lock.Unlock()
e.chans = append(e.chans, ch)
}
// Remove a subscriber
func (e *EventHandler) Remove(ch chan Event) int {
e.lock.Lock()
defer e.lock.Unlock()
for i, c := range e.chans {
if c == ch {
e.chans = append(e.chans[:i], e.chans[i+1:]...)
}
}
return len(e.chans)
}
// Basic reactor event
type Event struct {
Resource interface{}
Name string
}
// The reactor basic engine. Acts as bridge
// between the events and the subscribers/posters
type ReactorEngine struct {
lock sync.RWMutex
eventChannel chan Event
eventHandlers map[string]*EventHandler
quit chan chan error
running bool
drained chan bool
}
func New() *ReactorEngine {
return &ReactorEngine{
eventHandlers: make(map[string]*EventHandler),
eventChannel: make(chan Event, eventBufferSize),
quit: make(chan chan error, 1),
drained: make(chan bool, 1),
}
}
func (reactor *ReactorEngine) Start() {
reactor.lock.Lock()
defer reactor.lock.Unlock()
if !reactor.running {
go func() {
for {
select {
case status := <-reactor.quit:
reactor.lock.Lock()
defer reactor.lock.Unlock()
reactor.running = false
logger.Infoln("stopped")
status <- nil
return
case event := <-reactor.eventChannel:
// needs to be called syncronously to keep order of events
reactor.dispatch(event)
default:
reactor.drained <- true // blocking till message is coming in
}
}
}()
reactor.running = true
logger.Infoln("started")
}
}
func (reactor *ReactorEngine) Stop() {
if reactor.running {
status := make(chan error)
reactor.quit <- status
select {
case <-reactor.drained:
default:
}
<-status
}
}
func (reactor *ReactorEngine) Flush() {
<-reactor.drained
}
// Subscribe a channel to the specified event
func (reactor *ReactorEngine) Subscribe(event string, eventChannel chan Event) {
reactor.lock.Lock()
defer reactor.lock.Unlock()
eventHandler := reactor.eventHandlers[event]
// Create a new event handler if one isn't available
if eventHandler == nil {
eventHandler = &EventHandler{name: event}
reactor.eventHandlers[event] = eventHandler
}
// Add the events channel to reactor event handler
eventHandler.Add(eventChannel)
logger.Debugf("added new subscription to %s", event)
}
func (reactor *ReactorEngine) Unsubscribe(event string, eventChannel chan Event) {
reactor.lock.Lock()
defer reactor.lock.Unlock()
eventHandler := reactor.eventHandlers[event]
if eventHandler != nil {
len := eventHandler.Remove(eventChannel)
if len == 0 {
reactor.eventHandlers[event] = nil
}
logger.Debugf("removed subscription to %s", event)
}
}
func (reactor *ReactorEngine) Post(event string, resource interface{}) {
reactor.lock.Lock()
defer reactor.lock.Unlock()
if reactor.running {
reactor.eventChannel <- Event{Resource: resource, Name: event}
select {
case <-reactor.drained:
default:
}
}
}
func (reactor *ReactorEngine) dispatch(event Event) {
name := event.Name
eventHandler := reactor.eventHandlers[name]
// if no subscriptions to this event type - no event handler created
// then noone to notify
if eventHandler != nil {
// needs to be called syncronously
eventHandler.Post(event)
}
}

63
ethreact/reactor_test.go
View File

@ -1,63 +0,0 @@
package ethreact
import (
"fmt"
"testing"
)
func TestReactorAdd(t *testing.T) {
reactor := New()
ch := make(chan Event)
reactor.Subscribe("test", ch)
if reactor.eventHandlers["test"] == nil {
t.Error("Expected new eventHandler to be created")
}
reactor.Unsubscribe("test", ch)
if reactor.eventHandlers["test"] != nil {
t.Error("Expected eventHandler to be removed")
}
}
func TestReactorEvent(t *testing.T) {
var name string
reactor := New()
// Buffer the channel, so it doesn't block for this test
cap := 20
ch := make(chan Event, cap)
reactor.Subscribe("even", ch)
reactor.Subscribe("odd", ch)
reactor.Post("even", "disappears") // should not broadcast if engine not started
reactor.Start()
for i := 0; i < cap; i++ {
if i%2 == 0 {
name = "even"
} else {
name = "odd"
}
reactor.Post(name, i)
}
reactor.Post("test", cap) // this should not block
i := 0
reactor.Flush()
close(ch)
for event := range ch {
fmt.Printf("%d: %v", i, event)
if i%2 == 0 {
name = "even"
} else {
name = "odd"
}
if val, ok := event.Resource.(int); ok {
if i != val || event.Name != name {
t.Error("Expected event %d to be of type %s and resource %d, got ", i, name, i, val)
}
} else {
t.Error("Unable to cast")
}
i++
}
if i != cap {
t.Error("excpected exactly %d events, got ", i)
}
reactor.Stop()
}