federation/synchron/mainchain_keeper.go

   1 package synchron
   2
   3 import (
   4         "time"
   5
   6         btmTypes "github.com/bytom/protocol/bc/types"
   7         "github.com/jinzhu/gorm"
   8         log "github.com/sirupsen/logrus"
   9
  10         "github.com/vapor/errors"
  11         "github.com/vapor/federation/config"
  12         "github.com/vapor/federation/database/orm"
  13         "github.com/vapor/federation/service"
  14         "github.com/vapor/protocol/bc"
  15 )
  16
  17 type mainchainKeeper struct {
  18         cfg       *config.Chain
  19         db        *gorm.DB
  20         node      *service.Node
  21         chainName string
  22 }
  23
  24 func NewMainchainKeeper(db *gorm.DB, chainCfg *config.Chain) *mainchainKeeper {
  25         return &mainchainKeeper{
  26                 cfg:       chainCfg,
  27                 db:        db,
  28                 node:      service.NewNode(chainCfg.Upstream),
  29                 chainName: chainCfg.Name,
  30         }
  31 }
  32
  33 func (m *mainchainKeeper) Run() {
  34         ticker := time.NewTicker(time.Duration(m.cfg.SyncSeconds) * time.Second)
  35         for ; true; <-ticker.C {
  36                 for {
  37                         isUpdate, err := m.syncBlock()
  38                         if err != nil {
  39                                 log.WithField("error", err).Errorln("blockKeeper fail on process block")
  40                                 break
  41                         }
  42
  43                         if !isUpdate {
  44                                 break
  45                         }
  46                 }
  47         }
  48 }
  49
  50 func (m *mainchainKeeper) syncBlock() (bool, error) {
  51         chain := &orm.Chain{Name: m.chainName}
  52         if err := m.db.Where(chain).First(chain).Error; err != nil {
  53                 return false, errors.Wrap(err, "query chain")
  54         }
  55
  56         height, err := m.node.GetBlockCount()
  57         if err != nil {
  58                 return false, err
  59         }
  60
  61         if height <= chain.BlockHeight+m.cfg.Confirmations {
  62                 return false, nil
  63         }
  64
  65         nextBlockStr, txStatus, err := m.node.GetBlockByHeight(chain.BlockHeight + 1)
  66         if err != nil {
  67                 return false, err
  68         }
  69
  70         nextBlock := &btmTypes.Block{}
  71         nextBlock.UnmarshalText([]byte(nextBlockStr))
  72         if nextBlock.PreviousBlockHash.String() == chain.BlockHash {
  73                 return true, m.attachBlock(chain, nextBlock, txStatus)
  74         } else {
  75                 log.WithFields(log.Fields{
  76                         "remote PreviousBlockHash": nextBlock.PreviousBlockHash.String(),
  77                         "db block_hash":            chain.BlockHash,
  78                 }).Fatalf("PreviousBlockHash mismatch")
  79                 return false, nil
  80         }
  81 }
  82
  83 func (m *mainchainKeeper) attachBlock(chain *orm.Chain, block *btmTypes.Block, txStatus *bc.TransactionStatus) error {
  84         blockHash := block.Hash()
  85         log.WithFields(log.Fields{"block_height": block.Height, "block_hash": blockHash.String()}).Info("start to attachBlock")
  86         m.db.Begin()
  87         if err := m.processBlock(block); err != nil {
  88                 m.db.Rollback()
  89                 return err
  90         }
  91
  92         return m.db.Commit().Error
  93 }
  94
  95 func (m *mainchainKeeper) processBlock(block *btmTypes.Block) error {
  96         if err := m.processIssuing(block.Transactions); err != nil {
  97                 return err
  98         }
  99
 100         return nil
 101 }
 102
 103 func (m *mainchainKeeper) processIssuing(txs []*btmTypes.Tx) error {
 104         return nil
 105         // if err := m.processIssuing(block.Transactions); err != nil {
 106         //      return err
 107         // }
 108 }