netsync/consensusmgr/consensus_msg.go

   1 package consensusmgr
   2
   3 import (
   4         "bytes"
   5         "encoding/hex"
   6         "errors"
   7         "fmt"
   8
   9         "github.com/tendermint/go-wire"
  10
  11         "github.com/bytom/bytom/netsync/peers"
  12         "github.com/bytom/bytom/protocol/bc"
  13         "github.com/bytom/bytom/protocol/bc/types"
  14 )
  15
  16 const (
  17         blockSignatureByte = byte(0x10)
  18         blockProposeByte   = byte(0x11)
  19 )
  20
  21 // ConsensusMessage is a generic message for consensus reactor.
  22 type ConsensusMessage interface {
  23         String() string
  24         BroadcastMarkSendRecord(ps *peers.PeerSet, peers []string)
  25         BroadcastFilterTargetPeers(ps *peers.PeerSet) []string
  26 }
  27
  28 var _ = wire.RegisterInterface(
  29         struct{ ConsensusMessage }{},
  30         wire.ConcreteType{O: &BlockVerificationMsg{}, Byte: blockSignatureByte},
  31         wire.ConcreteType{O: &BlockProposeMsg{}, Byte: blockProposeByte},
  32 )
  33
  34 // decodeMessage decode msg
  35 func decodeMessage(bz []byte) (msgType byte, msg ConsensusMessage, err error) {
  36         msgType = bz[0]
  37         n := int(0)
  38         r := bytes.NewReader(bz)
  39         msg = wire.ReadBinary(struct{ ConsensusMessage }{}, r, maxBlockchainResponseSize, &n, &err).(struct{ ConsensusMessage }).ConsensusMessage
  40         if err != nil && n != len(bz) {
  41                 err = errors.New("DecodeMessage() had bytes left over")
  42         }
  43         return
  44 }
  45
  46 // BlockVerificationMsg block verification message transferred between nodes.
  47 type BlockVerificationMsg struct {
  48         SourceHeight uint64
  49         SourceHash   bc.Hash
  50         TargetHeight uint64
  51         TargetHash   bc.Hash
  52         PubKey       []byte
  53         Signature    []byte
  54 }
  55
  56 // NewBlockVerificationMsg create new block verification msg.
  57 func NewBlockVerificationMsg(sourceHeight, targetHeight uint64, sourceHash, targetHash bc.Hash, pubKey, signature []byte) ConsensusMessage {
  58         return &BlockVerificationMsg{
  59                 SourceHeight: sourceHeight,
  60                 SourceHash:   sourceHash,
  61                 TargetHeight: targetHeight,
  62                 TargetHash:   targetHash,
  63                 PubKey:       pubKey,
  64                 Signature:    signature,
  65         }
  66 }
  67
  68 func (b *BlockVerificationMsg) String() string {
  69         return fmt.Sprintf("{sourceHeight:%d,targetHeight:%d,sourceHash:%s,targetHash:%s,signature:%s,pubkey:%s}",
  70                 b.SourceHeight, b.TargetHeight, b.SourceHash.String(), b.TargetHash.String(), hex.EncodeToString(b.Signature), hex.EncodeToString(b.PubKey[:]))
  71 }
  72
  73 // BroadcastMarkSendRecord mark send message record to prevent messages from being sent repeatedly.
  74 func (b *BlockVerificationMsg) BroadcastMarkSendRecord(ps *peers.PeerSet, peers []string) {
  75         for _, peer := range peers {
  76                 ps.MarkBlockVerification(peer, b.Signature)
  77         }
  78 }
  79
  80 // BroadcastFilterTargetPeers filter target peers to filter the nodes that need to send messages.
  81 func (b *BlockVerificationMsg) BroadcastFilterTargetPeers(ps *peers.PeerSet) []string {
  82         return ps.PeersWithoutSignature(b.Signature)
  83 }
  84
  85 // BlockProposeMsg block propose message transferred between nodes.
  86 type BlockProposeMsg struct {
  87         RawBlock []byte
  88 }
  89
  90 // NewBlockProposeMsg create new block propose msg.
  91 func NewBlockProposeMsg(block *types.Block) (ConsensusMessage, error) {
  92         rawBlock, err := block.MarshalText()
  93         if err != nil {
  94                 return nil, err
  95         }
  96         return &BlockProposeMsg{RawBlock: rawBlock}, nil
  97 }
  98
  99 // GetProposeBlock get propose block from msg.
 100 func (bp *BlockProposeMsg) GetProposeBlock() (*types.Block, error) {
 101         block := &types.Block{}
 102         if err := block.UnmarshalText(bp.RawBlock); err != nil {
 103                 return nil, err
 104         }
 105         return block, nil
 106 }
 107
 108 func (bp *BlockProposeMsg) String() string {
 109         block, err := bp.GetProposeBlock()
 110         if err != nil {
 111                 return "{err: wrong message}"
 112         }
 113         blockHash := block.Hash()
 114         return fmt.Sprintf("{block_height: %d, block_hash: %s}", block.Height, blockHash.String())
 115 }
 116
 117 // BroadcastMarkSendRecord mark send message record to prevent messages from being sent repeatedly.
 118 func (bp *BlockProposeMsg) BroadcastMarkSendRecord(ps *peers.PeerSet, peers []string) {
 119         block, err := bp.GetProposeBlock()
 120         if err != nil {
 121                 return
 122         }
 123
 124         hash := block.Hash()
 125         height := block.Height
 126         for _, peer := range peers {
 127                 ps.MarkBlock(peer, &hash)
 128                 ps.MarkStatus(peer, height)
 129         }
 130 }
 131
 132 // BroadcastFilterTargetPeers filter target peers to filter the nodes that need to send messages.
 133 func (bp *BlockProposeMsg) BroadcastFilterTargetPeers(ps *peers.PeerSet) []string {
 134         block, err := bp.GetProposeBlock()
 135         if err != nil {
 136                 return nil
 137         }
 138
 139         return ps.PeersWithoutBlock(block.Hash())
 140 }