Versions in this module Expand all Collapse all v1 v1.20.2 Jul 24, 2026 Changes in this version + const BLOCK_FLUSH_INTERVAL_DISABLED + var FlushCount = metrics.NewCounter("substreams_sink_postgres_store_flush_count", ...) + var FlushDuration = metrics.NewCounter("substreams_sink_postgres_store_flush_duration", ...) + var FlushedHeadBlockNumber = metrics.NewHeadBlockNumber("substreams_sink_postgres") + var FlushedHeadBlockTimeDrift = metrics.NewHeadTimeDrift("substreams_sink_postgres") + var FlushedRowsCount = metrics.NewCounter("substreams_sink_postgres_flushed_rows_count", ...) + func RegisterMetrics() + func SinkerSetup(ctx context.Context, dsnString string, pkg *pbsubstreams.Package, ...) error + type GenerateCSVSinker struct + func NewGenerateCSVSinker(sink *sink.Sinker, destFolder string, workingDir string, ...) (*GenerateCSVSinker, error) + func (s *GenerateCSVSinker) HandleBlockRangeCompletion(ctx context.Context, cursor *sink.Cursor) error + func (s *GenerateCSVSinker) HandleBlockScopedData(ctx context.Context, data *pbsubstreamsrpc.BlockScopedData, isLive *bool, ...) error + func (s *GenerateCSVSinker) HandleBlockUndoSignal(ctx context.Context, data *pbsubstreamsrpc.BlockUndoSignal, ...) error + func (s *GenerateCSVSinker) Run(ctx context.Context) + type SQLSinker struct + func New(sink *sink.Sinker, loader *db2.Loader, logger *zap.Logger, ...) (*SQLSinker, error) + func (s *SQLSinker) Close() error + func (s *SQLSinker) HandleBlockRangeCompletion(ctx context.Context, cursor *sink.Cursor) error + func (s *SQLSinker) HandleBlockScopedData(ctx context.Context, data *pbsubstreamsrpc.BlockScopedData, isLive *bool, ...) error + func (s *SQLSinker) HandleBlockUndoSignal(ctx context.Context, data *pbsubstreamsrpc.BlockUndoSignal, ...) error + func (s *SQLSinker) Run(ctx context.Context) + type SinkerFactoryFunc func(ctx context.Context, dsnString string, logger *zap.Logger, ...) (*SQLSinker, error) + func SinkerFactory(baseSink *sink.Sinker, options SinkerFactoryOptions) SinkerFactoryFunc + type SinkerFactoryOptions struct + BatchBlockFlushInterval int + BatchRowFlushInterval int + ClickhouseCluster string + CursorTableName string + FlushRetryCount int + FlushRetryDelay time.Duration + HandleReorgs bool + HistoryTableName string + LiveBlockFlushInterval int + OnModuleHashMismatch string + type SinkerSetupOptions struct + ClickhouseCluster string + CursorTableName string + HistoryTableName string + IgnoreDuplicateTableErrors bool + OnModuleHashMismatch string + Postgraphile bool + SystemTablesOnly bool + type Stats struct + func NewStats(logger *zap.Logger) *Stats + func (s *Stats) AverageFlushDuration() time.Duration + func (s *Stats) Close() + func (s *Stats) LogNow() + func (s *Stats) RecordBlock(block bstream.BlockRef) + func (s *Stats) RecordFlushDuration(duration time.Duration) + func (s *Stats) RecordHandleBlockDuration(duration time.Duration) + func (s *Stats) RecordHandleUndoDuration(duration time.Duration) + func (s *Stats) Start(each time.Duration, cursor *sink.Cursor)
Go to main page