sinker

package

v1.20.2 Latest Latest Go to latest Published: Jul 24, 2026 License: Apache-2.0 Imports: 29 Imported by: 0

Details

Valid go.mod file
Redistributable license
Tagged version
Stable version
Learn more about best practices

Repository

github.com/streamingfast/substreams

Links

Documentation ¶

Index ¶

Constants
Variables
func RegisterMetrics()
func SinkerSetup(ctx context.Context, dsnString string, pkg *pbsubstreams.Package, ...) error
type GenerateCSVSinker
- func NewGenerateCSVSinker(sink *sink.Sinker, destFolder string, workingDir string, ...) (*GenerateCSVSinker, error)
type SQLSinker
- func New(sink *sink.Sinker, loader *db2.Loader, logger *zap.Logger, ...) (*SQLSinker, error)
type SinkerFactoryFunc
- func SinkerFactory(baseSink *sink.Sinker, options SinkerFactoryOptions) SinkerFactoryFunc
type SinkerFactoryOptions
type SinkerSetupOptions
type Stats
- func NewStats(logger *zap.Logger) *Stats

Constants ¶

View Source

const BLOCK_FLUSH_INTERVAL_DISABLED = 0

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 (in nanoseconds)")

View Source

var FlushedHeadBlockNumber = metrics.NewHeadBlockNumber("substreams_sink_postgres")

View Source

var FlushedHeadBlockTimeDrift = metrics.NewHeadTimeDrift("substreams_sink_postgres")

View Source

var FlushedRowsCount = metrics.NewCounter("substreams_sink_postgres_flushed_rows_count", "The number of flushed rows so far")

Functions ¶

func RegisterMetrics ¶

func RegisterMetrics()

func SinkerSetup ¶

func SinkerSetup(
	ctx context.Context,
	dsnString string,
	pkg *pbsubstreams.Package,
	options SinkerSetupOptions,
	logger *zap.Logger,
	tracer logging.Tracer,
) error

SinkerSetup sets up the required infrastructure for a Substreams SQL sink

Types ¶

type GenerateCSVSinker ¶

type GenerateCSVSinker struct {
	*shutter.Shutter
	*sink.Sinker
	// contains filtered or unexported fields
}

func NewGenerateCSVSinker ¶

func NewGenerateCSVSinker(
	sink *sink.Sinker,
	destFolder string,
	workingDir string,
	cursorTableName string,
	bundleSize uint64,
	bufferSize uint64,
	loader *db2.Loader,
	lastCursorFilename string,
	logger *zap.Logger,
	tracer logging.Tracer,
) (*GenerateCSVSinker, error)

func (*GenerateCSVSinker) HandleBlockRangeCompletion ¶

func (s *GenerateCSVSinker) HandleBlockRangeCompletion(ctx context.Context, cursor *sink.Cursor) error

func (*GenerateCSVSinker) HandleBlockScopedData ¶

func (s *GenerateCSVSinker) HandleBlockScopedData(ctx context.Context, data *pbsubstreamsrpc.BlockScopedData, isLive *bool, cursor *sink.Cursor) error

func (*GenerateCSVSinker) HandleBlockUndoSignal ¶

func (s *GenerateCSVSinker) HandleBlockUndoSignal(ctx context.Context, data *pbsubstreamsrpc.BlockUndoSignal, cursor *sink.Cursor) error

func (*GenerateCSVSinker) Run ¶

func (s *GenerateCSVSinker) Run(ctx context.Context)

type SQLSinker ¶

type SQLSinker struct {
	*shutter.Shutter
	*sink.Sinker
	// contains filtered or unexported fields
}

func New ¶

func New(sink *sink.Sinker, loader *db2.Loader, logger *zap.Logger, tracer logging.Tracer, flushRetryCount int, flushRetryDelay time.Duration) (*SQLSinker, error)

func (*SQLSinker) Close ¶

func (s *SQLSinker) Close() error

func (*SQLSinker) HandleBlockRangeCompletion ¶

func (s *SQLSinker) HandleBlockRangeCompletion(ctx context.Context, cursor *sink.Cursor) error

func (*SQLSinker) HandleBlockScopedData ¶

func (s *SQLSinker) HandleBlockScopedData(ctx context.Context, data *pbsubstreamsrpc.BlockScopedData, isLive *bool, cursor *sink.Cursor) error

func (*SQLSinker) HandleBlockUndoSignal ¶

func (s *SQLSinker) HandleBlockUndoSignal(ctx context.Context, data *pbsubstreamsrpc.BlockUndoSignal, cursor *sink.Cursor) error

func (*SQLSinker) Run ¶

func (s *SQLSinker) Run(ctx context.Context)

type SinkerFactoryFunc ¶

type SinkerFactoryFunc func(ctx context.Context, dsnString string, logger *zap.Logger, tracer logging.Tracer) (*SQLSinker, error)

func SinkerFactory ¶

func SinkerFactory(
	baseSink *sink.Sinker,
	options SinkerFactoryOptions,
) SinkerFactoryFunc

type SinkerFactoryOptions ¶

type SinkerFactoryOptions struct {
	CursorTableName         string
	HistoryTableName        string
	ClickhouseCluster       string
	BatchBlockFlushInterval int
	BatchRowFlushInterval   int
	LiveBlockFlushInterval  int
	OnModuleHashMismatch    string
	HandleReorgs            bool
	FlushRetryCount         int
	FlushRetryDelay         time.Duration
}

type SinkerSetupOptions ¶

type SinkerSetupOptions struct {
	CursorTableName            string
	HistoryTableName           string
	ClickhouseCluster          string
	OnModuleHashMismatch       string
	SystemTablesOnly           bool
	IgnoreDuplicateTableErrors bool
	Postgraphile               bool
}

SinkerSetupOptions contains configuration for the setup operation

type Stats ¶

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

func NewStats ¶

func NewStats(logger *zap.Logger) *Stats

func (*Stats) AverageFlushDuration ¶

func (s *Stats) AverageFlushDuration() time.Duration

func (*Stats) Close ¶

func (s *Stats) Close()

func (*Stats) LogNow ¶

func (s *Stats) LogNow()

func (*Stats) RecordBlock ¶

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

func (*Stats) RecordFlushDuration ¶

func (s *Stats) RecordFlushDuration(duration time.Duration)

func (*Stats) RecordHandleBlockDuration ¶

func (s *Stats) RecordHandleBlockDuration(duration time.Duration)

func (*Stats) RecordHandleUndoDuration ¶

func (s *Stats) RecordHandleUndoDuration(duration time.Duration)

func (*Stats) Start ¶

func (s *Stats) Start(each time.Duration, cursor *sink.Cursor)

Source Files ¶

View all Source files

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