eth, whisper: adapt for RLP encoder switch in package p2p

I have rewritten the protocol test to use p2p.MsgPipe because
p2p.NewMsg is gone.

5 files changed, 124 insertions, 175 deletions

diff --git a/eth/backend.go b/eth/backend.go
index 9e5324688..afe314d74 100644
--- a/eth/backend.go
+++ b/eth/backend.go
@@ -405,7 +405,7 @@ func (self *Ethereum) txBroadcastLoop() {
 	// automatically stops if unsubscribe
 	for obj := range self.txSub.Chan() {
 		event := obj.(core.TxPreEvent)
-		self.net.Broadcast("eth", TxMsg, event.Tx.RlpData())
+		self.net.Broadcast("eth", TxMsg, []*types.Transaction{event.Tx})
 	}
 }
 
@@ -414,7 +414,7 @@ func (self *Ethereum) blockBroadcastLoop() {
 	for obj := range self.blockSub.Chan() {
 		switch ev := obj.(type) {
 		case core.NewMinedBlockEvent:
-			self.net.Broadcast("eth", NewBlockMsg, ev.Block.RlpData(), ev.Block.Td)
+			self.net.Broadcast("eth", NewBlockMsg, []interface{}{ev.Block, ev.Block.Td})
 		}
 	}
 }
diff --git a/eth/protocol.go b/eth/protocol.go
index a1c2ae688..6d610a663 100644
--- a/eth/protocol.go
+++ b/eth/protocol.go
@@ -89,7 +89,7 @@ type blockPool interface {
 	RemovePeer(peerId string)
 }
 
-// message structs used for rlp decoding
+// message structs used for RLP serialization
 type newBlockMsgData struct {
 	Block *types.Block
 	TD    *big.Int
@@ -100,6 +100,14 @@ type getBlockHashesMsgData struct {
 	Amount uint64
 }
 
+type statusMsgData struct {
+	ProtocolVersion uint32
+	NetworkId       uint32
+	TD              *big.Int
+	CurrentBlock    common.Hash
+	GenesisBlock    common.Hash
+}
+
 // main entrypoint, wrappers starting a server running the eth protocol
 // use this constructor to attach the protocol ("class") to server caps
 // the Dev p2p layer then runs the protocol instance on each peer
@@ -132,18 +140,25 @@ func runEthProtocol(protocolVersion, networkId int, txPool txPool, chainManager
 		},
 		id: fmt.Sprintf("%x", id[:8]),
 	}
