clean up code and fix bug

blockpoller/fetcher.go

@@ -0,0 +1,13 @@
+package blockpoller
+
+import (
+	"context"
+	"time"
+
+	pbbstream "github.com/streamingfast/bstream/pb/sf/bstream/v1"
+)
+
+type BlockFetcher interface {
+	PollingInterval() time.Duration
+	Fetch(ctx context.Context, blkNum uint64) (*pbbstream.Block, error)
+}

blockpoller/types.go → blockpoller/handler.go

@@ -1,52 +1,39 @@
 package blockpoller
 
 import (
-	"context"
 	"encoding/base64"
 	"fmt"
-	"time"
+	"sync"
 
 	pbbstream "github.com/streamingfast/bstream/pb/sf/bstream/v1"
-	"google.golang.org/protobuf/types/known/anypb"
 )
 
-type BlockFireFunc func(b *block) error
-
-type BlockFetcher interface {
-	PollingInterval() time.Duration
-	Fetch(ctx context.Context, blkNum uint64) (*pbbstream.Block, error)
-}
-
-type BlockFinalizer interface {
+type BlockHandler interface {
 	Init()
 	Fire(blk *pbbstream.Block) error
 }
 
-var _ BlockFinalizer = (*FireBlockFinalizer)(nil)
+var _ BlockHandler = (*FireBlockHandler)(nil)
 
-type FireBlockFinalizer struct {
+type FireBlockHandler struct {
 	blockTypeURL string
+	init         sync.Once
 }
 
-func NewFireBlockFinalizer(blockTypeURL string) *FireBlockFinalizer {
-	return &FireBlockFinalizer{
+func NewFireBlockHandler(blockTypeURL string) *FireBlockHandler {
+	return &FireBlockHandler{
 		blockTypeURL: blockTypeURL,
 	}
 }
 
-func (f *FireBlockFinalizer) Init() {
+func (f *FireBlockHandler) Init() {
 	fmt.Println("FIRE INIT 1.0 ", f.blockTypeURL)
 }
 
-func (f *FireBlockFinalizer) Fire(b *pbbstream.Block) error {
+func (f *FireBlockHandler) Fire(b *pbbstream.Block) error {
 	//blockLine := "FIRE BLOCK 18571000 d2836a703a02f3ca2a13f05efe26fc48c6fa0db0d754a49e56b066d3b7d54659 18570999 55de88c909fa368ae1e93b6b8ffb3fbb12e64aefec1d4a1fcc27ae7633de2f81 18570800 1699992393935935000 Ci10eXBlLmdvb2dsZWFwaXMuY29tL3NmLmV0aGVyZXVtLnR5cGUudjIuQmxvY2sSJxIg0oNqcDoC88oqE/Be/ib8SMb6DbDXVKSeVrBm07fVRlkY+L3tCA=="
-	anyBlock, err := anypb.New(b)
-	if err != nil {
-		return fmt.Errorf("converting block to anypb: %w", err)
-	}
-
-	if anyBlock.TypeUrl != f.blockTypeURL {
-		return fmt.Errorf("block type url %q does not match expected type %q", anyBlock.TypeUrl, f.blockTypeURL)
+	if b.Payload.TypeUrl != f.blockTypeURL {
+		return fmt.Errorf("block type url %q does not match expected type %q", b.Payload.TypeUrl, f.blockTypeURL)
 	}
 
 	blockLine := fmt.Sprintf(
@@ -57,7 +44,7 @@ func (f *FireBlockFinalizer) Fire(b *pbbstream.Block) error {
 		b.ParentId,
 		b.LibNum,
 		b.Timestamp.AsTime().UnixNano(),
-		base64.StdEncoding.EncodeToString(anyBlock.Value),
+		base64.StdEncoding.EncodeToString(b.Payload.Value),
 	)
 
 	fmt.Println(blockLine)

blockpoller/init_test.go

@@ -70,7 +70,7 @@ func (b *TestBlockFetcher) check(t *testing.T) {
 	require.Equal(b.t, uint64(len(b.blocks)), b.idx, "we should have fetched all %d blocks, only fired %d blocks", len(b.blocks), b.idx)
 }
 
-var _ BlockFinalizer = &TestBlockFinalizer{}
+var _ BlockHandler = &TestBlockFinalizer{}
 
 type TestBlockFinalizer struct {
 	t          *testing.T
@@ -111,7 +111,7 @@ func (b *TestBlockFinalizer) check(t *testing.T) {
 	require.Equal(b.t, uint64(len(b.fireBlocks)), b.idx, "we should have fired all %d blocks, only fired %d blocks", len(b.fireBlocks), b.idx)
 }
 
-var _ BlockFinalizer = &TestNoopBlockFinalizer{}
+var _ BlockHandler = &TestNoopBlockFinalizer{}
 
 type TestNoopBlockFinalizer struct{}
 

blockpoller/poller.go

@@ -19,23 +19,23 @@ type BlockPoller struct {
 	fetchBlockRetryCount uint64
 	stateStorePath       string
 
-	blockFetcher   BlockFetcher
-	blockFinalizer BlockFinalizer
-	forkDB         *forkable.ForkDB
+	blockFetcher BlockFetcher
+	blockHandler BlockHandler
+	forkDB       *forkable.ForkDB
 
 	logger *zap.Logger
 }
 
 func New(
 	blockFetcher BlockFetcher,
-	blockFinalizer BlockFinalizer,
+	blockFinalizer BlockHandler,
 	opts ...Option,
 ) *BlockPoller {
 
 	b := &BlockPoller{
 		Shutter:              shutter.New(),
 		blockFetcher:         blockFetcher,
-		blockFinalizer:       blockFinalizer,
+		blockHandler:         blockFinalizer,
 		fetchBlockRetryCount: 4,
 		logger:               zap.NewNop(),
 	}
@@ -56,7 +56,7 @@ func (p *BlockPoller) Run(ctx context.Context, startBlockNum uint64, finalizedBl
 		zap.Uint64("resolved_start_block_num", resolveStartBlockNum),
 	)
 
-	p.blockFinalizer.Init()
+	p.blockHandler.Init()
 
 	startBlock, err := p.blockFetcher.Fetch(ctx, resolveStartBlockNum)
 	if err != nil {
@@ -208,7 +208,7 @@ func (p *BlockPoller) fire(blk *block) (bool, error) {
 		return false, nil
 	}
 
-	if err := p.blockFinalizer.Fire(blk.Block); err != nil {
+	if err := p.blockHandler.Fire(blk.Block); err != nil {
 		return false, err
 	}
 

blockpoller/poller_test.go

@@ -222,7 +222,7 @@ func TestForkHandler_fire(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			poller := &BlockPoller{startBlockNumGate: test.startBlockNum, blockFinalizer: &TestNoopBlockFinalizer{}}
+			poller := &BlockPoller{startBlockNumGate: test.startBlockNum, blockHandler: &TestNoopBlockFinalizer{}}
 			ok, err := poller.fire(test.block)
 			require.NoError(t, err)
 			assert.Equal(t, test.expect, ok)

go.mod

@@ -12,7 +12,7 @@ require (
 	github.com/spf13/cobra v1.7.0
 	github.com/spf13/pflag v1.0.5
 	github.com/spf13/viper v1.15.0
-	github.com/streamingfast/bstream v0.0.2-0.20231121211820-e45c1b42f472
+	github.com/streamingfast/bstream v0.0.2-0.20231122151642-aa0d174c0e39
 	github.com/streamingfast/cli v0.0.4-0.20230825151644-8cc84512cd80
 	github.com/streamingfast/dauth v0.0.0-20230929180355-921f9c9be330
 	github.com/streamingfast/dbin v0.9.1-0.20231117225723-59790c798e2c

go.sum

@@ -574,10 +574,10 @@ github.com/spf13/pflag v1.0.5/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An
 github.com/spf13/viper v1.15.0 h1:js3yy885G8xwJa6iOISGFwd+qlUo5AvyXb7CiihdtiU=
 github.com/spf13/viper v1.15.0/go.mod h1:fFcTBJxvhhzSJiZy8n+PeW6t8l+KeT/uTARa0jHOQLA=
 github.com/stoewer/go-strcase v1.2.0/go.mod h1:IBiWB2sKIp3wVVQ3Y035++gc+knqhUQag1KpM8ahLw8=
-github.com/streamingfast/bstream v0.0.2-0.20231121140754-a458ffe57f0d h1:yPqseR0o2RFU2HNoAi6hqIszWr8WVoOQbnHKcEI2zZE=
-github.com/streamingfast/bstream v0.0.2-0.20231121140754-a458ffe57f0d/go.mod h1:08GVb+DXyz6jVNIsbf+2zlaC81UeEGu5o1h49KrSR3Y=
 github.com/streamingfast/bstream v0.0.2-0.20231121211820-e45c1b42f472 h1:HhWgNV7PIdKiK7IZIXLQ/3mNsERqLm5ztzLwn43AcOo=
 github.com/streamingfast/bstream v0.0.2-0.20231121211820-e45c1b42f472/go.mod h1:08GVb+DXyz6jVNIsbf+2zlaC81UeEGu5o1h49KrSR3Y=
+github.com/streamingfast/bstream v0.0.2-0.20231122151642-aa0d174c0e39 h1:4MP75Z2TGPh/TpaoqFZH1fDI9yEv6ZeShqqjKMb6r/A=
+github.com/streamingfast/bstream v0.0.2-0.20231122151642-aa0d174c0e39/go.mod h1:08GVb+DXyz6jVNIsbf+2zlaC81UeEGu5o1h49KrSR3Y=
 github.com/streamingfast/cli v0.0.4-0.20230825151644-8cc84512cd80 h1:UxJUTcEVkdZy8N77E3exz0iNlgQuxl4m220GPvzdZ2s=
 github.com/streamingfast/cli v0.0.4-0.20230825151644-8cc84512cd80/go.mod h1:QxjVH73Lkqk+mP8bndvhMuQDUINfkgsYhdCH/5TJFKI=
 github.com/streamingfast/dauth v0.0.0-20230929180355-921f9c9be330 h1:49JYZkn8ALGe+LhcACZyX3L9B8tIxRZ3F3l+OxmNMhY=
@@ -617,8 +617,6 @@ github.com/streamingfast/shutter v1.5.0 h1:NpzDYzj0HVpSiDJVO/FFSL6QIK/YKOxY0gJAt
 github.com/streamingfast/shutter v1.5.0/go.mod h1:B/T6efqdeMGbGwjzPS1ToXzYZI4kDzI5/u4I+7qbjY8=
 github.com/streamingfast/snapshotter v0.0.0-20230316190750-5bcadfde44d0 h1:Y15G1Z4fpEdm2b+/70owI7TLuXadlqBtGM7rk4Hxrzk=
 github.com/streamingfast/snapshotter v0.0.0-20230316190750-5bcadfde44d0/go.mod h1:/Rnz2TJvaShjUct0scZ9kKV2Jr9/+KBAoWy4UMYxgv4=
-github.com/streamingfast/substreams v1.1.21-0.20231120175501-9d89549d81a1 h1:PQeoATwMtCXKw2ztdS3ruwIOMzCj4GEGkkxB8LvcMOE=
-github.com/streamingfast/substreams v1.1.21-0.20231120175501-9d89549d81a1/go.mod h1:HimUVtUnRKCdWfkNKFdVgqGCcrNu49+Az5Cyzdpuc4Q=
 github.com/streamingfast/substreams v1.1.21-0.20231122013157-938ec26b0ef6 h1:eNA736ywGv/XXCotwx4LkfRDmOrHUyPbAqCKI+RzYI4=
 github.com/streamingfast/substreams v1.1.21-0.20231122013157-938ec26b0ef6/go.mod h1:JHCOsJtgXUM2KWNxsvi5/hjrLy4KpClaMRriBR3ybnI=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=