tools/monitor_tx/main.go

   1 package main
   2
   3 import (
   4         "bytes"
   5         "encoding/json"
   6         "fmt"
   7         "log"
   8         "os"
   9         "sync"
  10         "time"
  11
  12         "github.com/spf13/cobra"
  13         bytomtypes "github.com/vapor/claim/bytom/protocolbc/types"
  14         "github.com/vapor/config"
  15         chainjson "github.com/vapor/encoding/json"
  16         "github.com/vapor/util"
  17 )
  18
  19 var lock sync.Mutex
  20 var gClaimTxMap map[uint64]claimTx
  21 var currentHeight uint64
  22
  23 type WSRequest struct {
  24         Topic string `json:"topic"`
  25 }
  26
  27 type WSResponse struct {
  28         NotificationType string      `json:"notification_type"`
  29         Data             interface{} `json:"data"`
  30         ErrorDetail      string      `json:"error_detail,omitempty"`
  31 }
  32
  33 type claimTx struct {
  34         Password     string                 `json:"password"`
  35         RawTx        bytomtypes.Tx          `json:"raw_transaction"`
  36         BlockHeader  bytomtypes.BlockHeader `json:"block_header"`
  37         TxHashes     []chainjson.HexBytes   `json:"tx_hashes"`
  38         StatusHashes []chainjson.HexBytes   `json:"status_hashes"`
  39         Flags        []uint32               `json:"flags"`
  40         MatchedTxIDs []chainjson.HexBytes   `json:"matched_tx_ids"`
  41         ClaimScript  chainjson.HexBytes     `json:"claim_script"`
  42 }
  43
  44 var startHeight uint64 = 0
  45
  46 var runCmd = &cobra.Command{
  47         Use:   "run",
  48         Short: "monitor claim tx",
  49         Run:   run,
  50 }
  51
  52 func init() {
  53         runCmd.PersistentFlags().Uint64Var(&startHeight, "start_height", 0, "Start monitoring block height for transactions")
  54 }
  55
  56 func run(cmd *cobra.Command, args []string) {
  57
  58 }
  59
  60 func main() {
  61
  62         if _, err := runCmd.ExecuteC(); err != nil {
  63                 os.Exit(util.ErrLocalExe)
  64         }
  65
  66         gClaimTxMap = make(map[uint64]claimTx)
  67         currentHeight = 0
  68         client := &WSClient{}
  69         if err := client.New("127.0.0.1:9888"); err != nil {
  70                 log.Println(err)
  71                 return
  72         }
  73         go sendClaimTx()
  74         go getRawTransactionWithHeight()
  75         req := WSRequest{
  76                 Topic: "notify_raw_blocks",
  77         }
  78         client.SendData(req)
  79
  80         for {
  81                 msg, err := client.RecvData()
  82                 if err != nil {
  83                         fmt.Println(err)
  84                         break
  85                 }
  86                 var rep WSResponse
  87                 err = json.Unmarshal(msg, &rep)
  88                 if err != nil {
  89                         log.Printf("Unmarshal error: %v", err)
  90                 }
  91
  92                 block := &bytomtypes.Block{}
  93                 switch rep.NotificationType {
  94                 case "raw_blocks_connected":
  95                         data := fmt.Sprint(rep.Data)
  96                         err = block.UnmarshalText([]byte(data))
  97                         if err != nil {
  98                                 block = nil
  99                         }
 100                 case "raw_blocks_disconnected":
 101                         data := fmt.Sprint(rep.Data)
 102                         err = block.UnmarshalText([]byte(data))
 103                         if err != nil {
 104                                 block = nil
 105                         }
 106                 case "request_status":
 107                         if rep.ErrorDetail != "" {
 108                                 log.Println(rep.ErrorDetail)
 109                         }
 110                         block = nil
 111                 default:
 112                         block = nil
 113                 }
 114                 if block != nil {
 115                         currentHeight = block.Height
 116                         err := getRawTransaction(block)
 117                         if err != nil {
 118                                 log.Fatal(err)
 119                         }
 120                 }
 121         }
 122 }
 123
 124 func sendClaimTx() {
 125         for {
 126                 // 存储tx到数据库、列表中
 127                 for k, v := range gClaimTxMap {
 128                         if k <= currentHeight {
 129                                 resp, exitCode := util.ClientCall("/claim-pegin-transaction", &v)
 130                                 if exitCode != util.Success {
 131                                         lock.Lock()
 132                                         delete(gClaimTxMap, k)
 133                                         lock.Unlock()
 134                                         continue
 135                                 }
 136                                 type txID struct {
 137                                         Txid string `json:"tx_id"`
 138                                 }
 139                                 var out txID
 140                                 restoreStruct(resp, &out)
 141                                 lock.Lock()
 142                                 delete(gClaimTxMap, k)
 143                                 lock.Unlock()
 144                                 fmt.Println(out.Txid)
 145                                 time.Sleep(3 * time.Second)
 146                         }
 147                 }
 148         }
 149
 150 }
 151
 152 func restoreStruct(data interface{}, out interface{}) {
 153         dataMap, ok := data.(map[string]interface{})
 154         if ok != true {
 155                 fmt.Println("invalid type assertion")
 156                 os.Exit(util.ErrLocalParse)
 157         }
 158
 159         rawData, err := json.MarshalIndent(dataMap, "", "  ")
 160         if err != nil {
 161                 fmt.Println(err)
 162                 os.Exit(util.ErrLocalParse)
 163         }
 164         json.Unmarshal(rawData, out)
 165 }
 166
 167 func getRawTransaction(block *bytomtypes.Block) error {
 168         peginInfo, err := getPeginInfo()
 169         if err != nil {
 170                 return err
 171         }
 172         for _, tx := range block.Transactions {
 173                 for _, output := range tx.Outputs {
 174                         for k, v := range peginInfo {
 175                                 var claimScript chainjson.HexBytes
 176                                 if err := claimScript.UnmarshalText([]byte(k)); err != nil {
 177                                         return err
 178                                 }
 179                                 var controlProgram chainjson.HexBytes
 180                                 if err := controlProgram.UnmarshalText([]byte(v)); err != nil {
 181                                         return err
 182                                 }
 183
 184                                 if !bytes.Equal(output.ControlProgram, controlProgram) {
 185                                         continue
 186                                 }
 187                                 blockHash := block.Hash()
 188                                 merkleProof, err := getMerkleProof(blockHash.String(), tx.ID.String())
 189                                 if err != nil {
 190                                         return err
 191                                 }
 192                                 tmp := claimTx{
 193                                         Password:     "123456",
 194                                         RawTx:        *tx,
 195                                         BlockHeader:  merkleProof.BlockHeader,
 196                                         TxHashes:     merkleProof.TxHashes,
 197                                         StatusHashes: merkleProof.StatusHashes,
 198                                         Flags:        merkleProof.Flags,
 199                                         MatchedTxIDs: merkleProof.MatchedTxIDs,
 200                                         ClaimScript:  claimScript,
 201                                 }
 202                                 // 存储tx到数据库、列表中
 203                                 height := block.Height + 6
 204                                 lock.Lock()
 205                                 gClaimTxMap[height] = tmp
 206                                 lock.Unlock()
 207                         }
 208                 }
 209         }
 210
 211         return nil
 212 }
 213
 214 func getRawTransactionWithHeight() {
 215         for {
 216                 if currentHeight > 0 {
 217                         num := currentHeight - startHeight
 218                         for i := uint64(0); i < num; i++ {
 219                                 block, err := getBlockWithHeight(startHeight)
 220                                 if err != nil {
 221                                         log.Fatal(err)
 222                                 }
 223                                 err = getRawTransaction(&block)
 224                                 if err != nil {
 225                                         log.Fatal(err)
 226                                 }
 227                                 startHeight += 1
 228                         }
 229                         if startHeight >= currentHeight {
 230                                 break
 231                         }
 232                 }
 233                 time.Sleep(1 * time.Second)
 234         }
 235 }
 236
 237 type MerkleBlockResp struct {
 238         BlockHeader  bytomtypes.BlockHeader `json:"block_header"`
 239         TxHashes     []chainjson.HexBytes   `json:"tx_hashes"`
 240         StatusHashes []chainjson.HexBytes   `json:"status_hashes"`
 241         Flags        []uint32               `json:"flags"`
 242         MatchedTxIDs []chainjson.HexBytes   `json:"matched_tx_ids"`
 243 }
 244
 245 func getMerkleProof(blockHash string, txId string) (MerkleBlockResp, error) {
 246         //body_json = {"tx_id": tx_id,"block_hash": block_hash}
 247         type Req struct {
 248                 TxID      string `json:"tx_id"`
 249                 BlockHash string `json:"block_hash"`
 250         }
 251         util.MainchainConfig = &config.MainChainRpcConfig{
 252                 MainchainRpcHost: "127.0.0.1",
 253                 MainchainRpcPort: "9888",
 254         }
 255         var blockHeader MerkleBlockResp
 256         resp, err := util.CallRPC("/get-merkle-proof", &Req{TxID: txId, BlockHash: blockHash})
 257         if err != nil {
 258                 return blockHeader, err
 259         }
 260         tmp, _ := json.Marshal(resp)
 261
 262         json.Unmarshal(tmp, &blockHeader)
 263
 264         return blockHeader, nil
 265 }
 266
 267 func getBlockWithHeight(blockHeight uint64) (bytomtypes.Block, error) {
 268         type Req struct {
 269                 BlockHeight uint64 `json:"block_height"`
 270         }
 271         util.MainchainConfig = &config.MainChainRpcConfig{
 272                 MainchainRpcHost: "127.0.0.1",
 273                 MainchainRpcPort: "9888",
 274         }
 275         type RawBlockResp struct {
 276                 RawBlock *bytomtypes.Block `json:"raw_block"`
 277         }
 278         var block RawBlockResp
 279         resp, err := util.CallRPC("/get-raw-block", &Req{BlockHeight: blockHeight})
 280         if err != nil {
 281                 return *block.RawBlock, err
 282         }
 283         tmp, _ := json.Marshal(resp)
 284
 285         json.Unmarshal(tmp, &block)
 286         return *block.RawBlock, nil
 287 }