postgres

Documentation

Index

• type Checkpoint
  • func New(appName, tableName, connectionStr string, opts ...Option) (*Checkpoint, error)
  • func (c *Checkpoint) GetCheckpoint(ctx context.Context, streamName, shardID string) (string, error)
  • func (c *Checkpoint) GetMaxInterval() time.Duration
  • func (c *Checkpoint) SetCheckpoint(_ context.Context, streamName, shardID, sequenceNumber string) error
  • func (c *Checkpoint) Shutdown(ctx context.Context) error
• type Option
  • func WithConnection(connection *sql.DB) Option
  • func WithMaxInterval(maxInterval time.Duration) Option

Types

type Checkpoint

type Checkpoint struct {
	// contains filtered or unexported fields
}

Checkpoint stores and retrieves the last evaluated key from a DDB scan

func New

func New(appName, tableName, connectionStr string, opts ...Option) (*Checkpoint, error)

New returns a checkpoint that uses PostgresDB for underlying storage Using connectionStr turn it more flexible to use specific db configs

func (*Checkpoint) GetCheckpoint

func (c *Checkpoint) GetCheckpoint(ctx context.Context, streamName, shardID string) (string, error)

GetCheckpoint determines if a checkpoint for a particular Shard exists. Typically used to determine whether we should start processing the shard with TRIM_HORIZON or AFTER_SEQUENCE_NUMBER (if checkpoint exists).

func (*Checkpoint) GetMaxInterval

func (c *Checkpoint) GetMaxInterval() time.Duration

GetMaxInterval returns the maximum interval before the checkpoint

func (*Checkpoint) SetCheckpoint

func (c *Checkpoint) SetCheckpoint(_ context.Context, streamName, shardID, sequenceNumber string) error

SetCheckpoint stores a checkpoint for a shard (e.g. sequence number of last record processed by application). Upon fail over, record processing is resumed from this point.

func (*Checkpoint) Shutdown

func (c *Checkpoint) Shutdown(ctx context.Context) error

Shutdown the checkpoint. Save any in-flight data.

type Option

type Option func(*Checkpoint)

Option is used to override defaults when creating a new Checkpoint

func WithConnection

func WithConnection(connection *sql.DB) Option

WithConnection overwrites the default sql.DB

func WithMaxInterval

func WithMaxInterval(maxInterval time.Duration) Option

WithMaxInterval sets the flush interval