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

87 lines
2.1 KiB
Go
Raw Normal View History

2023-01-05 12:45:09 +01:00
package syncer
import (
"context"
"time"
2023-01-05 12:45:09 +01:00
"github.com/alitto/pond"
"github.com/celo-org/celo-blockchain/core/types"
"github.com/celo-org/celo-blockchain/ethclient"
"github.com/celo-org/celo-blockchain/event"
2023-01-05 12:45:09 +01:00
"github.com/grassrootseconomics/cic-chain-events/internal/pipeline"
"github.com/zerodha/logf"
)
const (
jobTimeout = 5 * time.Second
resubscribeBackoff = 2 * time.Second
)
2023-01-05 12:45:09 +01:00
type (
HeadSyncerOpts struct {
Logg logf.Logger
Pipeline *pipeline.Pipeline
Pool *pond.WorkerPool
Stats *Stats
WsEndpoint string
}
HeadSyncer struct {
ethClient *ethclient.Client
logg logf.Logger
pipeline *pipeline.Pipeline
pool *pond.WorkerPool
stats *Stats
}
)
2023-01-05 12:45:09 +01:00
func NewHeadSyncer(o HeadSyncerOpts) (*HeadSyncer, error) {
ethClient, err := ethclient.Dial(o.WsEndpoint)
if err != nil {
return nil, err
}
return &HeadSyncer{
ethClient: ethClient,
logg: o.Logg,
pipeline: o.Pipeline,
2023-01-05 12:45:09 +01:00
pool: o.Pool,
stats: o.Stats,
2023-01-05 12:45:09 +01:00
}, nil
}
// Start creates a websocket subscription and actively receives new blocks until 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 := event.ResubscribeErr(resubscribeBackoff, func(ctx context.Context, err error) (event.Subscription, error) {
if err != nil {
hs.logg.Error("head syncer: resubscribe error", "error", err)
}
return hs.ethClient.SubscribeNewHead(ctx, headerReceiver)
})
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
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)
hs.stats.UpdateHeadCursor(blockNumber)
2023-01-05 12:45:09 +01:00
hs.pool.Submit(func() {
ctx, cancel := context.WithTimeout(context.Background(), jobTimeout)
defer cancel()
if err := hs.pipeline.Run(ctx, blockNumber); err != nil {
hs.logg.Error("head syncer: pipeline run error", "error", err)
2023-01-05 12:45:09 +01:00
}
})
}
}
}