Added timeout for filter & removed clipboard. Closes #350

Author: obscuren

rpc/packages.go

@@ -13,6 +13,7 @@ import (
 	"math/big"
 	"strings"
 	"sync"
+	"time"
 
 	"github.com/ethereum/go-ethereum/core"
 	"github.com/ethereum/go-ethereum/core/types"
@@ -31,13 +32,14 @@ const (
 
 type EthereumApi struct {
 	xeth          *xeth.XEth
+	quit          chan struct{}
 	filterManager *filter.FilterManager
 
 	logMut sync.RWMutex
-	logs   map[int]state.Logs
+	logs   map[int]*logFilter
 
 	messagesMut sync.RWMutex
-	messages    map[int][]xeth.WhisperMessage
+	messages    map[int]*whisperFilter
 	// Register keeps a list of accounts and transaction data
 	regmut   sync.Mutex
 	register map[string][]*NewTxArgs
@@ -49,12 +51,14 @@ func NewEthereumApi(eth *xeth.XEth) *EthereumApi {
 	db, _ := ethdb.NewLDBDatabase("dapps")
 	api := &EthereumApi{
 		xeth:          eth,
+		quit:          make(chan struct{}),
 		filterManager: filter.NewFilterManager(eth.Backend().EventMux()),
-		logs:          make(map[int]state.Logs),
-		messages:      make(map[int][]xeth.WhisperMessage),
+		logs:          make(map[int]*logFilter),
+		messages:      make(map[int]*whisperFilter),
 		db:            db,
 	}
 	go api.filterManager.Start()
+	go api.start()
 
 	return api
 }
@@ -97,7 +101,11 @@ func (self *EthereumApi) NewFilter(args *FilterOptions, reply *interface{}) erro
 		self.logMut.Lock()
 		defer self.logMut.Unlock()
 
-		self.logs[id] = append(self.logs[id], logs...)
+		if self.logs[id] == nil {
+			self.logs[id] = &logFilter{timeout: time.Now()}
+		}
+
+		self.logs[id].add(logs...)
 	}
 	id = self.filterManager.InstallFilter(filter)
 	*reply = id
@@ -113,7 +121,11 @@ func (self *EthereumApi) NewFilterString(args string, reply *interface{}) error
 		self.logMut.Lock()
 		defer self.logMut.Unlock()
 
-		self.logs[id] = append(self.logs[id], &state.StateLog{})
+		if self.logs[id] == nil {
+			self.logs[id] = &logFilter{timeout: time.Now()}
+		}
+
+		self.logs[id].add(&state.StateLog{})
 	}
 	if args == "pending" {
 		filter.PendingCallback = callback
@@ -131,9 +143,9 @@ func (self *EthereumApi) FilterChanged(id int, reply *interface{}) error {
 	self.logMut.Lock()
 	defer self.logMut.Unlock()
 
-	*reply = toLogs(self.logs[id])
-
-	self.logs[id] = nil // empty the logs
+	if self.logs[id] != nil {
+		*reply = toLogs(self.logs[id].get())
+	}
 
 	return nil
 }
@@ -331,7 +343,10 @@ func (p *EthereumApi) NewWhisperFilter(args *xeth.Options, reply *interface{}) e
 	args.Fn = func(msg xeth.WhisperMessage) {
 		p.messagesMut.Lock()
 		defer p.messagesMut.Unlock()
-		p.messages[id] = append(p.messages[id], msg)
+		if p.messages[id] == nil {
+			p.messages[id] = &whisperFilter{timeout: time.Now()}
+		}
+		p.messages[id].add(msg) // = append(p.messages[id], msg)
 	}
 	id = p.xeth.Whisper().Watch(args)
 	*reply = id
@@ -342,9 +357,9 @@ func (self *EthereumApi) MessagesChanged(id int, reply *interface{}) error {
 	self.messagesMut.Lock()
 	defer self.messagesMut.Unlock()
 
-	*reply = self.messages[id]
-
-	self.messages[id] = nil // empty the messages
+	if self.messages[id] != nil {
+		*reply = self.messages[id].get()
+	}
 
 	return nil
 }
@@ -535,3 +550,34 @@ func (p *EthereumApi) GetRequestReply(req *RpcRequest, reply *interface{}) error
 	rpclogger.DebugDetailf("Reply: %T %s", reply, reply)
 	return nil
 }
+
+var filterTickerTime = 15 * time.Second
+
+func (self *EthereumApi) start() {
+	timer := time.NewTicker(filterTickerTime)
+done:
+	for {
+		select {
+		case <-timer.C:
+			self.logMut.Lock()
+			self.messagesMut.Lock()
+			for id, filter := range self.logs {
+				if time.Since(filter.timeout) > 20*time.Second {
+					delete(self.logs, id)
+				}
+			}
+
+			for id, filter := range self.messages {
+				if time.Since(filter.timeout) > 20*time.Second {
+					delete(self.messages, id)
+				}
+			}
+		case <-self.quit:
+			break done
+		}
+	}
+}
+
+func (self *EthereumApi) stop() {
+	close(self.quit)
+}

rpc/packages_test.go

@@ -0,0 +1,37 @@
+package rpc
+
+import (
+	"sync"
+	"testing"
+	"time"
+)
+
+func TestFilterClose(t *testing.T) {
+	api := &EthereumApi{
+		logs:     make(map[int]*logFilter),
+		messages: make(map[int]*whisperFilter),
+		quit:     make(chan struct{}),
+	}
+
+	filterTickerTime = 1
+	api.logs[0] = &logFilter{}
+	api.messages[0] = &whisperFilter{}
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go api.start()
+	go func() {
+		select {
+		case <-time.After(500 * time.Millisecond):
+			api.stop()
+			wg.Done()
+		}
+	}()
+	wg.Wait()
+	if len(api.logs) != 0 {
+		t.Error("expected logs to be empty")
+	}
+
+	if len(api.messages) != 0 {
+		t.Error("expected messages to be empty")
+	}
+}

rpc/util.go

@@ -20,10 +20,12 @@ import (
 	"encoding/json"
 	"io"
 	"net/http"
+	"time"
 
 	"github.com/ethereum/go-ethereum/ethutil"
 	"github.com/ethereum/go-ethereum/logger"
 	"github.com/ethereum/go-ethereum/state"
+	"github.com/ethereum/go-ethereum/xeth"
 )
 
 var rpclogger = logger.NewLogger("RPC")
@@ -100,3 +102,34 @@ func toLogs(logs state.Logs) (ls []Log) {
 
 	return
 }
+
+type whisperFilter struct {
+	messages []xeth.WhisperMessage
+	timeout  time.Time
+}
+
+func (w *whisperFilter) add(msgs ...xeth.WhisperMessage) {
+	w.messages = append(w.messages, msgs...)
+}
+func (w *whisperFilter) get() []xeth.WhisperMessage {
+	w.timeout = time.Now()
+	tmp := w.messages
+	w.messages = nil
+	return tmp
+}
+
+type logFilter struct {
+	logs    state.Logs
+	timeout time.Time
+}
+
+func (l *logFilter) add(logs ...state.Log) {
+	l.logs = append(l.logs, logs...)
+}
+
+func (l *logFilter) get() state.Logs {
+	l.timeout = time.Now()
+	tmp := l.logs
+	l.logs = nil
+	return tmp
+}