initial commit for eth-p2p integration

2 files changed, 165 insertions, 1 deletions

diff --git a/eth/protocol.go b/eth/protocol.go
index 3b5b49696..985b42901 100644
--- a/eth/protocol.go
+++ b/eth/protocol.go
@@ -20,7 +20,7 @@ type ethProtocol struct {
 	peer         *p2p.Peer
 	id           string
 	rw           p2p.MsgReadWriter
-}
+	}
 
 // backend is the interface the ethereum protocol backend should implement
 // used as an argument to EthProtocol
@@ -68,6 +68,7 @@ type newBlockMsgData struct {
 
 type getBlockHashesMsgData struct {
 	Hash   []byte
+<<<<<<< HEAD
 	Amount uint64
 }
 
@@ -76,15 +77,29 @@ type getBlockHashesMsgData struct {
 // the Dev p2p layer then runs the protocol instance on each peer
 func EthProtocol(txPool txPool, chainManager chainManager, blockPool blockPool) p2p.Protocol {
 	return p2p.Protocol{
+=======
+	Amount uint32
+}
+
+// main entrypoint, wrappers starting a server running the eth protocol
+// use this constructor to attach the protocol (class) to server caps
+func EthProtocol(eth backend) *p2p.Protocol {
+	return &p2p.Protocol{
+>>>>>>> initial commit for eth-p2p integration
 		Name:    "eth",
 		Version: ProtocolVersion,
 		Length:  ProtocolLength,
 		Run: func(peer *p2p.Peer, rw p2p.MsgReadWriter) error {
+<<<<<<< HEAD
 			return runEthProtocol(txPool, chainManager, blockPool, peer, rw)
+=======
+			return runEthProtocol(eth, peer, rw)
+>>>>>>> initial commit for eth-p2p integration
 		},
 	}
 }
 
+<<<<<<< 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) {
@@ -95,6 +110,13 @@ func runEthProtocol(txPool txPool, chainManager chainManager, blockPool blockPoo
 		rw:           rw,
 		peer:         peer,
 		id:           (string)(peer.Identity().Pubkey()),
+=======
+func runEthProtocol(eth backend, peer *p2p.Peer, rw p2p.MsgReadWriter) (err error) {
+	self := &ethProtocol{
+		eth:  eth,
+		rw:   rw,
+		peer: peer,
+>>>>>>> initial commit for eth-p2p integration
 	}
 	err = self.handleStatus()
 	if err == nil {
@@ -102,7 +124,10 @@ func runEthProtocol(txPool txPool, chainManager chainManager, blockPool blockPoo
 			for {
 				err = self.handle()
 				if err != nil {
+<<<<<<< HEAD
 					self.blockPool.RemovePeer(self.id)
+=======
+>>>>>>> initial commit for eth-p2p integration
 					break
 				}
 			}
@@ -127,24 +152,46 @@ func (self *ethProtocol) handle() error {
 	case StatusMsg:
 		return ProtocolError(ErrExtraStatusMsg, "")
 
+<<<<<<< HEAD
 	case TxMsg:
 		// TODO: rework using lazy RLP stream
+=======
+	case GetTxMsg:
+		txs := self.eth.GetTransactions()
+		// TODO: rewrite using rlp flat
+		txsInterface := make([]interface{}, len(txs))
+		for i, tx := range txs {
+			txsInterface[i] = tx.RlpData()
+		}
+		return self.rw.EncodeMsg(TxMsg, txsInterface...)
+
+	case TxMsg:
+>>>>>>> initial commit for eth-p2p integration
 		var txs []*types.Transaction
 		if err := msg.Decode(&txs); err != nil {
 			return ProtocolError(ErrDecode, "%v", err)
 		}
+<<<<<<< HEAD
 		self.txPool.AddTransactions(txs)
+=======
+		self.eth.AddTransactions(txs)
+>>>>>>> initial commit for eth-p2p integration
 
 	case GetBlockHashesMsg:
 		var request getBlockHashesMsgData
 		if err := msg.Decode(&request); err != nil {
 			return ProtocolError(ErrDecode, "%v", err)
 		}
+<<<<<<< HEAD
 		hashes := self.chainManager.GetBlockHashesFromHash(request.Hash, request.Amount)
+=======
+		hashes := self.eth.GetBlockHashes(request.Hash, request.Amount)
+>>>>>>> initial commit for eth-p2p integration
 		return self.rw.EncodeMsg(BlockHashesMsg, ethutil.ByteSliceToInterface(hashes)...)
 
 	case BlockHashesMsg:
 		// TODO: redo using lazy decode , this way very inefficient on known chains
+<<<<<<< HEAD
 		msgStream := rlp.NewListStream(msg.Payload, uint64(msg.Size))
 		var err error
 		iter := func() (hash []byte, ok bool) {
@@ -160,17 +207,45 @@ func (self *ethProtocol) handle() error {
 		}
 
 	case GetBlocksMsg:
+=======
+		// s := rlp.NewListStream(msg.Payload, uint64(msg.Size))
 		var blockHashes [][]byte
 		if err := msg.Decode(&blockHashes); err != nil {
 			return ProtocolError(ErrDecode, "%v", err)
 		}
+		fetchMore := true
+		for _, hash := range blockHashes {
+			fetchMore = self.eth.AddHash(hash, self.peer)
+			if !fetchMore {
+				break
+			}
+		}
+		if fetchMore {
+			return self.FetchHashes(blockHashes[len(blockHashes)-1])
+		}
+
+	case GetBlocksMsg:
+		// Limit to max 300 blocks
+>>>>>>> initial commit for eth-p2p integration
+		var blockHashes [][]byte
+		if err := msg.Decode(&blockHashes); err != nil {
+			return ProtocolError(ErrDecode, "%v", err)
+		}
+<<<<<<< HEAD
 		max := int(math.Min(float64(len(blockHashes)), blockHashesBatchSize))
+=======
+		max := int(math.Min(float64(len(blockHashes)), 300.0))
+>>>>>>> initial commit for eth-p2p integration
 		var blocks []interface{}
 		for i, hash := range blockHashes {
 			if i >= max {
 				break
 			}
+<<<<<<< HEAD
 			block := self.chainManager.GetBlock(hash)
+=======
+			block := self.eth.GetBlock(hash)
+>>>>>>> initial commit for eth-p2p integration
 			if block != nil {
 				blocks = append(blocks, block.Value().Raw())
 			}
@@ -178,6 +253,7 @@ func (self *ethProtocol) handle() error {
 		return self.rw.EncodeMsg(BlocksMsg, blocks...)
 
 	case BlocksMsg:
+<<<<<<< HEAD
 		msgStream := rlp.NewListStream(msg.Payload, uint64(msg.Size))
 		for {
 			var block *types.Block
@@ -189,6 +265,22 @@ func (self *ethProtocol) handle() error {
 				}
 			}
 			self.blockPool.AddBlock(block, self.id)
+=======
+		var blocks []*types.Block
+		if err := msg.Decode(&blocks); err != nil {
+			return ProtocolError(ErrDecode, "%v", err)
+		}
+		for _, block := range blocks {
+			fetchHashes, err := self.eth.AddBlock(nil, block, self.peer)
+			if err != nil {
+				return ProtocolError(ErrInvalidBlock, "%v", err)
+			}
+			if fetchHashes {
+				if err := self.FetchHashes(block.Hash()); err != nil {
+					return err
+				}
+			}
+>>>>>>> initial commit for eth-p2p integration
 		}
 
 	case NewBlockMsg:
@@ -196,6 +288,7 @@ func (self *ethProtocol) handle() error {
 		if err := msg.Decode(&request); err != nil {
 			return ProtocolError(ErrDecode, "%v", err)
 		}
+<<<<<<< HEAD
 		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
@@ -212,6 +305,15 @@ func (self *ethProtocol) handle() error {
 			}
 			self.blockPool.AddBlockHashes(iter, self.id)
 			self.blockPool.AddBlock(request.Block, self.id)
+=======
+		var fetchHashes bool
+		// this should reset td and offer blockpool as candidate new peer?
+		if fetchHashes, err = self.eth.AddBlock(request.TD, request.Block, self.peer); err != nil {
+			return ProtocolError(ErrInvalidBlock, "%v", err)
+		}
+		if fetchHashes {
+			return self.FetchHashes(request.Block.Hash())
+>>>>>>> initial commit for eth-p2p integration
 		}
 
 	default:
@@ -229,7 +331,11 @@ type statusMsgData struct {
 }
 
 func (self *ethProtocol) statusMsg() p2p.Msg {
+<<<<<<< HEAD
 	td, currentBlock, genesisBlock := self.chainManager.Status()
+=======
+	td, currentBlock, genesisBlock := self.eth.Status()
+>>>>>>> initial commit for eth-p2p integration
 
 	return p2p.NewMsg(StatusMsg,
 		uint32(ProtocolVersion),
@@ -265,7 +371,11 @@ func (self *ethProtocol) handleStatus() error {
 		return ProtocolError(ErrDecode, "%v", err)
 	}
 
+<<<<<<< HEAD
 	_, _, genesisBlock := self.chainManager.Status()
+=======
+	_, _, genesisBlock := self.eth.Status()
+>>>>>>> initial commit for eth-p2p integration
 
 	if bytes.Compare(status.GenesisBlock, genesisBlock) != 0 {
 		return ProtocolError(ErrGenesisBlockMismatch, "%x (!= %x)", status.GenesisBlock, genesisBlock)
@@ -279,13 +389,22 @@ func (self *ethProtocol) handleStatus() error {
 		return ProtocolError(ErrProtocolVersionMismatch, "%d (!= %d)", status.ProtocolVersion, ProtocolVersion)
 	}
 
+<<<<<<< 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)
+=======
+	logger.Infof("Peer is [eth] capable (%d/%d). TD = %v ~ %x", status.ProtocolVersion, status.NetworkId, status.CurrentBlock)
+
+	if self.eth.AddPeer(status.TD, status.CurrentBlock, self.peer) {
+		return self.FetchHashes(status.CurrentBlock)
+	}
+>>>>>>> initial commit for eth-p2p integration
 
 	return nil
 }
 
+<<<<<<< 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)
@@ -316,3 +435,9 @@ func (self *ethProtocol) protoErrorDisconnect(code int, format string, params ..
 	}
 
 }
+=======
+func (self *ethProtocol) FetchHashes(from []byte) error {
+	logger.Debugf("Fetching hashes (%d) %x...\n", blockHashesBatchSize, from[0:4])
+	return self.rw.EncodeMsg(GetBlockHashesMsg, from, blockHashesBatchSize)
+}
+>>>>>>> initial commit for eth-p2p integration
diff --git a/eth/protocol_test.go b/eth/protocol_test.go
index a166ea6cd..93696213a 100644
--- a/eth/protocol_test.go
+++ b/eth/protocol_test.go
@@ -56,11 +56,18 @@ 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)
 }
 
@@ -84,12 +91,21 @@ func (self *TestBackend) GetBlockHashes(hash []byte, amount uint32) (hashes [][]
 	return
 }
 
+<<<<<<< HEAD
 func (self *TestBackend) AddBlockHashes(next func() ([]byte, bool), peerId string) {
 	if self.addBlockHashes != nil {
 		self.addBlockHashes(next, peerId)
 	}
 }
 
+=======
+func (self *TestBackend) AddHash(hash []byte, peer *p2p.Peer) (more bool) {
+	if self.addHash != nil {
+		more = self.addHash(hash, peer)
+	}
+	return
+}
+>>>>>>> initial commit for eth-p2p integration
 func (self *TestBackend) GetBlock(hash []byte) (block *types.Block) {
 	if self.getBlock != nil {
 		block = self.getBlock(hash)
@@ -97,26 +113,41 @@ func (self *TestBackend) GetBlock(hash []byte) (block *types.Block) {
 	return
 }
 
+<<<<<<< HEAD
 func (self *TestBackend) AddBlock(block *types.Block, peerId string) (err error) {
 	if self.addBlock != nil {
 		err = self.addBlock(block, peerId)
+=======
+func (self *TestBackend) AddBlock(td *big.Int, block *types.Block, peer *p2p.Peer) (fetchHashes bool, err error) {
+	if self.addBlock != nil {
+		fetchHashes, err = self.addBlock(td, block, peer)
+>>>>>>> initial commit for eth-p2p integration
 	}
 	return
 }
 
+<<<<<<< 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)
+=======
+func (self *TestBackend) AddPeer(td *big.Int, currentBlock []byte, peer *p2p.Peer) (fetchHashes bool) {
+	if self.addPeer != nil {
+		fetchHashes = self.addPeer(td, currentBlock, peer)
+>>>>>>> initial commit for eth-p2p integration
 	}
 	return
 }
 
+<<<<<<< HEAD
 func (self *TestBackend) RemovePeer(peerId string) {
 	if self.removePeer != nil {
 		self.removePeer(peerId)
 	}
 }
 
+=======
+>>>>>>> initial commit for eth-p2p integration
 func (self *TestBackend) Status() (td *big.Int, currentBlock []byte, genesisBlock []byte) {
 	if self.status != nil {
 		td, currentBlock, genesisBlock = self.status()
@@ -124,6 +155,7 @@ func (self *TestBackend) Status() (td *big.Int, currentBlock []byte, genesisBloc
 	return
 }
 
+<<<<<<< HEAD
 // TODO: refactor this into p2p/client_identity
 type peerId struct {
 	pubkey []byte
@@ -147,12 +179,19 @@ func testPeer() *p2p.Peer {
 }
 
 func TestErrNoStatusMsg(t *testing.T) {
+=======
+func TestEth(t *testing.T) {
+>>>>>>> initial commit for eth-p2p integration
 	quit := make(chan bool)
 	rw := &testMsgReadWriter{make(chan p2p.Msg, 10), make(chan p2p.Msg, 10)}
 	testBackend := &TestBackend{}
 	var err error
 	go func() {
+<<<<<<< HEAD
 		err = runEthProtocol(testBackend, testPeer(), rw)
+=======
+		err = runEthProtocol(testBackend, nil, rw)
+>>>>>>> initial commit for eth-p2p integration
 		close(quit)
 	}()
 	statusMsg := p2p.NewMsg(4)