Skip to content

Commit

Permalink
Merge pull request #140 from fjl/feature/event
Browse files Browse the repository at this point in the history 
Adapt to new event package
  • Loading branch information
@obscuren
obscuren committed Oct 17, 2014
2 parents 294b437 + 0aea5fc commit 9e6411d
Show file tree
Hide file tree
Showing 3 changed files with 95 additions and 130 deletions.
35 changes: 12 additions & 23 deletions javascript/javascript_runtime.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -11,9 +11,9 @@ import (
"github.com/ethereum/eth-go/ethchain"
"github.com/ethereum/eth-go/ethlog"
"github.com/ethereum/eth-go/ethpipe"
"github.com/ethereum/eth-go/ethreact"
"github.com/ethereum/eth-go/ethstate"
"github.com/ethereum/eth-go/ethutil"
"github.com/ethereum/eth-go/event"
"github.com/ethereum/go-ethereum/utils"
"github.com/obscuren/otto"
)
Expand All @@ -25,9 +25,8 @@ type JSRE struct {
Vm *otto.Otto
pipe *ethpipe.JSPipe

blockChan chan ethreact.Event
changeChan chan ethreact.Event
quitChan chan bool
events event.Subscription
quitChan chan bool

objectCb map[string][]otto.Value
}
Expand All @@ -51,8 +50,7 @@ func NewJSRE(ethereum *eth.Ethereum) *JSRE {
ethereum,
otto.New(),
ethpipe.NewJSPipe(ethereum),
make(chan ethreact.Event, 10),
make(chan ethreact.Event, 10),
nil,
make(chan bool),
make(map[string][]otto.Value),
}
Expand All @@ -68,8 +66,8 @@ func NewJSRE(ethereum *eth.Ethereum) *JSRE {
go re.mainLoop()

// Subscribe to events
reactor := ethereum.Reactor()
reactor.Subscribe("newBlock", re.blockChan)
mux := ethereum.EventMux()
re.events = mux.Subscribe(ethchain.NewBlockEvent{})

re.Bind("eth", &JSEthereum{re.pipe, re.Vm, ethereum})

Expand Down Expand Up @@ -105,25 +103,16 @@ func (self *JSRE) Require(file string) error {
}

func (self *JSRE) Stop() {
self.events.Unsubscribe()
// Kill the main loop
self.quitChan <- true

close(self.blockChan)
close(self.quitChan)
close(self.changeChan)
jsrelogger.Infoln("stopped")
}

func (self *JSRE) mainLoop() {
out:
for {
select {
case <-self.quitChan:
break out
case block := <-self.blockChan:
if _, ok := block.Resource.(*ethchain.Block); ok {
}
}
for _ = range self.events.Chan() {
}
}

Expand Down Expand Up @@ -201,13 +190,13 @@ func (self *JSRE) watch(call otto.FunctionCall) otto.Value {
if storageCallback {
self.objectCb[addr+storageAddr] = append(self.objectCb[addr+storageAddr], cb)

event := "storage:" + string(ethutil.Hex2Bytes(addr)) + ":" + string(ethutil.Hex2Bytes(storageAddr))
self.ethereum.Reactor().Subscribe(event, self.changeChan)
// event := "storage:" + string(ethutil.Hex2Bytes(addr)) + ":" + string(ethutil.Hex2Bytes(storageAddr))
// self.ethereum.EventMux().Subscribe(event, self.changeChan)
} else {
self.objectCb[addr] = append(self.objectCb[addr], cb)

event := "object:" + string(ethutil.Hex2Bytes(addr))
self.ethereum.Reactor().Subscribe(event, self.changeChan)
// event := "object:" + string(ethutil.Hex2Bytes(addr))
// self.ethereum.EventMux().Subscribe(event, self.changeChan)
}

return otto.UndefinedValue()
Expand Down
69 changes: 24 additions & 45 deletions mist/ext_app.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -5,8 +5,8 @@ import (

"github.com/ethereum/eth-go/ethchain"
"github.com/ethereum/eth-go/ethpipe"
"github.com/ethereum/eth-go/ethreact"
"github.com/ethereum/eth-go/ethstate"
"github.com/ethereum/eth-go/event"
"github.com/ethereum/go-ethereum/javascript"
"gopkg.in/qml.v1"
)
Expand All @@ -28,9 +28,7 @@ type ExtApplication struct {
*ethpipe.JSPipe
eth ethchain.EthManager

blockChan chan ethreact.Event
messageChan chan ethreact.Event
quitChan chan bool
events event.Subscription
watcherQuitChan chan bool

filters map[string]*ethchain.Filter
Expand All @@ -40,19 +38,14 @@ type ExtApplication struct {
}

func NewExtApplication(container AppContainer, lib *UiLib) *ExtApplication {
app := &ExtApplication{
ethpipe.NewJSPipe(lib.eth),
lib.eth,
make(chan ethreact.Event, 100),
make(chan ethreact.Event, 100),
make(chan bool),
make(chan bool),
make(map[string]*ethchain.Filter),
container,
lib,
return &ExtApplication{
JSPipe: ethpipe.NewJSPipe(lib.eth),
eth: lib.eth,
watcherQuitChan: make(chan bool),
filters: make(map[string]*ethchain.Filter),
container: container,
lib: lib,
}

return app
}

func (app *ExtApplication) run() {
Expand All @@ -67,14 +60,13 @@ func (app *ExtApplication) run() {
return
}

// Subscribe to events
mux := app.lib.eth.EventMux()
app.events = mux.Subscribe(ethchain.NewBlockEvent{}, ethstate.Messages(nil))

// Call the main loop
go app.mainLoop()

// Subscribe to events
reactor := app.lib.eth.Reactor()
reactor.Subscribe("newBlock", app.blockChan)
reactor.Subscribe("messages", app.messageChan)

app.container.NewWatcher(app.watcherQuitChan)

win := app.container.Window()
Expand All @@ -85,42 +77,29 @@ func (app *ExtApplication) run() {
}

func (app *ExtApplication) stop() {
// Clean up
reactor := app.lib.eth.Reactor()
reactor.Unsubscribe("newBlock", app.blockChan)
app.events.Unsubscribe()

// Kill the main loop
app.quitChan <- true
app.watcherQuitChan <- true

close(app.blockChan)
close(app.quitChan)

app.container.Destroy()
}

func (app *ExtApplication) mainLoop() {
out:
for {
select {
case <-app.quitChan:
break out
case block := <-app.blockChan:
if block, ok := block.Resource.(*ethchain.Block); ok {
app.container.NewBlock(block)
}
case msg := <-app.messageChan:
if messages, ok := msg.Resource.(ethstate.Messages); ok {
for id, filter := range app.filters {
msgs := filter.FilterMessages(messages)
if len(msgs) > 0 {
app.container.Messages(msgs, id)
}
for ev := range app.events.Chan() {
switch ev := ev.(type) {
case ethchain.NewBlockEvent:
app.container.NewBlock(ev.Block)

case ethstate.Messages:
for id, filter := range app.filters {
msgs := filter.FilterMessages(ev)
if len(msgs) > 0 {
app.container.Messages(msgs, id)
}
}
}
}

}

func (self *ExtApplication) Watch(filterOptions map[string]interface{}, identifier string) {
Expand Down
121 changes: 59 additions & 62 deletions mist/gui.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,6 @@ import (
"github.com/ethereum/eth-go/ethlog"
"github.com/ethereum/eth-go/ethminer"
"github.com/ethereum/eth-go/ethpipe"
"github.com/ethereum/eth-go/ethreact"
"github.com/ethereum/eth-go/ethutil"
"github.com/ethereum/eth-go/ethwire"
"gopkg.in/qml.v1"
Expand Down Expand Up @@ -376,15 +375,6 @@ func (gui *Gui) update() {
gui.win.Root().Call("addPlugin", plugin.Path, "")
}

var (
blockChan = make(chan ethreact.Event, 100)
txChan = make(chan ethreact.Event, 100)
objectChan = make(chan ethreact.Event, 100)
peerChan = make(chan ethreact.Event, 100)
chainSyncChan = make(chan ethreact.Event, 100)
miningChan = make(chan ethreact.Event, 100)
)

peerUpdateTicker := time.NewTicker(5 * time.Second)
generalUpdateTicker := time.NewTicker(500 * time.Millisecond)
statsUpdateTicker := time.NewTicker(5 * time.Second)
Expand All @@ -397,61 +387,82 @@ func (gui *Gui) update() {
lastBlockLabel := gui.getObjectByName("lastBlockLabel")
miningLabel := gui.getObjectByName("miningLabel")

events := gui.eth.EventMux().Subscribe(
eth.ChainSyncEvent{},
eth.PeerListEvent{},
ethchain.NewBlockEvent{},
ethchain.TxEvent{},
ethminer.Event{},
)

// nameReg := gui.pipe.World().Config().Get("NameReg")
// mux.Subscribe("object:"+string(nameReg.Address()), objectChan)

go func() {
defer events.Unsubscribe()
for {
select {
case b := <-blockChan:
block := b.Resource.(*ethchain.Block)
gui.processBlock(block, false)
if bytes.Compare(block.Coinbase, gui.address()) == 0 {
gui.setWalletValue(gui.eth.StateManager().CurrentState().GetAccount(gui.address()).Balance, nil)
case ev, isopen := <-events.Chan():
if !isopen {
return
}
case txMsg := <-txChan:
tx := txMsg.Resource.(*ethchain.Transaction)
switch ev := ev.(type) {
case ethchain.NewBlockEvent:
gui.processBlock(ev.Block, false)
if bytes.Compare(ev.Block.Coinbase, gui.address()) == 0 {
gui.setWalletValue(gui.eth.StateManager().CurrentState().GetAccount(gui.address()).Balance, nil)
}

if txMsg.Name == "newTx:pre" {
object := state.GetAccount(gui.address())
case ethchain.TxEvent:
tx := ev.Tx
if ev.Type == ethchain.TxPre {
object := state.GetAccount(gui.address())

if bytes.Compare(tx.Sender(), gui.address()) == 0 {
unconfirmedFunds.Sub(unconfirmedFunds, tx.Value)
} else if bytes.Compare(tx.Recipient, gui.address()) == 0 {
unconfirmedFunds.Add(unconfirmedFunds, tx.Value)
}
if bytes.Compare(tx.Sender(), gui.address()) == 0 {
unconfirmedFunds.Sub(unconfirmedFunds, tx.Value)
} else if bytes.Compare(tx.Recipient, gui.address()) == 0 {
unconfirmedFunds.Add(unconfirmedFunds, tx.Value)
}

gui.setWalletValue(object.Balance, unconfirmedFunds)

gui.insertTransaction("pre", tx)

gui.setWalletValue(object.Balance, unconfirmedFunds)
} else if ev.Type == ethchain.TxPost {
object := state.GetAccount(gui.address())
if bytes.Compare(tx.Sender(), gui.address()) == 0 {
object.SubAmount(tx.Value)

gui.insertTransaction("pre", tx)
} else {
object := state.GetAccount(gui.address())
if bytes.Compare(tx.Sender(), gui.address()) == 0 {
object.SubAmount(tx.Value)
//gui.getObjectByName("transactionView").Call("addTx", ethpipe.NewJSTx(tx), "send")
gui.txDb.Put(tx.Hash(), tx.RlpEncode())
} else if bytes.Compare(tx.Recipient, gui.address()) == 0 {
object.AddAmount(tx.Value)

//gui.getObjectByName("transactionView").Call("addTx", ethpipe.NewJSTx(tx), "send")
gui.txDb.Put(tx.Hash(), tx.RlpEncode())
} else if bytes.Compare(tx.Recipient, gui.address()) == 0 {
object.AddAmount(tx.Value)
//gui.getObjectByName("transactionView").Call("addTx", ethpipe.NewJSTx(tx), "recv")
gui.txDb.Put(tx.Hash(), tx.RlpEncode())
}

//gui.getObjectByName("transactionView").Call("addTx", ethpipe.NewJSTx(tx), "recv")
gui.txDb.Put(tx.Hash(), tx.RlpEncode())
gui.setWalletValue(object.Balance, nil)

state.UpdateStateObject(object)
}

gui.setWalletValue(object.Balance, nil)
// case object:
// gui.loadAddressBook()

case eth.PeerListEvent:
gui.setPeerInfo()

state.UpdateStateObject(object)
case ethminer.Event:
if ev.Type == ethminer.Started {
gui.miner = ev.Miner
} else {
gui.miner = nil
}
}

case <-objectChan:
gui.loadAddressBook()
case <-peerChan:
gui.setPeerInfo()
case <-peerUpdateTicker.C:
gui.setPeerInfo()
case msg := <-miningChan:
if msg.Name == "miner:start" {
gui.miner = msg.Resource.(*ethminer.Miner)
} else {
gui.miner = nil
}
case <-generalUpdateTicker.C:
statusText := "#" + gui.eth.BlockChain().CurrentBlock.Number.String()
lastBlockLabel.Set("text", statusText)
Expand All @@ -478,20 +489,6 @@ func (gui *Gui) update() {
}
}
}()

reactor := gui.eth.Reactor()

reactor.Subscribe("newBlock", blockChan)
reactor.Subscribe("newTx:pre", txChan)
reactor.Subscribe("newTx:post", txChan)
reactor.Subscribe("chainSync", chainSyncChan)
reactor.Subscribe("miner:start", miningChan)
reactor.Subscribe("miner:stop", miningChan)

nameReg := gui.pipe.World().Config().Get("NameReg")
reactor.Subscribe("object:"+string(nameReg.Address()), objectChan)

reactor.Subscribe("peerList", peerChan)
}

func (gui *Gui) setStatsPane() {
Expand Down

0 comments on commit 9e6411d

Please sign in to comment.