(ctx context.Context)
| 105 | } |
| 106 | |
| 107 | func (bs *BusService) subscribeBlock(ctx context.Context) { |
| 108 | defer bs.wg.Done() |
| 109 | |
| 110 | log.Info("start to subscribe blocks") |
| 111 | for { |
| 112 | select { |
| 113 | case <-ctx.Done(): |
| 114 | log.Info("exit subscription service") |
| 115 | return |
| 116 | case <-time.After(bs.checkInterval): |
| 117 | // fetch block from remote block producer |
| 118 | c := atomic.LoadUint32(&bs.blockCount) |
| 119 | log.Debugf("fetch block in count: %d", c) |
| 120 | b, profiles, newCount := bs.requestLastBlock() |
| 121 | if b == nil { |
| 122 | continue |
| 123 | } |
| 124 | if newCount <= c { |
| 125 | continue |
| 126 | } |
| 127 | |
| 128 | log.WithFields(log.Fields{ |
| 129 | "last_count": c, |
| 130 | "new_count": newCount, |
| 131 | "block_hash": b.BlockHash().Short(4), |
| 132 | "tx_num": len(b.Transactions), |
| 133 | }).Debug("success fetch block") |
| 134 | |
| 135 | // Write sqlchain profile state first (bound to the last irreversible block) |
| 136 | bs.updateState(newCount, profiles) |
| 137 | |
| 138 | // Fetch any intermediate irreversible blocks and extract txs |
| 139 | for i := c + 1; i < newCount; i++ { |
| 140 | var ( |
| 141 | block *types.BPBlock |
| 142 | err error |
| 143 | ) |
| 144 | if block, err = bs.fetchBlockByCount(i); err != nil { |
| 145 | log.WithError(err).WithFields(log.Fields{ |
| 146 | "count": i, |
| 147 | }).Warn("failed to fetch block") |
| 148 | continue |
| 149 | } |
| 150 | bs.extractTxs(block, i) |
| 151 | } |
| 152 | |
| 153 | // Extract txs in last irreversible block |
| 154 | bs.extractTxs(b, c) |
| 155 | } |
| 156 | } |
| 157 | } |
| 158 | |
| 159 | func (bs *BusService) fetchBlockByCount(count uint32) (block *types.BPBlock, err error) { |
| 160 | var ( |
no test coverage detected