roysc/plugeth
1
0
Fork 0
forked from cerc-io/plugeth
Code Pull Requests Activity

pending / chain event

Browse Source
This commit is contained in:
obscuren 2015-02-05 11:55:03 -08:00
parent bcacaaa4f4
commit c64852dbcc
6 changed files with 71 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
cmd/mist/assets/html/home.html
View File

@ -60,7 +60,7 @@
var web3 = require('web3');
var eth = web3.eth;
web3.setProvider(new web3.providers.HttpSyncProvider('http://localhost:8080'));
web3.setProvider(new web3.providers.HttpSyncProvider('http://localhost:8545'));
document.querySelector("#number").innerHTML = eth.number;
document.querySelector("#coinbase").innerHTML = eth.coinbase
@ -69,6 +69,14 @@
document.querySelector("#gas_price").innerHTML = eth.gasPrice;
document.querySelector("#mining").innerHTML = eth.mining;
document.querySelector("#listening").innerHTML = eth.listening;
eth.watch('pending').changed(function() {
console.log("pending changed");
});
eth.watch('chain').changed(function() {
console.log("chain changed");
});
</script>
</html>

8
core/block_processor.go
View File

@ -19,6 +19,10 @@ import (
"gopkg.in/fatih/set.v0"
)
type PendingBlockEvent struct {
Block *types.Block
}
var statelogger = logger.NewLogger("BLOCK")
type EthManager interface {
@ -154,6 +158,10 @@ done:
block.Reward = cumulativeSum
block.Header().GasUsed = totalUsedGas
if transientProcess {
go self.eventMux.Post(PendingBlockEvent{block})
}
return receipts, handled, unhandled, erroneous, err
}

5
core/filter.go
View File

@ -33,8 +33,9 @@ type Filter struct {
max int
topics [][]byte
BlockCallback func(*types.Block)
LogsCallback func(state.Logs)
BlockCallback func(*types.Block)
PendingCallback func(*types.Block)
LogsCallback func(state.Logs)
}
// Create a new filter which uses a bloom filter on blocks to figure out whether a particular block

11
event/filter/old_filter.go
View File

@ -59,7 +59,7 @@ func (self *FilterManager) GetFilter(id int) *core.Filter {
func (self *FilterManager) filterLoop() {
// Subscribe to events
events := self.eventMux.Subscribe(core.NewBlockEvent{}, state.Logs(nil))
events := self.eventMux.Subscribe(core.PendingBlockEvent{}, core.NewBlockEvent{}, state.Logs(nil))
out:
for {
@ -77,6 +77,15 @@ out:
}
self.filterMu.RUnlock()
case core.PendingBlockEvent:
self.filterMu.RLock()
for _, filter := range self.filters {
if filter.PendingCallback != nil {
filter.PendingCallback(event.Block)
}
}
self.filterMu.RUnlock()
case state.Logs:
self.filterMu.RLock()
for _, filter := range self.filters {

16
rpc/message.go
View File

@ -205,7 +205,6 @@ func (req *RpcRequest) ToFilterArgs() (*FilterOptions, error) {
if len(req.Params) < 1 {
return nil, NewErrorResponse(ErrorArguments)
}
fmt.Println("FILTER PARAMS", string(req.Params[0]))
args := new(FilterOptions)
r := bytes.NewReader(req.Params[0])
@ -217,6 +216,21 @@ func (req *RpcRequest) ToFilterArgs() (*FilterOptions, error) {
return args, nil
}
func (req *RpcRequest) ToFilterStringArgs() (string, error) {
if len(req.Params) < 1 {
return "", NewErrorResponse(ErrorArguments)
}
var args string
err := json.Unmarshal(req.Params[0], &args)
if err != nil {
return "", NewErrorResponse(ErrorDecodeArgs)
}
rpclogger.DebugDetailf("%T %v", args, args)
return args, nil
}
func (req *RpcRequest) ToFilterChangedArgs() (int, error) {
if len(req.Params) < 1 {
return 0, NewErrorResponse(ErrorArguments)

26
rpc/packages.go
View File

@ -32,6 +32,7 @@ import (
"sync"
"github.com/ethereum/go-ethereum/core"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/ethdb"
"github.com/ethereum/go-ethereum/ethutil"
@ -88,6 +89,25 @@ func (self *EthereumApi) NewFilter(args *FilterOptions, reply *interface{}) erro
return nil
}
func (self *EthereumApi) NewFilterString(args string, reply *interface{}) error {
var id int
filter := core.NewFilter(self.xeth.Backend())
callback := func(block *types.Block) {
self.logs[id] = append(self.logs[id], &state.StateLog{})
}
if args == "pending" {
filter.PendingCallback = callback
} else if args == "chain" {
filter.BlockCallback = callback
}
id = self.filterManager.InstallFilter(filter)
*reply = id
return nil
}
func (self *EthereumApi) FilterChanged(id int, reply *interface{}) error {
self.logMut.RLock()
defer self.logMut.RUnlock()
@ -389,6 +409,12 @@ func (p *EthereumApi) GetRequestReply(req *RpcRequest, reply *interface{}) error
return err
}
return p.NewFilter(args, reply)
case "eth_newFilterString":
args, err := req.ToFilterStringArgs()
if err != nil {
return err
}
return p.NewFilterString(args, reply)
case "eth_changed":
args, err := req.ToFilterChangedArgs()
if err != nil {