injector

Documentation

Index

• func PreprocessingFunc(blockFilter func(blk *bstream.Block) error) bstream.PreprocessFunc
• type FacetShardDetail
• type FacetShardSummary
• type Injector
  • func NewInjector(kvdb store.KVStore, blocksStore dstore.Store, ...) *Injector
  • func (i *Injector) DeleteCheckpoint(ctx context.Context, shard byte) error
  • func (i *Injector) FacetShardSummary(ctx context.Context, facet accounthist.Facet, shardNum byte) (*FacetShardDetail, error)
  • func (i *Injector) FacetShardsSummary(ctx context.Context, key accounthist.Facet) ([]*FacetShardSummary, error)
  • func (i *Injector) ForceFlush(ctx context.Context) error
  • func (i *Injector) GetShardCheckpoint(ctx context.Context) (*pbaccounthist.ShardCheckpoint, error)
  • func (i *Injector) Launch()
  • func (i *Injector) ProcessBlock(blk *bstream.Block, obj interface{}) error
  • func (i *Injector) SetFacetFactory(facetFactory accounthist.FacetFactory)
  • func (i *Injector) SetupMetrics(serviceName string)
  • func (i *Injector) SetupSource(ignoreCheckpointOnLaunch bool) error
  • func (i *Injector) ShardCheckpointAnalysis(ctx context.Context, checkpointPrefix byte) (out []*ShardDetail, err error)
  • func (i *Injector) Shutdown(err error)
  • func (i *Injector) UpdateSeqData(key accounthist.Facet, seqData accounthist.SequenceData)
  • func (i *Injector) WriteAction(ctx context.Context, key accounthist.Facet, ...) error
• type RWCache
  • func NewRWCache(backingStore store.KVStore) *RWCache
  • func (c *RWCache) BatchDelete(ctx context.Context, keys [][]byte) error
  • func (c *RWCache) FlushPuts(ctx context.Context) error
  • func (c *RWCache) OrderedPuts(f func(sKey string, value []byte) error) error
  • func (c *RWCache) Put(ctx context.Context, key, value []byte) error
• type ShardDetail

Functions

func PreprocessingFunc

func PreprocessingFunc(blockFilter func(blk *bstream.Block) error) bstream.PreprocessFunc

Types

type FacetShardDetail

type FacetShardDetail struct {
	ShardNum      byte
	LatestSeqData accounthist.SequenceData
	RowKeyCount   uint64
}

type FacetShardSummary

type FacetShardSummary struct {
	ShardNum byte
	SeqData  accounthist.SequenceData
}

type Injector

type Injector struct {
	*shutter.Shutter

	ShardNum byte // 0 is live

	BlockFilter func(blk *bstream.Block) error
	MaxEntries  uint64

	KvStore store.KVStore
	// contains filtered or unexported fields
}

func NewInjector

func NewInjector(
	kvdb store.KVStore,
	blocksStore dstore.Store,
	blockFilter func(blk *bstream.Block) error,
	shardNum byte,
	maxEntries uint64,
	flushBlocksInterval uint64,
	startBlockNum uint64,
	stopBlockNum uint64,
	tracker *bstream.Tracker,
) *Injector

func (*Injector) DeleteCheckpoint

func (i *Injector) DeleteCheckpoint(ctx context.Context, shard byte) error

func (*Injector) FacetShardSummary

func (i *Injector) FacetShardSummary(ctx context.Context, facet accounthist.Facet, shardNum byte) (*FacetShardDetail, error)

func (*Injector) FacetShardsSummary

func (i *Injector) FacetShardsSummary(ctx context.Context, key accounthist.Facet) ([]*FacetShardSummary, error)

func (*Injector) ForceFlush

func (i *Injector) ForceFlush(ctx context.Context) error

func (*Injector) GetShardCheckpoint

func (i *Injector) GetShardCheckpoint(ctx context.Context) (*pbaccounthist.ShardCheckpoint, error)

func (*Injector) Launch

func (i *Injector) Launch()

func (*Injector) ProcessBlock

func (i *Injector) ProcessBlock(blk *bstream.Block, obj interface{}) error

func (*Injector) SetFacetFactory

func (i *Injector) SetFacetFactory(facetFactory accounthist.FacetFactory)

func (*Injector) SetupMetrics

func (i *Injector) SetupMetrics(serviceName string)

func (*Injector) SetupSource

func (i *Injector) SetupSource(ignoreCheckpointOnLaunch bool) error

func (*Injector) ShardCheckpointAnalysis

func (i *Injector) ShardCheckpointAnalysis(ctx context.Context, checkpointPrefix byte) (out []*ShardDetail, err error)

func (*Injector) Shutdown

func (i *Injector) Shutdown(err error)

func (*Injector) UpdateSeqData

func (i *Injector) UpdateSeqData(key accounthist.Facet, seqData accounthist.SequenceData)

func (*Injector) WriteAction

func (i *Injector) WriteAction(ctx context.Context, key accounthist.Facet, acctSeqData accounthist.SequenceData, rawTrace []byte) error

type RWCache

type RWCache struct {
	store.KVStore
	// contains filtered or unexported fields
}

func NewRWCache

func NewRWCache(backingStore store.KVStore) *RWCache

func (*RWCache) BatchDelete

func (c *RWCache) BatchDelete(ctx context.Context, keys [][]byte) error

func (*RWCache) FlushPuts

func (c *RWCache) FlushPuts(ctx context.Context) error

func (*RWCache) OrderedPuts

func (c *RWCache) OrderedPuts(f func(sKey string, value []byte) error) error

func (*RWCache) Put

func (c *RWCache) Put(ctx context.Context, key, value []byte) error

type ShardDetail

type ShardDetail struct {
	ShardNum   byte
	Checkpoint *pbaccounthist.ShardCheckpoint
}