diff --git a/observer/block_peer_common.go b/observer/block_peer_common.go index aab3f16..2386943 100644 --- a/observer/block_peer_common.go +++ b/observer/block_peer_common.go @@ -19,7 +19,8 @@ type ( peerChannels PeerChannels blockDeliverer api.BlocksDeliverer channelObservers map[string]*BlockPeerChannel - seekFrom map[string]SeekFromFetcher + seekFromFetcher SeekFromFetcher + seekFromMap map[string]uint64 observePeriod time.Duration stopRecreateStream bool logger *zap.Logger @@ -37,7 +38,8 @@ type ( } BlockPeerOpts struct { - seekFrom map[string]SeekFromFetcher + seekFromFetcher SeekFromFetcher + seekFromMap map[string]uint64 observePeriod time.Duration stopRecreateStream bool logger *zap.Logger @@ -62,9 +64,15 @@ func WithBlockPeerLogger(logger *zap.Logger) BlockPeerOpt { } } -func WithSeekFrom(seekFrom map[string]SeekFromFetcher) BlockPeerOpt { +func WithSeekFromFetcher(seekFromFetcher SeekFromFetcher) BlockPeerOpt { return func(opts *BlockPeerOpts) { - opts.seekFrom = seekFrom + opts.seekFromFetcher = seekFromFetcher + } +} + +func WithSeekFromMap(seekFromMap map[string]uint64) BlockPeerOpt { + return func(opts *BlockPeerOpts) { + opts.seekFromMap = seekFromMap } } @@ -94,7 +102,8 @@ func NewBlockPeer(peerChannels PeerChannels, blockDeliverer api.BlocksDeliverer, channelObservers: make(map[string]*BlockPeerChannel), blocks: make(chan *Block), blocksByChannels: make(map[string]chan *Block), - seekFrom: blockPeerOpts.seekFrom, + seekFromFetcher: blockPeerOpts.seekFromFetcher, + seekFromMap: blockPeerOpts.seekFromMap, observePeriod: blockPeerOpts.observePeriod, stopRecreateStream: blockPeerOpts.stopRecreateStream, logger: blockPeerOpts.logger, @@ -181,9 +190,9 @@ func (bp *BlockPeer) initChannels(ctx context.Context) { } func (bp *BlockPeer) peerChannel(ctx context.Context, channel string) *BlockPeerChannel { - seekFrom, exist := bp.seekFrom[channel] - if !exist { - seekFrom = ChannelSeekOldest() + seekFrom := bp.seekFromFetcher + if seekFrom == nil { + seekFrom = ChannelSeekFrom(bp.seekFromMap[channel]) } peerChannel := &BlockPeerChannel{} diff --git a/observer/block_peer_common_concurrently.go b/observer/block_peer_common_concurrently.go index 75add18..0320bbb 100644 --- a/observer/block_peer_common_concurrently.go +++ b/observer/block_peer_common_concurrently.go @@ -64,9 +64,9 @@ func (bp *BlockPeer) initChannelsConcurrently(ctx context.Context, blocksByChann } func (bp *BlockPeer) peerChannelConcurrently(ctx context.Context, channel string, blocksByChannels *BlocksByChannels) *BlockPeerChannel { - seekFrom, exist := bp.seekFrom[channel] - if !exist { - seekFrom = ChannelSeekOldest() + seekFrom := bp.seekFromFetcher + if seekFrom == nil { + seekFrom = ChannelSeekFrom(bp.seekFromMap[channel]) } peerChannel := &BlockPeerChannel{} diff --git a/observer/block_peer_parsed.go b/observer/block_peer_parsed.go index a06af51..55deaba 100644 --- a/observer/block_peer_parsed.go +++ b/observer/block_peer_parsed.go @@ -144,9 +144,9 @@ func (pbp *ParsedBlockPeer) initParsedChannels(ctx context.Context) { } func (pbp *ParsedBlockPeer) peerParsedChannel(ctx context.Context, channel string) *ParsedBlockPeerChannel { - seekFrom, exist := pbp.blockPeer.seekFrom[channel] - if !exist { - seekFrom = ChannelSeekOldest() + seekFrom := pbp.blockPeer.seekFromFetcher + if seekFrom == nil { + seekFrom = ChannelSeekFrom(pbp.blockPeer.seekFromMap[channel]) } commonBlockChannel := NewBlockChannel( diff --git a/observer/block_peer_parsed_concurrently.go b/observer/block_peer_parsed_concurrently.go index 6c96b0e..351d041 100644 --- a/observer/block_peer_parsed_concurrently.go +++ b/observer/block_peer_parsed_concurrently.go @@ -66,9 +66,9 @@ func (pbp *ParsedBlockPeer) initParsedChannelsConcurrently(ctx context.Context, } func (pbp *ParsedBlockPeer) peerParsedChannelConcurrently(ctx context.Context, channel string, blocksByChannels *ParsedBlocksByChannels) *ParsedBlockPeerChannel { - seekFrom, exist := pbp.blockPeer.seekFrom[channel] - if !exist { - seekFrom = ChannelSeekOldest() + seekFrom := pbp.blockPeer.seekFromFetcher + if seekFrom == nil { + seekFrom = ChannelSeekFrom(pbp.blockPeer.seekFromMap[channel]) } commonBlockChannel := NewBlockChannel(