eth protocol changes

- changed backend interface
- using callbacks for blockPool
- use rlp stream for lazy decoding
- use peer as logger
- add id (peer pubkey) to ethProtocol fields
- add testPeer to protocol test (temporary)

2 files changed, 132 insertions, 11 deletions

diff --git a/eth/protocol.go b/eth/protocol.go
index 985b42901..fbc4610ec 100644
--- a/eth/protocol.go
+++ b/eth/protocol.go
@@ -82,7 +82,8 @@ func EthProtocol(txPool txPool, chainManager chainManager, blockPool blockPool)
 }
 
 // main entrypoint, wrappers starting a server running the eth protocol
-// use this constructor to attach the protocol (class) to server caps
+// use this constructor to attach the protocol ("class") to server caps
+// the Dev p2p layer then runs the protocol instance on each peer
 func EthProtocol(eth backend) *p2p.Protocol {
 	return &p2p.Protocol{
 >>>>>>> initial commit for eth-p2p integration
@@ -100,6 +101,7 @@ func EthProtocol(eth backend) *p2p.Protocol {
 }
 
 <<<<<<< HEAD
+<<<<<<< HEAD
 // the main loop that handles incoming messages
 // note RemovePeer in the post-disconnect hook
 func runEthProtocol(txPool txPool, chainManager chainManager, blockPool blockPool, peer *p2p.Peer, rw p2p.MsgReadWriter) (err error) {
@@ -111,12 +113,20 @@ func runEthProtocol(txPool txPool, chainManager chainManager, blockPool blockPoo
 		peer:         peer,
 		id:           (string)(peer.Identity().Pubkey()),
 =======
+=======
+// the main loop that handles incoming messages
+// note RemovePeer in the post-disconnect hook
+>>>>>>> eth protocol changes
 func runEthProtocol(eth backend, peer *p2p.Peer, rw p2p.MsgReadWriter) (err error) {
 	self := &ethProtocol{
 		eth:  eth,
 		rw:   rw,
 		peer: peer,
+<<<<<<< HEAD
 >>>>>>> initial commit for eth-p2p integration
+=======
+		id:   (string)(peer.Identity().Pubkey()),
+>>>>>>> eth protocol changes
 	}
 	err = self.handleStatus()
 	if err == nil {
@@ -125,9 +135,13 @@ func runEthProtocol(eth backend, peer *p2p.Peer, rw p2p.MsgReadWriter) (err erro
 				err = self.handle()
 				if err != nil {
 <<<<<<< HEAD
+<<<<<<< HEAD
 					self.blockPool.RemovePeer(self.id)
 =======
 >>>>>>> initial commit for eth-p2p integration
+=======
+					self.eth.RemovePeer(self.id)
+>>>>>>> eth protocol changes
 					break
 				}
 			}
@@ -166,7 +180,11 @@ func (self *ethProtocol) handle() error {
 		return self.rw.EncodeMsg(TxMsg, txsInterface...)
 
 	case TxMsg:
+<<<<<<< HEAD
 >>>>>>> initial commit for eth-p2p integration
+=======
+		// TODO: rework using lazy RLP stream
+>>>>>>> eth protocol changes
 		var txs []*types.Transaction
 		if err := msg.Decode(&txs); err != nil {
 			return ProtocolError(ErrDecode, "%v", err)
@@ -192,12 +210,16 @@ func (self *ethProtocol) handle() error {
 	case BlockHashesMsg:
 		// TODO: redo using lazy decode , this way very inefficient on known chains
 <<<<<<< HEAD
+<<<<<<< HEAD
+=======
+>>>>>>> eth protocol changes
 		msgStream := rlp.NewListStream(msg.Payload, uint64(msg.Size))
 		var err error
 		iter := func() (hash []byte, ok bool) {
 			hash, err = msgStream.Bytes()
 			if err == nil {
 				ok = true
+<<<<<<< HEAD
 			}
 			return
 		}
@@ -218,24 +240,35 @@ func (self *ethProtocol) handle() error {
 			fetchMore = self.eth.AddHash(hash, self.peer)
 			if !fetchMore {
 				break
+=======
+>>>>>>> eth protocol changes
 			}
+			return
 		}
-		if fetchMore {
-			return self.FetchHashes(blockHashes[len(blockHashes)-1])
+		self.eth.AddBlockHashes(iter, self.id)
+		if err != nil && err != rlp.EOL {
+			return ProtocolError(ErrDecode, "%v", err)
 		}
 
 	case GetBlocksMsg:
+<<<<<<< HEAD
 		// Limit to max 300 blocks
 >>>>>>> initial commit for eth-p2p integration
+=======
+>>>>>>> eth protocol changes
 		var blockHashes [][]byte
 		if err := msg.Decode(&blockHashes); err != nil {
 			return ProtocolError(ErrDecode, "%v", err)
 		}
 <<<<<<< HEAD
+<<<<<<< HEAD
 		max := int(math.Min(float64(len(blockHashes)), blockHashesBatchSize))
 =======
 		max := int(math.Min(float64(len(blockHashes)), 300.0))
 >>>>>>> initial commit for eth-p2p integration
+=======
+		max := int(math.Min(float64(len(blockHashes)), blockHashesBatchSize))
+>>>>>>> eth protocol changes
 		var blocks []interface{}
 		for i, hash := range blockHashes {
 			if i >= max {
@@ -254,6 +287,9 @@ func (self *ethProtocol) handle() error {
 
 	case BlocksMsg:
 <<<<<<< HEAD
+<<<<<<< HEAD
+=======
+>>>>>>> eth protocol changes
 		msgStream := rlp.NewListStream(msg.Payload, uint64(msg.Size))
 		for {
 			var block *types.Block
@@ -262,6 +298,7 @@ func (self *ethProtocol) handle() error {
 					break
 				} else {
 					return ProtocolError(ErrDecode, "%v", err)
+<<<<<<< HEAD
 				}
 			}
 			self.blockPool.AddBlock(block, self.id)
@@ -281,6 +318,13 @@ func (self *ethProtocol) handle() error {
 				}
 			}
 >>>>>>> initial commit for eth-p2p integration
+=======
+				}
+			}
+			if err := self.eth.AddBlock(block, self.id); err != nil {
+				return ProtocolError(ErrInvalidBlock, "%v", err)
+			}
+>>>>>>> eth protocol changes
 		}
 
 	case NewBlockMsg:
@@ -289,11 +333,18 @@ func (self *ethProtocol) handle() error {
 			return ProtocolError(ErrDecode, "%v", err)
 		}
 <<<<<<< HEAD
+<<<<<<< HEAD
+=======
+>>>>>>> eth protocol changes
 		hash := request.Block.Hash()
 		// to simplify backend interface adding a new block
 		// uses AddPeer followed by AddHashes, AddBlock only if peer is the best peer
 		// (or selected as new best peer)
+<<<<<<< HEAD
 		if self.blockPool.AddPeer(request.TD, hash, self.id, self.requestBlockHashes, self.requestBlocks, self.protoErrorDisconnect) {
+=======
+		if self.eth.AddPeer(request.TD, hash, self.id, self.requestBlockHashes, self.requestBlocks, self.invalidBlock) {
+>>>>>>> eth protocol changes
 			called := true
 			iter := func() (hash []byte, ok bool) {
 				if called {
@@ -303,6 +354,7 @@ func (self *ethProtocol) handle() error {
 					return
 				}
 			}
+<<<<<<< HEAD
 			self.blockPool.AddBlockHashes(iter, self.id)
 			self.blockPool.AddBlock(request.Block, self.id)
 =======
@@ -314,6 +366,12 @@ func (self *ethProtocol) handle() error {
 		if fetchHashes {
 			return self.FetchHashes(request.Block.Hash())
 >>>>>>> initial commit for eth-p2p integration
+=======
+			self.eth.AddBlockHashes(iter, self.id)
+			if err := self.eth.AddBlock(request.Block, self.id); err != nil {
+				return ProtocolError(ErrInvalidBlock, "%v", err)
+			}
+>>>>>>> eth protocol changes
 		}
 
 	default:
@@ -390,6 +448,7 @@ func (self *ethProtocol) handleStatus() error {
 	}
 
 <<<<<<< HEAD
+<<<<<<< HEAD
 	self.peer.Infof("Peer is [eth] capable (%d/%d). TD = %v ~ %x", status.ProtocolVersion, status.NetworkId, status.CurrentBlock)
 
 	self.blockPool.AddPeer(status.TD, status.CurrentBlock, self.id, self.requestBlockHashes, self.requestBlocks, self.protoErrorDisconnect)
@@ -400,11 +459,17 @@ func (self *ethProtocol) handleStatus() error {
 		return self.FetchHashes(status.CurrentBlock)
 	}
 >>>>>>> initial commit for eth-p2p integration
+=======
+	self.peer.Infof("Peer is [eth] capable (%d/%d). TD = %v ~ %x", status.ProtocolVersion, status.NetworkId, status.CurrentBlock)
+
+	self.eth.AddPeer(status.TD, status.CurrentBlock, self.id, self.requestBlockHashes, self.requestBlocks, self.invalidBlock)
+>>>>>>> eth protocol changes
 
 	return nil
 }
 
 <<<<<<< HEAD
+<<<<<<< HEAD
 func (self *ethProtocol) requestBlockHashes(from []byte) error {
 	self.peer.Debugf("fetching hashes (%d) %x...\n", blockHashesBatchSize, from[0:4])
 	return self.rw.EncodeMsg(GetBlockHashesMsg, from, blockHashesBatchSize)
@@ -441,3 +506,29 @@ func (self *ethProtocol) FetchHashes(from []byte) error {
 	return self.rw.EncodeMsg(GetBlockHashesMsg, from, blockHashesBatchSize)
 }
 >>>>>>> initial commit for eth-p2p integration
+=======
+func (self *ethProtocol) requestBlockHashes(from []byte) error {
+	self.peer.Debugf("fetching hashes (%d) %x...\n", blockHashesBatchSize, from[0:4])
+	return self.rw.EncodeMsg(GetBlockHashesMsg, from, blockHashesBatchSize)
+}
+
+func (self *ethProtocol) requestBlocks(hashes [][]byte) error {
+	self.peer.Debugf("fetching %v blocks", len(hashes))
+	return self.rw.EncodeMsg(GetBlocksMsg, ethutil.ByteSliceToInterface(hashes))
+}
+
+func (self *ethProtocol) invalidBlock(err error) {
+	ProtocolError(ErrInvalidBlock, "%v", err)
+	self.peer.Disconnect(p2p.DiscSubprotocolError)
+}
+
+func (self *ethProtocol) protoError(code int, format string, params ...interface{}) (err *protocolError) {
+	err = ProtocolError(code, format, params...)
+	if err.Fatal() {
+		self.peer.Errorln(err)
+	} else {
+		self.peer.Debugln(err)
+	}
+	return
+}
+>>>>>>> eth protocol changes
diff --git a/eth/protocol_test.go b/eth/protocol_test.go
index 93696213a..322aec7b7 100644
--- a/eth/protocol_test.go
+++ b/eth/protocol_test.go
@@ -56,18 +56,11 @@ type TestBackend struct {
 	getTransactions func() []*types.Transaction
 	addTransactions func(txs []*types.Transaction)
 	getBlockHashes  func(hash []byte, amount uint32) (hashes [][]byte)
-<<<<<<< HEAD
 	addBlockHashes  func(next func() ([]byte, bool), peerId string)
 	getBlock        func(hash []byte) *types.Block
 	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, invalidBlock func(error)) (best bool)
 	removePeer      func(peerId string)
-=======
-	addHash         func(hash []byte, peer *p2p.Peer) (more bool)
-	getBlock        func(hash []byte) *types.Block
-	addBlock        func(td *big.Int, block *types.Block, peer *p2p.Peer) (fetchHashes bool, err error)
-	addPeer         func(td *big.Int, currentBlock []byte, peer *p2p.Peer) (fetchHashes bool)
->>>>>>> initial commit for eth-p2p integration
 	status          func() (td *big.Int, currentBlock []byte, genesisBlock []byte)
 }
 
@@ -92,6 +85,7 @@ func (self *TestBackend) GetBlockHashes(hash []byte, amount uint32) (hashes [][]
 }
 
 <<<<<<< HEAD
+<<<<<<< HEAD
 func (self *TestBackend) AddBlockHashes(next func() ([]byte, bool), peerId string) {
 	if self.addBlockHashes != nil {
 		self.addBlockHashes(next, peerId)
@@ -102,10 +96,18 @@ func (self *TestBackend) AddBlockHashes(next func() ([]byte, bool), peerId strin
 func (self *TestBackend) AddHash(hash []byte, peer *p2p.Peer) (more bool) {
 	if self.addHash != nil {
 		more = self.addHash(hash, peer)
+=======
+func (self *TestBackend) AddBlockHashes(next func() ([]byte, bool), peerId string) {
+	if self.addBlockHashes != nil {
+		self.addBlockHashes(next, peerId)
+>>>>>>> eth protocol changes
 	}
-	return
 }
+<<<<<<< HEAD
 >>>>>>> initial commit for eth-p2p integration
+=======
+
+>>>>>>> eth protocol changes
 func (self *TestBackend) GetBlock(hash []byte) (block *types.Block) {
 	if self.getBlock != nil {
 		block = self.getBlock(hash)
@@ -114,6 +116,7 @@ func (self *TestBackend) GetBlock(hash []byte) (block *types.Block) {
 }
 
 <<<<<<< HEAD
+<<<<<<< HEAD
 func (self *TestBackend) AddBlock(block *types.Block, peerId string) (err error) {
 	if self.addBlock != nil {
 		err = self.addBlock(block, peerId)
@@ -122,11 +125,17 @@ func (self *TestBackend) AddBlock(td *big.Int, block *types.Block, peer *p2p.Pee
 	if self.addBlock != nil {
 		fetchHashes, err = self.addBlock(td, block, peer)
 >>>>>>> initial commit for eth-p2p integration
+=======
+func (self *TestBackend) AddBlock(block *types.Block, peerId string) (err error) {
+	if self.addBlock != nil {
+		err = self.addBlock(block, peerId)
+>>>>>>> eth protocol changes
 	}
 	return
 }
 
 <<<<<<< HEAD
+<<<<<<< HEAD
 func (self *TestBackend) AddPeer(td *big.Int, currentBlock []byte, peerId string, requestBlockHashes func([]byte) error, requestBlocks func([][]byte) error, invalidBlock func(error)) (best bool) {
 	if self.addPeer != nil {
 		best = self.addPeer(td, currentBlock, peerId, requestBlockHashes, requestBlocks, invalidBlock)
@@ -135,19 +144,30 @@ func (self *TestBackend) AddPeer(td *big.Int, currentBlock []byte, peer *p2p.Pee
 	if self.addPeer != nil {
 		fetchHashes = self.addPeer(td, currentBlock, peer)
 >>>>>>> initial commit for eth-p2p integration
+=======
+func (self *TestBackend) AddPeer(td *big.Int, currentBlock []byte, peerId string, requestBlockHashes func([]byte) error, requestBlocks func([][]byte) error, invalidBlock func(error)) (best bool) {
+	if self.addPeer != nil {
+		best = self.addPeer(td, currentBlock, peerId, requestBlockHashes, requestBlocks, invalidBlock)
+>>>>>>> eth protocol changes
 	}
 	return
 }
 
 <<<<<<< HEAD
+<<<<<<< HEAD
+=======
+>>>>>>> eth protocol changes
 func (self *TestBackend) RemovePeer(peerId string) {
 	if self.removePeer != nil {
 		self.removePeer(peerId)
 	}
 }
 
+<<<<<<< HEAD
 =======
 >>>>>>> initial commit for eth-p2p integration
+=======
+>>>>>>> eth protocol changes
 func (self *TestBackend) Status() (td *big.Int, currentBlock []byte, genesisBlock []byte) {
 	if self.status != nil {
 		td, currentBlock, genesisBlock = self.status()
@@ -156,6 +176,9 @@ func (self *TestBackend) Status() (td *big.Int, currentBlock []byte, genesisBloc
 }
 
 <<<<<<< HEAD
+<<<<<<< HEAD
+=======
+>>>>>>> eth protocol changes
 // TODO: refactor this into p2p/client_identity
 type peerId struct {
 	pubkey []byte
@@ -179,19 +202,26 @@ func testPeer() *p2p.Peer {
 }
 
 func TestErrNoStatusMsg(t *testing.T) {
+<<<<<<< HEAD
 =======
 func TestEth(t *testing.T) {
 >>>>>>> initial commit for eth-p2p integration
+=======
+>>>>>>> eth protocol changes
 	quit := make(chan bool)
 	rw := &testMsgReadWriter{make(chan p2p.Msg, 10), make(chan p2p.Msg, 10)}
 	testBackend := &TestBackend{}
 	var err error
 	go func() {
 <<<<<<< HEAD
+<<<<<<< HEAD
 		err = runEthProtocol(testBackend, testPeer(), rw)
 =======
 		err = runEthProtocol(testBackend, nil, rw)
 >>>>>>> initial commit for eth-p2p integration
+=======
+		err = runEthProtocol(testBackend, testPeer(), rw)
+>>>>>>> eth protocol changes
 		close(quit)
 	}()
 	statusMsg := p2p.NewMsg(4)