sinker

Documentation

Index

• Constants
• Variables
• func RegisterMetrics()
• type Config
• type PostgresSinker
  • func New(config *Config, logger *zap.Logger, tracer logging.Tracer) (*PostgresSinker, error)
  • func (s *PostgresSinker) Run(ctx context.Context) error
  • func (s *PostgresSinker) Start(ctx context.Context) error
  • func (s *PostgresSinker) Stop(ctx context.Context, err error)
• type Stats
  • func NewStats(logger *zap.Logger) *Stats
  • func (s *Stats) Close()
  • func (s *Stats) RecordBlock(block bstream.BlockRef)
  • func (s *Stats) Start(each time.Duration)

Constants

const (
	DEFAULT_BLOCK_PROGRESS = 1000
	LIVE_BLOCK_PROGRESS    = 1
)

Variables

var FlushCount = metrics.NewCounter("substreams_sink_postgres_store_flush_count", "The amount of flush that happened so far")

var FlushDuration = metrics.NewCounter("substreams_sink_postgres_store_flush_duration", "The amount of time spent flushing cache to db")

var FlushedEntriesCount = metrics.NewCounter("substreams_sink_postgres_flushed_entries_count", "The number of flushed entries")

Functions

func RegisterMetrics

func RegisterMetrics()

Types

type Config

type Config struct {
	DBLoader         *db.Loader
	BlockRange       string
	Pkg              *pbsubstreams.Package
	OutputModule     *pbsubstreams.Module
	OutputModuleName string
	OutputModuleHash manifest.ModuleHash
	ClientConfig     *client.SubstreamsClientConfig

	UndoBufferSize     int
	LiveBlockTimeDelta time.Duration
	FlushInterval      int

	SubstreamsDevelopmentMode bool
	IrreversibleOnly          bool

	InfiniteRetry bool
}

type PostgresSinker

type PostgresSinker struct {
	*shutter.Shutter

	DBLoader         *db.Loader
	Pkg              *pbsubstreams.Package
	OutputModule     *pbsubstreams.Module
	OutputModuleName string
	OutputModuleHash manifest.ModuleHash
	ClientConfig     *client.SubstreamsClientConfig

	UndoBufferSize  int
	LivenessTracker *sink.LivenessChecker
	FlushInterval   int

	InfiniteRetry bool

	SubstreamsDevelopmentMode bool
	IrreversibleOnly          bool
	// contains filtered or unexported fields
}

func New

func New(config *Config, logger *zap.Logger, tracer logging.Tracer) (*PostgresSinker, error)

func (*PostgresSinker) Run

func (s *PostgresSinker) Run(ctx context.Context) error

func (*PostgresSinker) Start

func (s *PostgresSinker) Start(ctx context.Context) error

func (*PostgresSinker) Stop

func (s *PostgresSinker) Stop(ctx context.Context, err error)

type Stats

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

func NewStats

func NewStats(logger *zap.Logger) *Stats

func (*Stats) Close

func (s *Stats) Close()

func (*Stats) RecordBlock

func (s *Stats) RecordBlock(block bstream.BlockRef)

func (*Stats) Start

func (s *Stats) Start(each time.Duration)