-	err = self.handleStatus()
-	if err == nil {
-		self.propagateTxs()
-		for {
-			err = self.handle()
-			if err != nil {
-				self.blockPool.RemovePeer(self.id)
-				break
-			}
+
+	// handshake.
+	if err := self.handleStatus(); err != nil {
+		return err
+	}
+	defer self.blockPool.RemovePeer(self.id)
+
+	// propagate existing transactions. new transactions appearing
+	// after this will be sent via broadcasts.
+	if err := p2p.Send(rw, TxMsg, txPool.GetTransactions()); err != nil {
+		return err
+	}
+
+	// main loop. handle incoming messages.
+	for {
+		if err := self.handle(); err != nil {
+			return err
 		}
 	}
-	return
 }
 
 func (self *ethProtocol) handle() error {
@@ -186,7 +201,7 @@ func (self *ethProtocol) handle() error {
 			request.Amount = maxHashes
 		}
 		hashes := self.chainManager.GetBlockHashesFromHash(request.Hash, request.Amount)
-		return p2p.EncodeMsg(self.rw, BlockHashesMsg, rlp.Flat(hashes))
+		return p2p.Send(self.rw, BlockHashesMsg, hashes)
 
 	case BlockHashesMsg:
 		msgStream := rlp.NewStream(msg.Payload)
@@ -216,7 +231,7 @@ func (self *ethProtocol) handle() error {
 			return err
 		}
 
-		var blocks []interface{}
+		var blocks []*types.Block
 		var i int
 		for {
 			i++
@@ -224,7 +239,7 @@ func (self *ethProtocol) handle() error {
 			err := msgStream.Decode(&hash)
 			if err == rlp.EOL {
 				break
-			} else {
+			} else if err != nil {
 				return self.protoError(ErrDecode, "msg %v: %v", msg, err)
 			}
 
@@ -236,7 +251,7 @@ func (self *ethProtocol) handle() error {
 				break
 			}
 		}
-		return p2p.EncodeMsg(self.rw, BlocksMsg, blocks...)
+		return p2p.Send(self.rw, BlocksMsg, blocks)
 
 	case BlocksMsg:
 		msgStream := rlp.NewStream(msg.Payload)
@@ -283,29 +298,8 @@ func (self *ethProtocol) handle() error {
 	return nil
 }
 
-type statusMsgData struct {
-	ProtocolVersion uint32
-	NetworkId       uint32
-	TD              *big.Int
-	CurrentBlock    common.Hash
-	GenesisBlock    common.Hash
-}
-
-func (self *ethProtocol) statusMsg() p2p.Msg {
-	td, currentBlock, genesisBlock := self.chainManager.Status()
-
-	return p2p.NewMsg(StatusMsg,
-		uint32(self.protocolVersion),
-		uint32(self.networkId),
-		td,
-		currentBlock,
-		genesisBlock,
-	)
-}
-
 func (self *ethProtocol) handleStatus() error {
-	// send precanned status message
-	if err := self.rw.WriteMsg(self.statusMsg()); err != nil {
+	if err := self.sendStatus(); err != nil {
 		return err
 	}
 
@@ -314,11 +308,9 @@ func (self *ethProtocol) handleStatus() error {
 	if err != nil {
 		return err
 	}
-
 	if msg.Code != StatusMsg {
 		return self.protoError(ErrNoStatusMsg, "first msg has code %x (!= %x)", msg.Code, StatusMsg)
 	}
-
 	if msg.Size > ProtocolMaxMsgSize {
 		return self.protoError(ErrMsgTooLarge, "%v > %v", msg.Size, ProtocolMaxMsgSize)
 	}
@@ -351,12 +343,12 @@ func (self *ethProtocol) handleStatus() error {
 
 func (self *ethProtocol) requestBlockHashes(from common.Hash) error {
 	self.peer.Debugf("fetching hashes (%d) %x...\n", maxHashes, from[0:4])
-	return p2p.EncodeMsg(self.rw, GetBlockHashesMsg, interface{}(from), uint64(maxHashes))
+	return p2p.Send(self.rw, GetBlockHashesMsg, getBlockHashesMsgData{from, maxHashes})
 }
 
 func (self *ethProtocol) requestBlocks(hashes []common.Hash) error {
 	self.peer.Debugf("fetching %v blocks", len(hashes))
-	return p2p.EncodeMsg(self.rw, GetBlocksMsg, rlp.Flat(hashes))
+	return p2p.Send(self.rw, GetBlocksMsg, hashes)
 }
 
 func (self *ethProtocol) protoError(code int, format string, params ...interface{}) (err *errs.Error) {
@@ -365,19 +357,20 @@ func (self *ethProtocol) protoError(code int, format string, params ...interface
 	return
 }
 
+func (self *ethProtocol) sendStatus() error {
+	td, currentBlock, genesisBlock := self.chainManager.Status()
+	return p2p.Send(self.rw, StatusMsg, &statusMsgData{
+		ProtocolVersion: uint32(self.protocolVersion),
+		NetworkId:       uint32(self.networkId),
+		TD:              td,
+		CurrentBlock:    currentBlock,
+		GenesisBlock:    genesisBlock,
+	})
+}
+
 func (self *ethProtocol) protoErrorDisconnect(err *errs.Error) {
 	err.Log(self.peer.Logger)
 	if err.Fatal() {
 		self.peer.Disconnect(p2p.DiscSubprotocolError)
 	}
 }
-
-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...))
-}
diff --git a/eth/protocol_test.go b/eth/protocol_test.go
index 108fb4475..7620b3854 100644
--- a/eth/protocol_test.go
+++ b/eth/protocol_test.go
@@ -1,8 +1,6 @@
 package eth
 
 import (
-	"bytes"
-	"io"
 	"log"
 	"math/big"
 	"os"
@@ -29,52 +27,21 @@ func logInit() {
 	}
 }
 
-type testMsgReadWriter struct {
-	in  chan p2p.Msg
-	out []p2p.Msg
-}
-
-func (self *testMsgReadWriter) In(msg p2p.Msg) {
-	self.in <- msg
-}
-
-func (self *testMsgReadWriter) Out() (msg p2p.Msg, ok bool) {
-	if len(self.out) > 0 {
-		msg = self.out[0]
-		self.out = self.out[1:]
-		ok = true
-	}
-	return
-}
-
-func (self *testMsgReadWriter) WriteMsg(msg p2p.Msg) error {
-	self.out = append(self.out, msg)
-	return nil
-}
-
-func (self *testMsgReadWriter) ReadMsg() (p2p.Msg, error) {
-	msg, ok := <-self.in
-	if !ok {
-		return msg, io.EOF
-	}
-	return msg, nil
-}
-
 type testTxPool struct {
 	getTransactions func() []*types.Transaction
 	addTransactions func(txs []*types.Transaction)
 }
 
 type testChainManager struct {
-	getBlockHashes func(hash []byte, amount uint64) (hashes [][]byte)
-	getBlock       func(hash []byte) *types.Block
-	status         func() (td *big.Int, currentBlock []byte, genesisBlock []byte)
+	getBlockHashes func(hash common.Hash, amount uint64) (hashes []common.Hash)
+	getBlock       func(hash common.Hash) *types.Block
+	status         func() (td *big.Int, currentBlock common.Hash, genesisBlock common.Hash)
 }
 
 type testBlockPool struct {
-	addBlockHashes func(next func() ([]byte, bool), peerId string)
+	addBlockHashes func(next func() (common.Hash, bool), peerId string)
 	addBlock       func(block *types.Block, peerId string) (err error)
-	addPeer        func(td *big.Int, currentBlock []byte, peerId string, requestHashes func([]byte) error, requestBlocks func([][]byte) error, peerError func(*errs.Error)) (best bool)
+	addPeer        func(td *big.Int, currentBlock common.Hash, peerId string, requestHashes func(common.Hash) error, requestBlocks func([]common.Hash) error, peerError func(*errs.Error)) (best bool)
 	removePeer     func(peerId string)
 }
 
@@ -93,28 +60,28 @@ func (self *testTxPool) AddTransactions(txs []*types.Transaction) {
 
 func (self *testTxPool) GetTransactions() types.Transactions { return nil }
 
-func (self *testChainManager) GetBlockHashesFromHash(hash []byte, amount uint64) (hashes [][]byte) {
+func (self *testChainManager) GetBlockHashesFromHash(hash common.Hash, amount uint64) (hashes []common.Hash) {
 	if self.getBlockHashes != nil {
 		hashes = self.getBlockHashes(hash, amount)
 	}
 	return
 }
 
-func (self *testChainManager) Status() (td *big.Int, currentBlock []byte, genesisBlock []byte) {
+func (self *testChainManager) Status() (td *big.Int, currentBlock common.Hash, genesisBlock common.Hash) {
 	if self.status != nil {
 		td, currentBlock, genesisBlock = self.status()
 	}
 	return
 }
 
-func (self *testChainManager) GetBlock(hash []byte) (block *types.Block) {
+func (self *testChainManager) GetBlock(hash common.Hash) (block *types.Block) {
 	if self.getBlock != nil {
 		block = self.getBlock(hash)
 	}
 	return
 }
 
-func (self *testBlockPool) AddBlockHashes(next func() ([]byte, bool), peerId string) {
+func (self *testBlockPool) AddBlockHashes(next func() (common.Hash, bool), peerId string) {
 	if self.addBlockHashes != nil {
 		self.addBlockHashes(next, peerId)
 	}
@@ -126,7 +93,7 @@ func (self *testBlockPool) AddBlock(block *types.Block, peerId string) {
 	}
 }
 
-func (self *testBlockPool) AddPeer(td *big.Int, currentBlock []byte, peerId string, requestBlockHashes func([]byte) error, requestBlocks func([][]byte) error, peerError func(*errs.Error)) (best bool) {
+func (self *testBlockPool) AddPeer(td *big.Int, currentBlock common.Hash, peerId string, requestBlockHashes func(common.Hash) error, requestBlocks func([]common.Hash) error, peerError func(*errs.Error)) (best bool) {
 	if self.addPeer != nil {
 		best = self.addPeer(td, currentBlock, peerId, requestBlockHashes, requestBlocks, peerError)
 	}
@@ -147,28 +114,36 @@ func testPeer() *p2p.Peer {
 }
 
 type ethProtocolTester struct {
+	p2p.MsgReadWriter // writing to the tester feeds the protocol
+
 	quit         chan error
-	rw           *testMsgReadWriter // p2p.MsgReadWriter
-	txPool       *testTxPool        // txPool
-	chainManager *testChainManager  // chainManager
-	blockPool    *testBlockPool     // blockPool
+	pipe         *p2p.MsgPipeRW    // the protocol read/writes on this end
+	txPool       *testTxPool       // txPool
+	chainManager *testChainManager // chainManager
+	blockPool    *testBlockPool    // blockPool
 	t            *testing.T
 }
 
 func newEth(t *testing.T) *ethProtocolTester {
+	p1, p2 := p2p.MsgPipe()
 	return &ethProtocolTester{
-		quit:         make(chan error),
-		rw:           &testMsgReadWriter{in: make(chan p2p.Msg, 10)},
-		txPool:       &testTxPool{},
-		chainManager: &testChainManager{},
-		blockPool:    &testBlockPool{},
-		t:            t,
+		MsgReadWriter: p1,
+		quit:          make(chan error, 1),
+		pipe:          p2,
+		txPool:        &testTxPool{},
+		chainManager:  &testChainManager{},
+		blockPool:     &testBlockPool{},
+		t:             t,
 	}
 }
 
 func (self *ethProtocolTester) reset() {
-	self.rw = &testMsgReadWriter{in: make(chan p2p.Msg, 10)}
-	self.quit = make(chan error)
+	self.pipe.Close()
+
+	p1, p2 := p2p.MsgPipe()
+	self.MsgReadWriter = p1
+	self.pipe = p2
+	self.quit = make(chan error, 1)
 }
 
 func (self *ethProtocolTester) checkError(expCode int, delay time.Duration) (err error) {
@@ -190,33 +165,8 @@ func (self *ethProtocolTester) checkError(expCode int, delay time.Duration) (err
 	return
 }
 
-func (self *ethProtocolTester) In(msg p2p.Msg) {
-	self.rw.In(msg)
-}
-
-func (self *ethProtocolTester) Out() (p2p.Msg, bool) {
-	return self.rw.Out()
-}
-
-func (self *ethProtocolTester) checkMsg(i int, code uint64, val interface{}) (msg p2p.Msg) {
-	if i >= len(self.rw.out) {
-		self.t.Errorf("expected at least %v msgs, got %v", i, len(self.rw.out))
-		return
-	}
-	msg = self.rw.out[i]
-	if msg.Code != code {
-		self.t.Errorf("expected msg code %v, got %v", code, msg.Code)
-	}
-	if val != nil {
-		if err := msg.Decode(val); err != nil {
-			self.t.Errorf("rlp encoding error: %v", err)
-		}
-	}
-	return
-}
-
 func (self *ethProtocolTester) run() {
-	err := runEthProtocol(ProtocolVersion, NetworkId, self.txPool, self.chainManager, self.blockPool, testPeer(), self.rw)
+	err := runEthProtocol(ProtocolVersion, NetworkId, self.txPool, self.chainManager, self.blockPool, testPeer(), self.pipe)
 	self.quit <- err
 }
 
@@ -224,41 +174,52 @@ func TestStatusMsgErrors(t *testing.T) {
 	logInit()
 	eth := newEth(t)
 	td := common.Big1
-	currentBlock := []byte{1}
-	genesis := []byte{2}
-	eth.chainManager.status = func() (*big.Int, []byte, []byte) { return td, currentBlock, genesis }
-	go eth.run()
-	statusMsg := p2p.NewMsg(4)
-	eth.In(statusMsg)
-	delay := 1 * time.Second
-	eth.checkError(ErrNoStatusMsg, delay)
-	var status statusMsgData
-	eth.checkMsg(0, StatusMsg, &status) // first outgoing msg should be StatusMsg
-	if status.TD.Cmp(td) != 0 ||
-		status.ProtocolVersion != ProtocolVersion ||
-		status.NetworkId != NetworkId ||
-		status.TD.Cmp(td) != 0 ||
-		bytes.Compare(status.CurrentBlock, currentBlock) != 0 ||
-		bytes.Compare(status.GenesisBlock, genesis) != 0 {
-		t.Errorf("incorrect outgoing status")
-	}
-
-	eth.reset()
+	currentBlock := common.Hash{1}
+	genesis := common.Hash{2}
+	eth.chainManager.status = func() (*big.Int, common.Hash, common.Hash) { return td, currentBlock, genesis }
 	go eth.run()
-	statusMsg = p2p.NewMsg(0, uint32(48), uint32(0), td, currentBlock, genesis)
-	eth.In(statusMsg)
-	eth.checkError(ErrProtocolVersionMismatch, delay)
 
-	eth.reset()
-	go eth.run()
-	statusMsg = p2p.NewMsg(0, uint32(49), uint32(1), td, currentBlock, genesis)
-	eth.In(statusMsg)
-	eth.checkError(ErrNetworkIdMismatch, delay)
+	tests := []struct {
+		code          uint64
+		data          interface{}
+		wantErrorCode int
+	}{
+		{
+			code: TxMsg, data: []interface{}{},
+			wantErrorCode: ErrNoStatusMsg,
+		},
+		{
+			code: StatusMsg, data: statusMsgData{10, NetworkId, td, currentBlock, genesis},
+			wantErrorCode: ErrProtocolVersionMismatch,
+		},
+		{
+			code: StatusMsg, data: statusMsgData{ProtocolVersion, 999, td, currentBlock, genesis},
+			wantErrorCode: ErrNetworkIdMismatch,
+		},
+		{
+			code: StatusMsg, data: statusMsgData{ProtocolVersion, NetworkId, td, currentBlock, common.Hash{3}},
+			wantErrorCode: ErrGenesisBlockMismatch,
+		},
+	}
+	for _, test := range tests {
+		// first outgoing msg should be StatusMsg.
+		err := p2p.ExpectMsg(eth, StatusMsg, &statusMsgData{
+			ProtocolVersion: ProtocolVersion,
+			NetworkId:       NetworkId,
+			TD:              td,
+			CurrentBlock:    currentBlock,
+			GenesisBlock:    genesis,
+		})
+		if err != nil {
+			t.Fatalf("incorrect outgoing status: %v", err)
+		}
 
-	eth.reset()
-	go eth.run()
-	statusMsg = p2p.NewMsg(0, uint32(49), uint32(0), td, currentBlock, []byte{3})
-	eth.In(statusMsg)
-	eth.checkError(ErrGenesisBlockMismatch, delay)
+		// the send call might hang until reset because
+		// the protocol might not read the payload.
+		go p2p.Send(eth, test.code, test.data)
+		eth.checkError(test.wantErrorCode, 1*time.Second)
 
+		eth.reset()
+		go eth.run()
+	}
 }
diff --git a/whisper/envelope.go b/whisper/envelope.go
index 577638046..9ec9fc318 100644
--- a/whisper/envelope.go
+++ b/whisper/envelope.go
@@ -6,9 +6,9 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/crypto/ecies"
-	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/rlp"
 )
 
diff --git a/whisper/peer.go b/whisper/peer.go
index 4bd9428f5..ee5bffd0b 100644
--- a/whisper/peer.go
+++ b/whisper/peer.go
@@ -77,12 +77,11 @@ func (self *peer) broadcast(envelopes []*Envelope) error {
 	}
 
 	if i > 0 {
-		if err := p2p.EncodeMsg(self.ws, envelopesMsg, envs[:i]...); err != nil {
+		if err := p2p.Send(self.ws, envelopesMsg, envs[:i]); err != nil {
 			return err
 		}
 		self.peer.DebugDetailln("broadcasted", i, "message(s)")
 	}
-
 	return nil
 }
 
@@ -92,7 +91,7 @@ func (self *peer) addKnown(envelope *Envelope) {
 
 func (self *peer) handleStatus() error {
 	ws := self.ws
-	if err := ws.WriteMsg(self.statusMsg()); err != nil {
+	if err := p2p.SendItems(ws, statusMsg, protocolVersion); err != nil {
 		return err
 	}
 	msg, err := ws.ReadMsg()
@@ -115,7 +114,3 @@ func (self *peer) handleStatus() error {
 	}
 	return msg.Discard() // ignore anything after protocol version
 }
-
-func (self *peer) statusMsg() p2p.Msg {
-	return p2p.NewMsg(statusMsg, protocolVersion)
-}