sinker

package

v1.0.0 Latest Latest Go to latest Published: Apr 19, 2023 License: Apache-2.0 Imports: 19 Imported by: 0

Documentation ¶

Index ¶

Constants
Variables
func RegisterMetrics()
type Config
type PostgresSinker
- func New(config *Config, logger *zap.Logger, tracer logging.Tracer) (*PostgresSinker, error)
type Stats
- func NewStats(logger *zap.Logger) *Stats

Constants ¶

const (
	DEFAULT_BLOCK_PROGRESS = 1000
	LIVE_BLOCK_PROGRESS    = 1
)

Variables ¶

View Source

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

View Source

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

View Source

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)

Source Files ¶

View all Source files

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