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/vapor/netsync/peers"
  12         "github.com/vapor/protocol/bc"
  13         "github.com/vapor/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: &BlockSignatureMsg{}, 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 // BlockSignatureMsg block signature message transferred between nodes.
  47 type BlockSignatureMsg struct {
  48         BlockHash [32]byte
  49         Height    uint64
  50         Signature []byte
  51         PubKey    [64]byte
  52 }
  53
  54 //NewBlockSignatureMsg create new block signature msg.
  55 func NewBlockSignatureMsg(blockHash bc.Hash, height uint64, signature []byte, pubKey [64]byte) ConsensusMessage {
  56         hash := blockHash.Byte32()
  57         return &BlockSignatureMsg{BlockHash: hash, Height: height, Signature: signature, PubKey: pubKey}
  58 }
  59
  60 func (bs *BlockSignatureMsg) String() string {
  61         return fmt.Sprintf("{block_hash: %s,block_height:%d,signature:%s,pubkey:%s}", hex.EncodeToString(bs.BlockHash[:]), bs.Height, hex.EncodeToString(bs.Signature), hex.EncodeToString(bs.PubKey[:]))
  62 }
  63
  64 // BroadcastMarkSendRecord mark send message record to prevent messages from being sent repeatedly.
  65 func (bs *BlockSignatureMsg) BroadcastMarkSendRecord(ps *peers.PeerSet, peers []string) {
  66         for _, peer := range peers {
  67                 ps.MarkBlockSignature(peer, bs.Signature)
  68         }
  69 }
  70
  71 // BroadcastFilterTargetPeers filter target peers to filter the nodes that need to send messages.
  72 func (bs *BlockSignatureMsg) BroadcastFilterTargetPeers(ps *peers.PeerSet) []string {
  73         return ps.PeersWithoutSign(bs.Signature)
  74 }
  75
  76 // BlockProposeMsg block propose message transferred between nodes.
  77 type BlockProposeMsg struct {
  78         RawBlock []byte
  79 }
  80
  81 //NewBlockProposeMsg create new block propose msg.
  82 func NewBlockProposeMsg(block *types.Block) (ConsensusMessage, error) {
  83         rawBlock, err := block.MarshalText()
  84         if err != nil {
  85                 return nil, err
  86         }
  87         return &BlockProposeMsg{RawBlock: rawBlock}, nil
  88 }
  89
  90 //GetProposeBlock get propose block from msg.
  91 func (bp *BlockProposeMsg) GetProposeBlock() (*types.Block, error) {
  92         block := &types.Block{}
  93         if err := block.UnmarshalText(bp.RawBlock); err != nil {
  94                 return nil, err
  95         }
  96         return block, nil
  97 }
  98
  99 func (bp *BlockProposeMsg) String() string {
 100         block, err := bp.GetProposeBlock()
 101         if err != nil {
 102                 return "{err: wrong message}"
 103         }
 104         blockHash := block.Hash()
 105         return fmt.Sprintf("{block_height: %d, block_hash: %s}", block.Height, blockHash.String())
 106 }
 107
 108 // BroadcastMarkSendRecord mark send message record to prevent messages from being sent repeatedly.
 109 func (bp *BlockProposeMsg) BroadcastMarkSendRecord(ps *peers.PeerSet, peers []string) {
 110         block, err := bp.GetProposeBlock()
 111         if err != nil {
 112                 return
 113         }
 114
 115         hash := block.Hash()
 116         height := block.Height
 117         for _, peer := range peers {
 118                 ps.MarkBlock(peer, &hash)
 119                 ps.MarkStatus(peer, height)
 120         }
 121 }
 122
 123 // BroadcastFilterTargetPeers filter target peers to filter the nodes that need to send messages.
 124 func (bp *BlockProposeMsg) BroadcastFilterTargetPeers(ps *peers.PeerSet) []string {
 125         block, err := bp.GetProposeBlock()
 126         if err != nil {
 127                 return nil
 128         }
 129
 130         return ps.PeersWithoutBlock(block.Hash())
 131 }