checkpoint

Documentation

Overview

Package checkpoint implements a mechanism for tracking checkpointed integer offsets for sequential read at-least-once queue systems such as Kafka or Kinesis.

Index

• type Capped
  • func NewCapped(cap int) *Capped
  • func (c *Capped) Highest() int
  • func (c *Capped) Resolve(offset int) (int, error)
  • func (c *Capped) Track(ctx context.Context, i int) error

Types

type Capped

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

Capped receives an ordered feed of integer based offsets being tracked, and an unordered feed of integer based offsets that are resolved, and is able to return the highest offset currently able to be committed such that an unresolved offset is never committed.

If the number of unresolved tracked values meets a given cap the next attempt to track a value will be blocked until the next value is resolved.

This component is safe to use concurrently across goroutines.

func NewCapped

func NewCapped(cap int) *Capped

NewCapped returns a new capped checkpointer.

func (*Capped) Highest

func (c *Capped) Highest() int

Highest returns the current highest checkpoint.

func (*Capped) Resolve

func (c *Capped) Resolve(offset int) (int, error)

Resolve a tracked offset by allowing it to be committed. The highest possible offset to be committed is returned, or an error if the provided offset was not recognised.

func (*Capped) Track

func (c *Capped) Track(ctx context.Context, i int) error

Track a new unresolved integer offset. This offset will be cached until it is marked as resolved. While it is cached no higher valued offset will ever be committed. If the provided value is lower than an already provided value an error is returned.