db

Documentation

Index

• Constants
• func NewDB(ctx context.Context, dsn string, waitForDB, statementTimeout time.Duration) (*sql.DB, error)
• func NewNamespacedDB(ctx context.Context, dsn string, namespace string, ...) (*sql.DB, error)
• func NullInt32(v int32) sql.NullInt32
• func NullInt64(v int64) sql.NullInt64
• func RunInTx(ctx context.Context, db *sql.DB, opts *sql.TxOptions, ...) error
• func SetVectorClock(q *queries.SelectGatewayEnvelopesParams, vc VectorClock) *queries.SelectGatewayEnvelopesParams
• type DBSubscription
  • func NewDBSubscription[ValueType any, CursorType any](ctx context.Context, log *zap.Logger, ...) *DBSubscription[ValueType, CursorType]
  • func (s *DBSubscription[ValueType, CursorType]) Start() (<-chan []ValueType, error)
• type PollableDBQuery
• type PollingOptions
• type Topic
• type VectorClock
  • func ToVectorClock(rows []queries.SelectVectorClockRow) VectorClock

Constants

const MAX_NAMESPACE_LENGTH = 32

Functions

func NewDB

func NewDB(
	ctx context.Context,
	dsn string,
	waitForDB, statementTimeout time.Duration,
) (*sql.DB, error)

func NewNamespacedDB

func NewNamespacedDB(
	ctx context.Context,
	dsn string,
	namespace string,
	waitForDB, statementTimeout time.Duration,
) (*sql.DB, error)

Creates a new database with the given namespace if it doesn't exist and returns the full DSN for the new database.

func NullInt32

func NullInt32(v int32) sql.NullInt32

func NullInt64

func NullInt64(v int64) sql.NullInt64

func RunInTx

func RunInTx(
	ctx context.Context,
	db *sql.DB,
	opts *sql.TxOptions,
	fn func(ctx context.Context, txQueries *queries.Queries) error,
) error

func SetVectorClock

func SetVectorClock(
	q *queries.SelectGatewayEnvelopesParams,
	vc VectorClock,
) *queries.SelectGatewayEnvelopesParams

Types

type DBSubscription

type DBSubscription[ValueType any, CursorType any] struct {
	// contains filtered or unexported fields
}

A subscription that polls a DB for updates Assumes there is only one listener (updates block on a single unbuffered channel)

func NewDBSubscription

func NewDBSubscription[ValueType any, CursorType any](
	ctx context.Context,
	log *zap.Logger,
	query PollableDBQuery[ValueType, CursorType],
	lastSeen CursorType,
	options PollingOptions,
) *DBSubscription[ValueType, CursorType]

func (*DBSubscription[ValueType, CursorType]) Start

func (s *DBSubscription[ValueType, CursorType]) Start() (<-chan []ValueType, error)

type PollableDBQuery

type PollableDBQuery[ValueType any, CursorType any] func(
	ctx context.Context,
	lastSeen CursorType,
	numRows int32,
) (results []ValueType, nextCursor CursorType, err error)

type PollingOptions

type PollingOptions struct {
	Interval time.Duration
	Notifier <-chan bool
	NumRows  int32
}

Poll whenever notified, or at an interval if not notified

type Topic

type Topic = []byte

type VectorClock

type VectorClock = map[uint32]uint64

func ToVectorClock

func ToVectorClock(rows []queries.SelectVectorClockRow) VectorClock