hub

package

v0.0.2-0...-607e840 Latest Latest Go to latest Published: Apr 2, 2026 License: Apache-2.0 Imports: 18 Imported by: 0

Details

Valid go.mod file
Redistributable license
Tagged version
Stable version
Learn more about best practices

Repository

github.com/streamingfast/bstream

Links

Open Source Insights

Documentation ¶

Index ¶

Variables
type BlockstreamServer
type ForkableHub
- func NewForkableHub(liveSourceFactory bstream.SourceFactory, keepFinalBlocks int, ...) *ForkableHub
- func NewForkableHubWithOptions(liveSourceFactory bstream.SourceFactory, keepFinalBlocks int, ...) *ForkableHub
type Option
- func WithMaxConsecutiveUnlinkableBlocks(count int) Option
type Subscription
- func NewSubscription(handler bstream.Handler, chanSize int, withPartial bool) *Subscription
- func (s *Subscription) Run()

Constants ¶

This section is empty.

Variables ¶

View Source

var ErrSubscriptionChannelFull = fmt.Errorf("subscription channel at max capacity")

Functions ¶

This section is empty.

Types ¶

type BlockstreamServer ¶

type BlockstreamServer struct {
	// contains filtered or unexported fields
}

func (*BlockstreamServer) Blocks ¶

func (s *BlockstreamServer) Blocks(r *pbbstream.BlockRequest, stream pbbstream.BlockStream_BlocksServer) error

func (*BlockstreamServer) Close ¶

func (s *BlockstreamServer) Close()

func (*BlockstreamServer) GetHeadInfo ¶

func (s *BlockstreamServer) GetHeadInfo(ctx context.Context, req *pbheadinfo.HeadInfoRequest) (*pbheadinfo.HeadInfoResponse, error)

func (*BlockstreamServer) Launch ¶

func (s *BlockstreamServer) Launch(serverAddr string)

type ForkableHub ¶

type ForkableHub struct {
	*shutter.Shutter

	Ready chan struct{}
	// contains filtered or unexported fields
}

ForkableHub gives you block Sources for blocks close to head it keeps reversible segment in a Forkable it keeps small final segment in a buffer

func NewForkableHub ¶

func NewForkableHub(liveSourceFactory bstream.SourceFactory, keepFinalBlocks int, oneBlocksStore dstore.Store, extraForkableOptions ...forkable.Option) *ForkableHub

func NewForkableHubWithOptions ¶

func NewForkableHubWithOptions(liveSourceFactory bstream.SourceFactory, keepFinalBlocks int, oneBlocksStore dstore.Store, hubOptions []Option, extraForkableOptions ...forkable.Option) *ForkableHub

NewForkableHubWithOptions is like NewForkableHub but also accepts hub-level options such as WithMaxConsecutiveUnlinkableBlocks.

func (*ForkableHub) GetBlock ¶

func (h *ForkableHub) GetBlock(num uint64, id string) (out *pbbstream.Block)

func (*ForkableHub) GetBlockByHash ¶

func (h *ForkableHub) GetBlockByHash(id string) (out *pbbstream.Block)

func (*ForkableHub) HeadInfo ¶

func (h *ForkableHub) HeadInfo() (headNum uint64, headID string, headTime time.Time, libNum uint64, err error)

func (*ForkableHub) HeadNum ¶

func (h *ForkableHub) HeadNum() uint64

func (*ForkableHub) IsReady ¶

func (h *ForkableHub) IsReady() bool

func (*ForkableHub) LowestBlockNum ¶

func (h *ForkableHub) LowestBlockNum() uint64

func (*ForkableHub) MatchSuffix ¶

func (h *ForkableHub) MatchSuffix(req string) bool

func (*ForkableHub) NewBlockstreamServer ¶

func (h *ForkableHub) NewBlockstreamServer(dgrpcServer ggrpcserver.Server) *BlockstreamServer

implementation of blockstream.Server from the hub

func (*ForkableHub) ProcessBlock ¶

func (h *ForkableHub) ProcessBlock(blk *pbbstream.Block, obj any) error

func (*ForkableHub) Run ¶

func (h *ForkableHub) Run()

func (*ForkableHub) SourceFromBlockNum ¶

func (h *ForkableHub) SourceFromBlockNum(num uint64, handler bstream.Handler) (out bstream.Source)

func (*ForkableHub) SourceFromBlockNumWithForks ¶

func (h *ForkableHub) SourceFromBlockNumWithForks(num uint64, handler bstream.Handler, withPartials bool) (out bstream.Source)

func (*ForkableHub) SourceFromCursor ¶

func (h *ForkableHub) SourceFromCursor(cursor *bstream.Cursor, handler bstream.Handler) (out bstream.Source)

func (*ForkableHub) SourceThroughCursor ¶

func (h *ForkableHub) SourceThroughCursor(startBlock uint64, cursor *bstream.Cursor, handler bstream.Handler) (out bstream.Source)

func (*ForkableHub) WalkOneBlocksStore ¶

func (h *ForkableHub) WalkOneBlocksStore(ctx context.Context) ([]string, error)

func (*ForkableHub) WalkOneBlocksStoreFrom ¶

func (h *ForkableHub) WalkOneBlocksStoreFrom(ctx context.Context, startingBlock uint64) ([]string, error)

func WithMaxConsecutiveUnlinkableBlocks ¶

func WithMaxConsecutiveUnlinkableBlocks(count int) Option

WithMaxConsecutiveUnlinkableBlocks instructs the hub to shut itself down (with errRestartRequired) if it has already passed readiness and it receives count consecutive blocks that cannot be linked to its current head, even after attempting to fill the gap from the one-block store. A count of 0 (the default) disables the check.

type Subscription ¶

type Subscription struct {
	*shutter.Shutter
	// contains filtered or unexported fields
}

Subscription is a bstream.Source

func NewSubscription ¶

func NewSubscription(handler bstream.Handler, chanSize int, withPartial bool) *Subscription

s.hub.unsubscribe(sub)

func (*Subscription) Run ¶

func (s *Subscription) Run()

Source Files ¶

View all Source files

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL