sinker

Documentation

Index

• Constants
• func RegisterMetrics()
• type Config
• type MongoSinker
  • func New(config *Config, logger *zap.Logger, tracer logging.Tracer) (*MongoSinker, error)
  • func (s *MongoSinker) Run(ctx context.Context) error
  • func (s *MongoSinker) Start(ctx context.Context) error
  • func (s *MongoSinker) Stop(ctx context.Context, err error)

Constants

const (
	BLOCK_PROGRESS      = 1000
	LIVE_BLOCK_PROGRESS = 1
)

Functions

func RegisterMetrics

func RegisterMetrics()

Types

type Config

type Config struct {
	DBLoader *mongo.Loader
	DBName   string

	DDL mongo.Tables

	UndoBufferSize     int
	LiveBlockTimeDelta time.Duration

	BlockRange       string
	Pkg              *pbsubstreams.Package
	OutputModule     *pbsubstreams.Module
	OutputModuleName string
	OutputModuleHash manifest.ModuleHash
	ClientConfig     *client.SubstreamsClientConfig
}

type MongoSinker

type MongoSinker struct {
	*shutter.Shutter

	DBLoader *mongo.Loader
	Tables   mongo.Tables

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

	UndoBufferSize  int
	LivenessTracker *sink.LivenessChecker
	// contains filtered or unexported fields
}

func New

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

func (*MongoSinker) Run

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

func (*MongoSinker) Start

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

func (*MongoSinker) Stop

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