cic-chain-events/internal/syncer/head.go

75 lines
1.7 KiB
Go
Raw Normal View History

2023-01-05 12:45:09 +01:00
package syncer
import (
"context"
"github.com/alitto/pond"
"github.com/celo-org/celo-blockchain/core/types"
"github.com/celo-org/celo-blockchain/ethclient"
"github.com/grassrootseconomics/cic-chain-events/internal/pipeline"
"github.com/zerodha/logf"
)
type HeadSyncerOpts struct {
Stats *Stats
Pipeline *pipeline.Pipeline
Logg logf.Logger
Pool *pond.WorkerPool
WsEndpoint string
}
type HeadSyncer struct {
stats *Stats
pipeline *pipeline.Pipeline
logg logf.Logger
ethClient *ethclient.Client
pool *pond.WorkerPool
}
func NewHeadSyncer(o HeadSyncerOpts) (*HeadSyncer, error) {
ethClient, err := ethclient.Dial(o.WsEndpoint)
if err != nil {
return nil, err
}
return &HeadSyncer{
stats: o.Stats,
pipeline: o.Pipeline,
logg: o.Logg,
ethClient: ethClient,
pool: o.Pool,
}, nil
}
// Start creates a websocket subscription and actively receives new blocks untill stopped
// or a critical error occurs.
2023-01-05 12:45:09 +01:00
func (hs *HeadSyncer) Start(ctx context.Context) error {
headerReceiver := make(chan *types.Header, 1)
2023-01-05 12:45:09 +01:00
sub, err := hs.ethClient.SubscribeNewHead(ctx, headerReceiver)
if err != nil {
return err
}
defer sub.Unsubscribe()
2023-01-05 12:45:09 +01:00
for {
select {
case <-ctx.Done():
hs.logg.Info("head syncer: shutdown signal received")
return nil
case err := <-sub.Err():
return err
2023-01-05 12:45:09 +01:00
case header := <-headerReceiver:
blockNumber := header.Number.Uint64()
hs.logg.Debug("head syncer: received new block", "block", blockNumber)
2023-01-05 12:45:09 +01:00
hs.stats.UpdateHeadCursor(blockNumber)
2023-01-05 12:45:09 +01:00
hs.pool.Submit(func() {
if err := hs.pipeline.Run(context.Background(), blockNumber); err != nil {
hs.logg.Error("head syncer: piepline run error", "error", err)
2023-01-05 12:45:09 +01:00
}
})
}
}
}