manager.SubProtocols = append(manager.SubProtocols, p2p.Protocol{
...
Run: func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
peer := manager.newPeer(int(version), p, rw)
select {
case manager.newPeerCh <- peer:
manager.wg.Add(1)
defer manager.wg.Done()
return manager.handle(peer)
case <-manager.quitSync:
return p2p.DiscQuitting
}
},
...Last updated
func (pm *ProtocolManager) handle(p *peer) error {
...
for {
if err := pm.handleMsg(p); err != nil {
p.Log().Debug("Ethereum message handling failed", "err", err)
return err
}
}
}func (pm *ProtocolManager) handleMsg(p *peer) error {
// Read the next message from the remote peer, and ensure it's fully consumed
msg, err := p.rw.ReadMsg()
...
// Handle the message depending on its contents
switch {
case msg.Code == StatusMsg:
...
// Block header query, collect the requested headers and reply
case msg.Code == GetBlockHeadersMsg:
...
case msg.Code == BlockHeadersMsg:
...
case msg.Code == GetBlockBodiesMsg:
...
case msg.Code == BlockBodiesMsg:
...
case msg.Code == TxMsg:
...
// Transactions can be processed, parse all of them and deliver to the pool
var txs []*types.Transaction
if err := msg.Decode(&txs); err != nil {
return errResp(ErrDecode, "msg %v: %v", msg, err)
}
for i, tx := range txs {
// Validate and mark the remote transaction
if tx == nil {
return errResp(ErrDecode, "transaction %d is nil", i)
}
p.MarkTransaction(tx.Hash())
}
pm.txpool.AddBatch(txs)
default:
return errResp(ErrInvalidMsgCode, "%v", msg.Code)
}
return nil
}