diff --git a/ethchain/filter.go b/ethchain/filter.go
index 5ed9af977ab3938c9b4f5eb7ee540a6c05f0830d..d9f1796f4c87527e9022505b80580d84f97644c1 100644
--- a/ethchain/filter.go
+++ b/ethchain/filter.go
@@ -23,6 +23,9 @@ type Filter struct {
max int
altered []data
+
+ BlockCallback func(*Block)
+ MessageCallback func(ethstate.Messages)
}
// Create a new filter which uses a bloom filter on blocks to figure out whether a particular block
diff --git a/ethereum.go b/ethereum.go
index 4c5e13b6df12ca0c6d001f8419d249e6977591e8..fdfb59b09f71dac5530c5dccf2ebf50ebf1aa86d 100644
--- a/ethereum.go
+++ b/ethereum.go
@@ -16,6 +16,7 @@ import (
"github.com/ethereum/eth-go/ethlog"
"github.com/ethereum/eth-go/ethreact"
"github.com/ethereum/eth-go/ethrpc"
+ "github.com/ethereum/eth-go/ethstate"
"github.com/ethereum/eth-go/ethutil"
"github.com/ethereum/eth-go/ethwire"
)
@@ -87,6 +88,8 @@ type Ethereum struct {
clientIdentity ethwire.ClientIdentity
isUpToDate bool
+
+ filters map[int]*ethchain.Filter
}
func New(db ethutil.Database, clientIdentity ethwire.ClientIdentity, keyManager *ethcrypto.KeyManager, caps Caps, usePnp bool) (*Ethereum, error) {
@@ -116,6 +119,7 @@ func New(db ethutil.Database, clientIdentity ethwire.ClientIdentity, keyManager
keyManager: keyManager,
clientIdentity: clientIdentity,
isUpToDate: true,
+ filters: make(map[int]*ethchain.Filter),
}
ethereum.reactor = ethreact.New()
@@ -386,6 +390,7 @@ func (s *Ethereum) Start(seed bool) {
// Start the reaping processes
go s.ReapDeadPeerHandler()
go s.update()
+ go s.filterLoop()
if seed {
s.Seed()
@@ -536,6 +541,60 @@ out:
}
}
+var filterId = 0
+
+func (self *Ethereum) InstallFilter(object map[string]interface{}) (*ethchain.Filter, int) {
+ defer func() { filterId++ }()
+
+ filter := ethchain.NewFilterFromMap(object, self)
+ self.filters[filterId] = filter
+
+ return filter, filterId
+}
+
+func (self *Ethereum) UninstallFilter(id int) {
+ delete(self.filters, id)
+}
+
+func (self *Ethereum) GetFilter(id int) *ethchain.Filter {
+ return self.filters[id]
+}
+
+func (self *Ethereum) filterLoop() {
+ blockChan := make(chan ethreact.Event, 5)
+ messageChan := make(chan ethreact.Event, 5)
+ // Subscribe to events
+ reactor := self.Reactor()
+ reactor.Subscribe("newBlock", blockChan)
+ reactor.Subscribe("messages", messageChan)
+out:
+ for {
+ select {
+ case <-self.quit:
+ break out
+ case block := <-blockChan:
+ if block, ok := block.Resource.(*ethchain.Block); ok {
+ for _, filter := range self.filters {
+ if filter.BlockCallback != nil {
+ filter.BlockCallback(block)
+ }
+ }
+ }
+ case msg := <-messageChan:
+ if messages, ok := msg.Resource.(ethstate.Messages); ok {
+ for _, filter := range self.filters {
+ if filter.MessageCallback != nil {
+ msgs := filter.FilterMessages(messages)
+ if len(msgs) > 0 {
+ filter.MessageCallback(msgs)
+ }
+ }
+ }
+ }
+ }
+ }
+}
+
func bootstrapDb(db ethutil.Database) {
d, _ := db.Get([]byte("ProtocolVersion"))
protov := ethutil.NewValue(d).Uint()
diff --git a/ethpipe/js_pipe.go b/ethpipe/js_pipe.go
index eeece5179f0c3689aae55e348527b217f15133ea..7ee183c849ec9ffce6a9246475425be7e28614d2 100644
--- a/ethpipe/js_pipe.go
+++ b/ethpipe/js_pipe.go
@@ -3,12 +3,10 @@ package ethpipe
import (
"bytes"
"encoding/json"
- "fmt"
"sync/atomic"
"github.com/ethereum/eth-go/ethchain"
"github.com/ethereum/eth-go/ethcrypto"
- "github.com/ethereum/eth-go/ethreact"
"github.com/ethereum/eth-go/ethstate"
"github.com/ethereum/eth-go/ethutil"
)
@@ -234,102 +232,11 @@ func (self *JSPipe) CompileMutan(code string) string {
return ethutil.Bytes2Hex(data)
}
-func (self *JSPipe) Watch(object map[string]interface{}) *JSFilter {
- return NewJSFilterFromMap(object, self.Pipe.obj)
- /*} else if str, ok := object.(string); ok {
- println("str")
- return NewJSFilterFromString(str, self.Pipe.obj)
- */
-}
-
-func (self *JSPipe) Messages(object map[string]interface{}) string {
- filter := self.Watch(object)
- filter.Uninstall()
-
- return filter.Messages()
-
-}
-
-type JSFilter struct {
- eth ethchain.EthManager
- *ethchain.Filter
- quit chan bool
-
- BlockCallback func(*ethchain.Block)
- MessageCallback func(ethstate.Messages)
-}
-
-func NewJSFilterFromMap(object map[string]interface{}, eth ethchain.EthManager) *JSFilter {
- filter := &JSFilter{eth, ethchain.NewFilterFromMap(object, eth), make(chan bool), nil, nil}
-
- go filter.mainLoop()
-
- return filter
-}
-
-func NewJSFilterFromString(str string, eth ethchain.EthManager) *JSFilter {
- return nil
-}
-
-func (self *JSFilter) MessagesToJson(messages ethstate.Messages) string {
+func ToJSMessages(messages ethstate.Messages) *ethutil.List {
var msgs []JSMessage
for _, m := range messages {
msgs = append(msgs, NewJSMessage(m))
}
- // Return an empty array instead of "null"
- if len(msgs) == 0 {
- return "[]"
- }
-
- b, err := json.Marshal(msgs)
- if err != nil {
- return "{\"error\":" + err.Error() + "}"
- }
-
- return string(b)
-}
-
-func (self *JSFilter) Messages() string {
- return self.MessagesToJson(self.Find())
-}
-
-func (self *JSFilter) mainLoop() {
- blockChan := make(chan ethreact.Event, 5)
- messageChan := make(chan ethreact.Event, 5)
- // Subscribe to events
- reactor := self.eth.Reactor()
- reactor.Subscribe("newBlock", blockChan)
- reactor.Subscribe("messages", messageChan)
-out:
- for {
- select {
- case <-self.quit:
- break out
- case block := <-blockChan:
- if block, ok := block.Resource.(*ethchain.Block); ok {
- if self.BlockCallback != nil {
- self.BlockCallback(block)
- }
- }
- case msg := <-messageChan:
- if messages, ok := msg.Resource.(ethstate.Messages); ok {
- if self.MessageCallback != nil {
- println("messages!")
- msgs := self.FilterMessages(messages)
- if len(msgs) > 0 {
- self.MessageCallback(msgs)
- }
- }
- }
- }
- }
-}
-
-func (self *JSFilter) Changed(object interface{}) {
- fmt.Printf("%T\n", object)
-}
-
-func (self *JSFilter) Uninstall() {
- self.quit <- true
+ return ethutil.NewList(msgs)
}
diff --git a/ethutil/list.go b/ethutil/list.go
index 18bf047921441ff60d3f442f3685304b5d3243b6..0aa657a147494092a82f80d24a9d6709ee4a5ae9 100644
--- a/ethutil/list.go
+++ b/ethutil/list.go
@@ -20,6 +20,10 @@ func NewList(t interface{}) *List {
return &List{list, list.Len()}
}
+func EmptyList() *List {
+ return NewList([]interface{}{})
+}
+
// Get N element from the embedded slice. Returns nil if OOB.
func (self *List) Get(i int) interface{} {
if self.list.Len() > i {