Merge branch 'release/0.9.0'

3 files changed, 134 insertions, 5 deletions

diff --git a/event/filter/eth_filter.go b/event/filter/eth_filter.go
new file mode 100644
index 000000000..73d2cd935
--- /dev/null
+++ b/event/filter/eth_filter.go
@@ -0,0 +1,111 @@
+package filter
+
+// TODO make use of the generic filtering system
+
+import (
+	"fmt"
+	"sync"
+
+	"github.com/ethereum/go-ethereum/core"
+	"github.com/ethereum/go-ethereum/event"
+	"github.com/ethereum/go-ethereum/state"
+)
+
+type FilterManager struct {
+	eventMux *event.TypeMux
+
+	filterMu sync.RWMutex
+	filterId int
+	filters  map[int]*core.Filter
+
+	quit chan struct{}
+}
+
+func NewFilterManager(mux *event.TypeMux) *FilterManager {
+	return &FilterManager{
+		eventMux: mux,
+		filters:  make(map[int]*core.Filter),
+	}
+}
+
+func (self *FilterManager) Start() {
+	go self.filterLoop()
+}
+
+func (self *FilterManager) Stop() {
+	close(self.quit)
+}
+
+func (self *FilterManager) InstallFilter(filter *core.Filter) (id int) {
+	self.filterMu.Lock()
+	defer self.filterMu.Unlock()
+	id = self.filterId
+	self.filters[id] = filter
+	self.filterId++
+
+	return id
+}
+
+func (self *FilterManager) UninstallFilter(id int) {
+	self.filterMu.Lock()
+	defer self.filterMu.Unlock()
+	delete(self.filters, id)
+}
+
+// GetFilter retrieves a filter installed using InstallFilter.
+// The filter may not be modified.
+func (self *FilterManager) GetFilter(id int) *core.Filter {
+	self.filterMu.RLock()
+	defer self.filterMu.RUnlock()
+	return self.filters[id]
+}
+
+func (self *FilterManager) filterLoop() {
+	// Subscribe to events
+	events := self.eventMux.Subscribe(
+		core.PendingBlockEvent{},
+		//core.ChainEvent{},
+		state.Logs(nil))
+
+out:
+	for {
+		select {
+		case <-self.quit:
+			break out
+		case event := <-events.Chan():
+			switch event := event.(type) {
+			case core.ChainEvent:
+				fmt.Println("filter start")
+				self.filterMu.RLock()
+				for _, filter := range self.filters {
+					if filter.BlockCallback != nil {
+						filter.BlockCallback(event.Block)
+					}
+				}
+				self.filterMu.RUnlock()
+				fmt.Println("filter stop")
+
+			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 {
+					if filter.LogsCallback != nil {
+						msgs := filter.FilterLogs(event)
+						if len(msgs) > 0 {
+							filter.LogsCallback(msgs)
+						}
+					}
+				}
+				self.filterMu.RUnlock()
+			}
+		}
+	}
+}
diff --git a/event/filter/filter.go b/event/filter/filter.go
index 9817d5782..ca767f413 100644
--- a/event/filter/filter.go
+++ b/event/filter/filter.go
@@ -68,3 +68,11 @@ out:
 		}
 	}
 }
+
+func (self *Filters) Match(a, b Filter) bool {
+	return reflect.TypeOf(a) == reflect.TypeOf(b) && a.Compare(b)
+}
+
+func (self *Filters) Get(i int) Filter {
+	return self.watchers[i]
+}
diff --git a/event/filter/generic_filter.go b/event/filter/generic_filter.go
index b04b4801e..2ce0f0642 100644
--- a/event/filter/generic_filter.go
+++ b/event/filter/generic_filter.go
@@ -2,19 +2,29 @@ package filter
 
 type Generic struct {
 	Str1, Str2, Str3 string
+	Data             map[string]struct{}
 
 	Fn func(data interface{})
 }
 
+// self = registered, f = incoming
 func (self Generic) Compare(f Filter) bool {
+	var strMatch, dataMatch = true, true
+
 	filter := f.(Generic)
-	if (len(self.Str1) == 0 || filter.Str1 == self.Str1) &&
-		(len(self.Str2) == 0 || filter.Str2 == self.Str2) &&
-		(len(self.Str3) == 0 || filter.Str3 == self.Str3) {
-		return true
+	if (len(self.Str1) > 0 && filter.Str1 != self.Str1) ||
+		(len(self.Str2) > 0 && filter.Str2 != self.Str2) ||
+		(len(self.Str3) > 0 && filter.Str3 != self.Str3) {
+		strMatch = false
+	}
+
+	for k, _ := range self.Data {
+		if _, ok := filter.Data[k]; !ok {
+			return false
+		}
 	}
 
-	return false
+	return strMatch && dataMatch
 }
 
 func (self Generic) Trigger(data interface{}) {