hub

type HubSource ¶

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

HubSource has the following guarantees: * if a ProcessBlock() returns an error to this source, ProcessBlock will not be called again by this source. * if a ProcessBlock() calls source.Shutdown(), the source will not call ProcessBlock again. (this was an issue with BufferedHandler solution) WARNING: because of that second guarantee, allowing ProcessBlock to call its own source's shutdown function, you should never assume that there is no ongoing or incoming ProcessBlock() when calling source.Shutdown() from another thread. Don't do anything dangerous like closing a channel awaiting a Write from that source! Let the channel die from garbage collecting, wrap your reads and writes with select() on context.Terminating() or source.Terminating() and so on.

func (s *HubSource) Run()

type LIBGetter ¶

type LIBGetter interface {
	GetRealtimeLIB() (uint64, error)
}

type Option ¶

type Option func(*SubscriptionHub)

func WithName(name string) Option

func WithRealtimeTolerance(d time.Duration) Option

func WithSourceChannelSize(size int) Option

type SubscriptionHub ¶

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

SubscriptionHub hooks to a live data source

func NewSubscriptionHub(startBlock uint64, buffer *bstream.Buffer, tailLockFunc TailLockFunc, fileSourceFactory bstream.SourceFromNumFactory, liveSourceFactory bstream.SourceFromNumFactory, opts ...Option) (*SubscriptionHub, error)

func (h *SubscriptionHub) HeadBlock() bstream.BlockRef

func (h *SubscriptionHub) HeadBlockID() string

func (h *SubscriptionHub) Launch()

func (h *SubscriptionHub) NewHubSourceFromBlockNum(blockNum uint64, handler bstream.Handler) (*HubSource, error)

func (h *SubscriptionHub) NewSource(handler bstream.Handler, burst int) bstream.Source

func (h *SubscriptionHub) NewSourceFromBlockNum(blockNum uint64, handler bstream.Handler) bstream.Source

func (h *SubscriptionHub) NewSourceFromBlockNumWithOpts(blockNum uint64, handler bstream.Handler, opts ...bstream.JoiningSourceOption) bstream.Source

func (h *SubscriptionHub) NewSourceFromBlockRef(ref bstream.BlockRef, handler bstream.Handler) bstream.Source

func (h *SubscriptionHub) WaitReady()

func (h *SubscriptionHub) WaitUntilRealTime()

type TailLockFunc ¶

type TailLockFunc func(tailBlockNum uint64) (func(), error)

type TestLIBGetter ¶

type TestLIBGetter struct {
	LIB uint64
}

func (g *TestLIBGetter) GetRealtimeLIB() (uint64, error)