Propagate known transactions to new peers on connect

cmd/mist/assets/examples/coin.html

@@ -57,8 +57,8 @@
 		"name":"Changed",
 		"type":"event",
 		"inputs": [
-			{"name":"to","type":"address","indexed":false},
+			{"name":"to","type":"address","indexed":true},
-			{"name":"amount","type":"uint256","indexed":true},
+			{"name":"amount","type":"uint256","indexed":false},
 		],
 	}];
 
@@ -74,12 +74,12 @@
     }
 
 	var contract = web3.eth.contract(address, desc);
-	contract.Changed({to: "0xaabb"}).changed(function(e) {
+	contract.Changed({to: "0xaa"}).changed(function(e) {
 		console.log("e: " + JSON.stringify(e));
 	});
 	contract.transact({gas: "10000", gasprice: eth.gasPrice}).send( "0xaa", 10000 );
 	function reflesh() {
-        document.querySelector("#balance").innerHTML = contract.call().balance(eth.coinbase);
+        document.querySelector("#balance").innerHTML = contract.balance(eth.coinbase);
 
 		var table = document.querySelector("#table");
 		table.innerHTML = ""; // clear

core/filter.go

@@ -2,7 +2,6 @@ package core
 
 import (
 	"bytes"
-	"fmt"
 	"math"
 
 	"github.com/ethereum/go-ethereum/core/types"
@@ -129,37 +128,33 @@ func (self *Filter) Find() state.Logs {
 	return logs[skip:]
 }
 
-func includes(addresses [][]byte, a []byte) (found bool) {
+func includes(addresses [][]byte, a []byte) bool {
 	for _, addr := range addresses {
-		fmt.Println("INCLUDES", addr, a)
+		if !bytes.Equal(addr, a) {
-		if bytes.Compare(addr, a) == 0 {
+			return false
-			return true
 		}
 	}
 
-	return
+	return true
 }
 
 func (self *Filter) FilterLogs(logs state.Logs) state.Logs {
-	fmt.Println("FILTER LOGS", self.topics)
 	var ret state.Logs
 
 	// Filter the logs for interesting stuff
+Logs:
 	for _, log := range logs {
-		fmt.Println(log)
-
 		if len(self.address) > 0 && !bytes.Equal(self.address, log.Address()) {
 			continue
 		}
 
-		for _, topic := range self.topics {
+		max := int(math.Min(float64(len(self.topics)), float64(len(log.Topics()))))
-			fmt.Println("TOPIC:", topic)
+		for i := 0; i < max; i++ {
-			if !includes(log.Topics(), topic) {
+			if !bytes.Equal(log.Topics()[i], self.topics[i]) {
-				continue
+				continue Logs
 			}
 		}
 
-		fmt.Println("APPENDED")
 		ret = append(ret, log)
 	}
 

eth/protocol.go

@@ -46,6 +46,7 @@ type ethProtocol struct {
 // used as an argument to EthProtocol
 type txPool interface {
 	AddTransactions([]*types.Transaction)
+	GetTransactions() types.Transactions
 }
 
 type chainManager interface {
@@ -101,6 +102,7 @@ func runEthProtocol(txPool txPool, chainManager chainManager, blockPool blockPoo
 	}
 	err = self.handleStatus()
 	if err == nil {
+		self.propagateTxs()
 		for {
 			err = self.handle()
 			if err != nil {
@@ -324,3 +326,13 @@ func (self *ethProtocol) protoErrorDisconnect(code int, format string, params ..
 	}
 
 }
+
+func (self *ethProtocol) propagateTxs() {
+	transactions := self.txPool.GetTransactions()
+	iface := make([]interface{}, len(transactions))
+	for i, transaction := range transactions {
+		iface[i] = transaction
+	}
+
+	self.rw.WriteMsg(p2p.NewMsg(TxMsg, iface...))
+}

rpc/args.go

@@ -214,7 +214,7 @@ type FilterOptions struct {
 	Earliest int64
 	Latest   int64
 	Address  string
-	Topics   []string
+	Topic    []string
 	Skip     int
 	Max      int
 }
@@ -224,8 +224,8 @@ func toFilterOptions(options *FilterOptions) core.FilterOptions {
 	opts.Earliest = options.Earliest
 	opts.Latest = options.Latest
 	opts.Address = fromHex(options.Address)
-	opts.Topics = make([][]byte, len(options.Topics))
+	opts.Topics = make([][]byte, len(options.Topic))
-	for i, topic := range options.Topics {
+	for i, topic := range options.Topic {
 		opts.Topics[i] = fromHex(topic)
 	}
 

rpc/message.go

@@ -206,7 +206,7 @@ func (req *RpcRequest) ToFilterArgs() (*FilterOptions, error) {
 	if len(req.Params) < 1 {
 		return nil, NewErrorResponse(ErrorArguments)
 	}
-	fmt.Println("filter params", req.Params)
+	fmt.Println("FILTER PARAMS", string(req.Params[0]))
 
 	args := new(FilterOptions)
 	r := bytes.NewReader(req.Params[0])

rpc/packages.go

@@ -70,6 +70,7 @@ func NewEthereumApi(eth *xeth.XEth) *EthereumApi {
 func (self *EthereumApi) NewFilter(args *FilterOptions, reply *interface{}) error {
 	var id int
 	filter := core.NewFilter(self.xeth.Backend())
+	filter.SetOptions(toFilterOptions(args))
 	filter.LogsCallback = func(logs state.Logs) {
 		self.logMut.Lock()
 		defer self.logMut.Unlock()