Merge pull request #864 from obscuren/filter_changes

xeth, core, event/filter, rpc: new block and transaction filters

core/filter.go

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

core/transaction_pool.go

@@ -204,6 +204,27 @@ func (self *TxPool) AddTransactions(txs []*types.Transaction) {
 	}
 }
 
+// GetTransaction allows you to check the pending and queued transaction in the
+// transaction pool.
+// It has two stategies, first check the pool (map) then check the queue
+func (tp *TxPool) GetTransaction(hash common.Hash) *types.Transaction {
+	// check the txs first
+	if tx, ok := tp.txs[hash]; ok {
+		return tx
+	}
+
+	// check queue
+	for _, txs := range tp.queue {
+		for _, tx := range txs {
+			if tx.Hash() == hash {
+				return tx
+			}
+		}
+	}
+
+	return nil
+}
+
 func (self *TxPool) GetTransactions() (txs types.Transactions) {
 	self.mu.RLock()
 	defer self.mu.RUnlock()

event/filter/eth_filter.go

@@ -88,8 +88,8 @@ out:
 			case core.TxPreEvent:
 				self.filterMu.RLock()
 				for _, filter := range self.filters {
-					if filter.PendingCallback != nil {
-						filter.PendingCallback(event.Tx)
+					if filter.TransactionCallback != nil {
+						filter.TransactionCallback(event.Tx)
 					}
 				}
 				self.filterMu.RUnlock()

rpc/api.go

@@ -322,14 +322,13 @@ func (api *EthereumApi) GetRequestReply(req *RpcRequest, reply *interface{}) err
 			return err
 		}
 
-		id := api.xeth().RegisterFilter(args.Earliest, args.Latest, args.Skip, args.Max, args.Address, args.Topics)
+		id := api.xeth().NewLogFilter(args.Earliest, args.Latest, args.Skip, args.Max, args.Address, args.Topics)
 		*reply = newHexNum(big.NewInt(int64(id)).Bytes())
+
 	case "eth_newBlockFilter":
-		args := new(FilterStringArgs)
-		if err := json.Unmarshal(req.Params, &args); err != nil {
-			return err
-		}
-		*reply = newHexNum(api.xeth().NewFilterString(args.Word))
+		*reply = newHexNum(api.xeth().NewBlockFilter())
+	case "eth_newPendingTransactionFilter":
+		*reply = newHexNum(api.xeth().NewTransactionFilter())
 	case "eth_uninstallFilter":
 		args := new(FilterIdArgs)
 		if err := json.Unmarshal(req.Params, &args); err != nil {
@@ -341,7 +340,17 @@ func (api *EthereumApi) GetRequestReply(req *RpcRequest, reply *interface{}) err
 		if err := json.Unmarshal(req.Params, &args); err != nil {
 			return err
 		}
-		*reply = NewLogsRes(api.xeth().FilterChanged(args.Id))
+
+		switch api.xeth().GetFilterType(args.Id) {
+		case xeth.BlockFilterTy:
+			*reply = NewHashesRes(api.xeth().BlockFilterChanged(args.Id))
+		case xeth.TransactionFilterTy:
+			*reply = NewHashesRes(api.xeth().TransactionFilterChanged(args.Id))
+		case xeth.LogFilterTy:
+			*reply = NewLogsRes(api.xeth().LogFilterChanged(args.Id))
+		default:
+			*reply = []string{} // reply empty string slice
+		}
 	case "eth_getFilterLogs":
 		args := new(FilterIdArgs)
 		if err := json.Unmarshal(req.Params, &args); err != nil {

rpc/responses.go

@@ -3,6 +3,7 @@ package rpc
 import (
 	"encoding/json"
 
+	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/state"
 	"github.com/ethereum/go-ethereum/core/types"
 )
@@ -303,3 +304,13 @@ func NewLogsRes(logs state.Logs) (ls []LogRes) {
 
 	return
 }
+
+func NewHashesRes(hs []common.Hash) []string {
+	hashes := make([]string, len(hs))
+
+	for i, hash := range hs {
+		hashes[i] = hash.Hex()
+	}
+
+	return hashes
+}