netsync/message.go

   1 package netsync
   2
   3 import (
   4         "bytes"
   5         "encoding/hex"
   6         "encoding/json"
   7         "errors"
   8         "fmt"
   9
  10         "github.com/tendermint/go-wire"
  11
  12         "github.com/vapor/protocol/bc"
  13         "github.com/vapor/protocol/bc/types"
  14 )
  15
  16 //protocol msg byte
  17 const (
  18         BlockchainChannel = byte(0x40)
  19
  20         BlockRequestByte    = byte(0x10)
  21         BlockResponseByte   = byte(0x11)
  22         HeadersRequestByte  = byte(0x12)
  23         HeadersResponseByte = byte(0x13)
  24         BlocksRequestByte   = byte(0x14)
  25         BlocksResponseByte  = byte(0x15)
  26         StatusRequestByte   = byte(0x20)
  27         StatusResponseByte  = byte(0x21)
  28         NewTransactionByte  = byte(0x30)
  29         NewMineBlockByte    = byte(0x40)
  30         FilterLoadByte      = byte(0x50)
  31         FilterAddByte       = byte(0x51)
  32         FilterClearByte     = byte(0x52)
  33         MerkleRequestByte   = byte(0x60)
  34         MerkleResponseByte  = byte(0x61)
  35
  36         maxBlockchainResponseSize = 22020096 + 2
  37 )
  38
  39 //BlockchainMessage is a generic message for this reactor.
  40 type BlockchainMessage interface {
  41         String() string
  42 }
  43
  44 var _ = wire.RegisterInterface(
  45         struct{ BlockchainMessage }{},
  46         wire.ConcreteType{&GetBlockMessage{}, BlockRequestByte},
  47         wire.ConcreteType{&BlockMessage{}, BlockResponseByte},
  48         wire.ConcreteType{&GetHeadersMessage{}, HeadersRequestByte},
  49         wire.ConcreteType{&HeadersMessage{}, HeadersResponseByte},
  50         wire.ConcreteType{&GetBlocksMessage{}, BlocksRequestByte},
  51         wire.ConcreteType{&BlocksMessage{}, BlocksResponseByte},
  52         wire.ConcreteType{&StatusRequestMessage{}, StatusRequestByte},
  53         wire.ConcreteType{&StatusResponseMessage{}, StatusResponseByte},
  54         wire.ConcreteType{&TransactionMessage{}, NewTransactionByte},
  55         wire.ConcreteType{&MineBlockMessage{}, NewMineBlockByte},
  56         wire.ConcreteType{&FilterLoadMessage{}, FilterLoadByte},
  57         wire.ConcreteType{&FilterAddMessage{}, FilterAddByte},
  58         wire.ConcreteType{&FilterClearMessage{}, FilterClearByte},
  59         wire.ConcreteType{&GetMerkleBlockMessage{}, MerkleRequestByte},
  60         wire.ConcreteType{&MerkleBlockMessage{}, MerkleResponseByte},
  61 )
  62
  63 //DecodeMessage decode msg
  64 func DecodeMessage(bz []byte) (msgType byte, msg BlockchainMessage, err error) {
  65         msgType = bz[0]
  66         n := int(0)
  67         r := bytes.NewReader(bz)
  68         msg = wire.ReadBinary(struct{ BlockchainMessage }{}, r, maxBlockchainResponseSize, &n, &err).(struct{ BlockchainMessage }).BlockchainMessage
  69         if err != nil && n != len(bz) {
  70                 err = errors.New("DecodeMessage() had bytes left over")
  71         }
  72         return
  73 }
  74
  75 //GetBlockMessage request blocks from remote peers by height/hash
  76 type GetBlockMessage struct {
  77         Height  uint64
  78         RawHash [32]byte
  79 }
  80
  81 //GetHash reutrn the hash of the request
  82 func (m *GetBlockMessage) GetHash() *bc.Hash {
  83         hash := bc.NewHash(m.RawHash)
  84         return &hash
  85 }
  86
  87 func (m *GetBlockMessage) String() string {
  88         if m.Height > 0 {
  89                 return fmt.Sprintf("{height: %d}", m.Height)
  90         }
  91         return fmt.Sprintf("{hash: %s}", hex.EncodeToString(m.RawHash[:]))
  92 }
  93
  94 //BlockMessage response get block msg
  95 type BlockMessage struct {
  96         RawBlock []byte
  97 }
  98
  99 //NewBlockMessage construct bock response msg
 100 func NewBlockMessage(block *types.Block) (*BlockMessage, error) {
 101         rawBlock, err := block.MarshalText()
 102         if err != nil {
 103                 return nil, err
 104         }
 105         return &BlockMessage{RawBlock: rawBlock}, nil
 106 }
 107
 108 //GetBlock get block from msg
 109 func (m *BlockMessage) GetBlock() (*types.Block, error) {
 110         block := &types.Block{
 111                 BlockHeader:  types.BlockHeader{},
 112                 Transactions: []*types.Tx{},
 113         }
 114         if err := block.UnmarshalText(m.RawBlock); err != nil {
 115                 return nil, err
 116         }
 117         return block, nil
 118 }
 119
 120 func (m *BlockMessage) String() string {
 121         block, err := m.GetBlock()
 122         if err != nil {
 123                 return "{err: wrong message}"
 124         }
 125         blockHash := block.Hash()
 126         return fmt.Sprintf("{block_height: %d, block_hash: %s}", block.Height, blockHash.String())
 127 }
 128
 129 //GetHeadersMessage is one of the bytom msg type
 130 type GetHeadersMessage struct {
 131         RawBlockLocator [][32]byte
 132         RawStopHash     [32]byte
 133 }
 134
 135 //NewGetHeadersMessage return a new GetHeadersMessage
 136 func NewGetHeadersMessage(blockLocator []*bc.Hash, stopHash *bc.Hash) *GetHeadersMessage {
 137         msg := &GetHeadersMessage{
 138                 RawStopHash: stopHash.Byte32(),
 139         }
 140         for _, hash := range blockLocator {
 141                 msg.RawBlockLocator = append(msg.RawBlockLocator, hash.Byte32())
 142         }
 143         return msg
 144 }
 145
 146 //GetBlockLocator return the locator of the msg
 147 func (m *GetHeadersMessage) GetBlockLocator() []*bc.Hash {
 148         blockLocator := []*bc.Hash{}
 149         for _, rawHash := range m.RawBlockLocator {
 150                 hash := bc.NewHash(rawHash)
 151                 blockLocator = append(blockLocator, &hash)
 152         }
 153         return blockLocator
 154 }
 155
 156 func (m *GetHeadersMessage) String() string {
 157         return fmt.Sprintf("{stop_hash: %s}", hex.EncodeToString(m.RawStopHash[:]))
 158 }
 159
 160 //GetStopHash return the stop hash of the msg
 161 func (m *GetHeadersMessage) GetStopHash() *bc.Hash {
 162         hash := bc.NewHash(m.RawStopHash)
 163         return &hash
 164 }
 165
 166 //HeadersMessage is one of the bytom msg type
 167 type HeadersMessage struct {
 168         RawHeaders [][]byte
 169 }
 170
 171 //NewHeadersMessage create a new HeadersMessage
 172 func NewHeadersMessage(headers []*types.BlockHeader) (*HeadersMessage, error) {
 173         RawHeaders := [][]byte{}
 174         for _, header := range headers {
 175                 data, err := json.Marshal(header)
 176                 if err != nil {
 177                         return nil, err
 178                 }
 179
 180                 RawHeaders = append(RawHeaders, data)
 181         }
 182         return &HeadersMessage{RawHeaders: RawHeaders}, nil
 183 }
 184
 185 //GetHeaders return the headers in the msg
 186 func (m *HeadersMessage) GetHeaders() ([]*types.BlockHeader, error) {
 187         headers := []*types.BlockHeader{}
 188         for _, data := range m.RawHeaders {
 189                 header := &types.BlockHeader{}
 190                 if err := json.Unmarshal(data, header); err != nil {
 191                         return nil, err
 192                 }
 193
 194                 headers = append(headers, header)
 195         }
 196         return headers, nil
 197 }
 198
 199 func (m *HeadersMessage) String() string {
 200         return fmt.Sprintf("{header_length: %d}", len(m.RawHeaders))
 201 }
 202
 203 //GetBlocksMessage is one of the bytom msg type
 204 type GetBlocksMessage struct {
 205         RawBlockLocator [][32]byte
 206         RawStopHash     [32]byte
 207 }
 208
 209 //NewGetBlocksMessage create a new GetBlocksMessage
 210 func NewGetBlocksMessage(blockLocator []*bc.Hash, stopHash *bc.Hash) *GetBlocksMessage {
 211         msg := &GetBlocksMessage{
 212                 RawStopHash: stopHash.Byte32(),
 213         }
 214         for _, hash := range blockLocator {
 215                 msg.RawBlockLocator = append(msg.RawBlockLocator, hash.Byte32())
 216         }
 217         return msg
 218 }
 219
 220 //GetBlockLocator return the locator of the msg
 221 func (m *GetBlocksMessage) GetBlockLocator() []*bc.Hash {
 222         blockLocator := []*bc.Hash{}
 223         for _, rawHash := range m.RawBlockLocator {
 224                 hash := bc.NewHash(rawHash)
 225                 blockLocator = append(blockLocator, &hash)
 226         }
 227         return blockLocator
 228 }
 229
 230 //GetStopHash return the stop hash of the msg
 231 func (m *GetBlocksMessage) GetStopHash() *bc.Hash {
 232         hash := bc.NewHash(m.RawStopHash)
 233         return &hash
 234 }
 235
 236 func (m *GetBlocksMessage) String() string {
 237         return fmt.Sprintf("{stop_hash: %s}", hex.EncodeToString(m.RawStopHash[:]))
 238 }
 239
 240 //BlocksMessage is one of the bytom msg type
 241 type BlocksMessage struct {
 242         RawBlocks [][]byte
 243 }
 244
 245 //NewBlocksMessage create a new BlocksMessage
 246 func NewBlocksMessage(blocks []*types.Block) (*BlocksMessage, error) {
 247         rawBlocks := [][]byte{}
 248         for _, block := range blocks {
 249                 data, err := json.Marshal(block)
 250                 if err != nil {
 251                         return nil, err
 252                 }
 253
 254                 rawBlocks = append(rawBlocks, data)
 255         }
 256         return &BlocksMessage{RawBlocks: rawBlocks}, nil
 257 }
 258
 259 //GetBlocks returns the blocks in the msg
 260 func (m *BlocksMessage) GetBlocks() ([]*types.Block, error) {
 261         blocks := []*types.Block{}
 262         for _, data := range m.RawBlocks {
 263                 block := &types.Block{}
 264                 if err := json.Unmarshal(data, block); err != nil {
 265                         return nil, err
 266                 }
 267
 268                 blocks = append(blocks, block)
 269         }
 270         return blocks, nil
 271 }
 272
 273 func (m *BlocksMessage) String() string {
 274         return fmt.Sprintf("{blocks_length: %d}", len(m.RawBlocks))
 275 }
 276
 277 //StatusRequestMessage status request msg
 278 type StatusRequestMessage struct{}
 279
 280 func (m *StatusRequestMessage) String() string {
 281         return "{}"
 282 }
 283
 284 //StatusResponseMessage get status response msg
 285 type StatusResponseMessage struct {
 286         Height      uint64
 287         RawHash     [32]byte
 288         GenesisHash [32]byte
 289 }
 290
 291 //NewStatusResponseMessage construct get status response msg
 292 func NewStatusResponseMessage(blockHeader *types.BlockHeader, hash *bc.Hash) *StatusResponseMessage {
 293         return &StatusResponseMessage{
 294                 Height:      blockHeader.Height,
 295                 RawHash:     blockHeader.Hash().Byte32(),
 296                 GenesisHash: hash.Byte32(),
 297         }
 298 }
 299
 300 //GetHash get hash from msg
 301 func (m *StatusResponseMessage) GetHash() *bc.Hash {
 302         hash := bc.NewHash(m.RawHash)
 303         return &hash
 304 }
 305
 306 //GetGenesisHash get hash from msg
 307 func (m *StatusResponseMessage) GetGenesisHash() *bc.Hash {
 308         hash := bc.NewHash(m.GenesisHash)
 309         return &hash
 310 }
 311
 312 func (m *StatusResponseMessage) String() string {
 313         return fmt.Sprintf("{height: %d, hash: %s}", m.Height, hex.EncodeToString(m.RawHash[:]))
 314 }
 315
 316 //TransactionMessage notify new tx msg
 317 type TransactionMessage struct {
 318         RawTx []byte
 319 }
 320
 321 //NewTransactionMessage construct notify new tx msg
 322 func NewTransactionMessage(tx *types.Tx) (*TransactionMessage, error) {
 323         rawTx, err := tx.TxData.MarshalText()
 324         if err != nil {
 325                 return nil, err
 326         }
 327         return &TransactionMessage{RawTx: rawTx}, nil
 328 }
 329
 330 //GetTransaction get tx from msg
 331 func (m *TransactionMessage) GetTransaction() (*types.Tx, error) {
 332         tx := &types.Tx{}
 333         if err := tx.UnmarshalText(m.RawTx); err != nil {
 334                 return nil, err
 335         }
 336         return tx, nil
 337 }
 338
 339 func (m *TransactionMessage) String() string {
 340         tx, err := m.GetTransaction()
 341         if err != nil {
 342                 return "{err: wrong message}"
 343         }
 344         return fmt.Sprintf("{tx_size: %d, tx_hash: %s}", len(m.RawTx), tx.ID.String())
 345 }
 346
 347 //MineBlockMessage new mined block msg
 348 type MineBlockMessage struct {
 349         RawBlock []byte
 350 }
 351
 352 //NewMinedBlockMessage construct new mined block msg
 353 func NewMinedBlockMessage(block *types.Block) (*MineBlockMessage, error) {
 354         rawBlock, err := block.MarshalText()
 355         if err != nil {
 356                 return nil, err
 357         }
 358         return &MineBlockMessage{RawBlock: rawBlock}, nil
 359 }
 360
 361 //GetMineBlock get mine block from msg
 362 func (m *MineBlockMessage) GetMineBlock() (*types.Block, error) {
 363         block := &types.Block{}
 364         if err := block.UnmarshalText(m.RawBlock); err != nil {
 365                 return nil, err
 366         }
 367         return block, nil
 368 }
 369
 370 func (m *MineBlockMessage) String() string {
 371         block, err := m.GetMineBlock()
 372         if err != nil {
 373                 return "{err: wrong message}"
 374         }
 375         blockHash := block.Hash()
 376         return fmt.Sprintf("{block_height: %d, block_hash: %s}", block.Height, blockHash.String())
 377 }
 378
 379 //FilterLoadMessage tells the receiving peer to filter the transactions according to address.
 380 type FilterLoadMessage struct {
 381         Addresses [][]byte
 382 }
 383
 384 func (m *FilterLoadMessage) String() string {
 385         return fmt.Sprintf("{addresses_length: %d}", len(m.Addresses))
 386 }
 387
 388 // FilterAddMessage tells the receiving peer to add address to the filter.
 389 type FilterAddMessage struct {
 390         Address []byte
 391 }
 392
 393 func (m *FilterAddMessage) String() string {
 394         return fmt.Sprintf("{address: %s}", hex.EncodeToString(m.Address))
 395 }
 396
 397 //FilterClearMessage tells the receiving peer to remove a previously-set filter.
 398 type FilterClearMessage struct{}
 399
 400 func (m *FilterClearMessage) String() string {
 401         return "{}"
 402 }
 403
 404 //GetMerkleBlockMessage request merkle blocks from remote peers by height/hash
 405 type GetMerkleBlockMessage struct {
 406         Height  uint64
 407         RawHash [32]byte
 408 }
 409
 410 //GetHash reutrn the hash of the request
 411 func (m *GetMerkleBlockMessage) GetHash() *bc.Hash {
 412         hash := bc.NewHash(m.RawHash)
 413         return &hash
 414 }
 415
 416 func (m *GetMerkleBlockMessage) String() string {
 417         if m.Height > 0 {
 418                 return fmt.Sprintf("{height: %d}", m.Height)
 419         }
 420         return fmt.Sprintf("{hash: %s}", hex.EncodeToString(m.RawHash[:]))
 421 }
 422
 423 //MerkleBlockMessage return the merkle block to client
 424 type MerkleBlockMessage struct {
 425         RawBlockHeader []byte
 426         TxHashes       [][32]byte
 427         RawTxDatas     [][]byte
 428         StatusHashes   [][32]byte
 429         RawTxStatuses  [][]byte
 430         Flags          []byte
 431 }
 432
 433 func (m *MerkleBlockMessage) setRawBlockHeader(bh types.BlockHeader) error {
 434         rawHeader, err := bh.MarshalText()
 435         if err != nil {
 436                 return err
 437         }
 438
 439         m.RawBlockHeader = rawHeader
 440         return nil
 441 }
 442
 443 func (m *MerkleBlockMessage) setTxInfo(txHashes []*bc.Hash, txFlags []uint8, relatedTxs []*types.Tx) error {
 444         for _, txHash := range txHashes {
 445                 m.TxHashes = append(m.TxHashes, txHash.Byte32())
 446         }
 447         for _, tx := range relatedTxs {
 448                 rawTxData, err := tx.MarshalText()
 449                 if err != nil {
 450                         return err
 451                 }
 452
 453                 m.RawTxDatas = append(m.RawTxDatas, rawTxData)
 454         }
 455         m.Flags = txFlags
 456         return nil
 457 }
 458
 459 func (m *MerkleBlockMessage) setStatusInfo(statusHashes []*bc.Hash, relatedStatuses []*bc.TxVerifyResult) error {
 460         for _, statusHash := range statusHashes {
 461                 m.StatusHashes = append(m.StatusHashes, statusHash.Byte32())
 462         }
 463
 464         for _, status := range relatedStatuses {
 465                 rawStatusData, err := json.Marshal(status)
 466                 if err != nil {
 467                         return err
 468                 }
 469
 470                 m.RawTxStatuses = append(m.RawTxStatuses, rawStatusData)
 471         }
 472         return nil
 473 }
 474
 475 func (m *MerkleBlockMessage) String() string {
 476         return "{}"
 477 }
 478
 479 //NewMerkleBlockMessage construct merkle block message
 480 func NewMerkleBlockMessage() *MerkleBlockMessage {
 481         return &MerkleBlockMessage{}
 482 }