buffer

Documentation

Index

• func SetDefaults(flushPeriodMilliseconds int, flushItemsThreshold int)
• type BulkEventWriter
  • func NewBulkEventWriter(pool *pgxpool.Pool, v validator.Validator, l *zerolog.Logger) (*BulkEventWriter, error)
  • func (w *BulkEventWriter) BulkWriteStepRunEvents(ctx context.Context, opts []*repository.CreateStepRunEventOpts) ([]int, error)
  • func (w *BulkEventWriter) Cleanup() error
• type BulkQueueStepRunOpts
• type BulkSemaphoreReleaser
  • func NewBulkSemaphoreReleaser(pool *pgxpool.Pool, v validator.Validator, l *zerolog.Logger) (*BulkSemaphoreReleaser, error)
  • func (w *BulkSemaphoreReleaser) BulkReleaseSemaphores(ctx context.Context, opts []SemaphoreReleaseOpts) ([]pgtype.UUID, error)
  • func (w *BulkSemaphoreReleaser) Cleanup() error
• type BulkStepRunQueuer
  • func NewBulkStepRunQueuer(pool *pgxpool.Pool, v validator.Validator, l *zerolog.Logger) (*BulkStepRunQueuer, error)
  • func (w *BulkStepRunQueuer) BulkQueueStepRuns(ctx context.Context, opts []BulkQueueStepRunOpts) ([]pgtype.UUID, error)
  • func (w *BulkStepRunQueuer) Cleanup() error
• type FlushResponse
• type IngestBuf
  • func NewIngestBuffer[T any, U any](opts IngestBufOpts[T, U]) *IngestBuf[T, U]
  • func (b *IngestBuf[T, U]) BuffItem(item T) (chan *FlushResponse[U], error)
  • func (b *IngestBuf[T, U]) Start() (func() error, error)
  • func (b *IngestBuf[T, U]) StartDebugLoop()
• type IngestBufOpts
• type SemaphoreReleaseOpts
• type TenantBufManagerOpts
• type TenantBufferManager
  • func NewTenantBufManager[T any, U any](opts TenantBufManagerOpts[T, U]) (*TenantBufferManager[T, U], error)
  • func (t *TenantBufferManager[T, U]) BuffItem(tenantKey string, eventOps T) (chan *FlushResponse[U], error)
  • func (t *TenantBufferManager[T, U]) Cleanup() error

Functions

func SetDefaults

func SetDefaults(flushPeriodMilliseconds int, flushItemsThreshold int)

Types

type BulkEventWriter

type BulkEventWriter struct {
	*TenantBufferManager[*repository.CreateStepRunEventOpts, int]
	// contains filtered or unexported fields
}

func NewBulkEventWriter

func NewBulkEventWriter(pool *pgxpool.Pool, v validator.Validator, l *zerolog.Logger) (*BulkEventWriter, error)

func (*BulkEventWriter) BulkWriteStepRunEvents

func (w *BulkEventWriter) BulkWriteStepRunEvents(ctx context.Context, opts []*repository.CreateStepRunEventOpts) ([]int, error)

func (*BulkEventWriter) Cleanup

func (w *BulkEventWriter) Cleanup() error

type BulkQueueStepRunOpts

type BulkQueueStepRunOpts struct {
	*dbsqlc.GetStepRunForEngineRow

	Priority int
	IsRetry  bool
	Input    []byte
}

type BulkSemaphoreReleaser

type BulkSemaphoreReleaser struct {
	*TenantBufferManager[SemaphoreReleaseOpts, pgtype.UUID]
	// contains filtered or unexported fields
}

func NewBulkSemaphoreReleaser

func NewBulkSemaphoreReleaser(pool *pgxpool.Pool, v validator.Validator, l *zerolog.Logger) (*BulkSemaphoreReleaser, error)

func (*BulkSemaphoreReleaser) BulkReleaseSemaphores

func (w *BulkSemaphoreReleaser) BulkReleaseSemaphores(ctx context.Context, opts []SemaphoreReleaseOpts) ([]pgtype.UUID, error)

func (*BulkSemaphoreReleaser) Cleanup

func (w *BulkSemaphoreReleaser) Cleanup() error

type BulkStepRunQueuer

type BulkStepRunQueuer struct {
	*TenantBufferManager[BulkQueueStepRunOpts, pgtype.UUID]
	// contains filtered or unexported fields
}

func NewBulkStepRunQueuer

func NewBulkStepRunQueuer(pool *pgxpool.Pool, v validator.Validator, l *zerolog.Logger) (*BulkStepRunQueuer, error)

func (*BulkStepRunQueuer) BulkQueueStepRuns

func (w *BulkStepRunQueuer) BulkQueueStepRuns(ctx context.Context, opts []BulkQueueStepRunOpts) ([]pgtype.UUID, error)

func (*BulkStepRunQueuer) Cleanup

func (w *BulkStepRunQueuer) Cleanup() error

type FlushResponse

type FlushResponse[U any] struct {
	Result U
	Err    error
}

type IngestBuf

type IngestBuf[T any, U any] struct {
	// contains filtered or unexported fields
}

func NewIngestBuffer

func NewIngestBuffer[T any, U any](opts IngestBufOpts[T, U]) *IngestBuf[T, U]

NewIngestBuffer creates a new buffer for any type T

func (*IngestBuf[T, U]) BuffItem

func (b *IngestBuf[T, U]) BuffItem(item T) (chan *FlushResponse[U], error)

func (*IngestBuf[T, U]) Start

func (b *IngestBuf[T, U]) Start() (func() error, error)

func (*IngestBuf[T, U]) StartDebugLoop

func (b *IngestBuf[T, U]) StartDebugLoop()

type IngestBufOpts

type IngestBufOpts[T any, U any] struct {
	Name               string                                            `validate:"required"`
	MaxCapacity        int                                               `validate:"required,gt=0"`
	FlushPeriod        time.Duration                                     `validate:"required,gt=0"`
	MaxDataSizeInQueue int                                               `validate:"required,gt=0"`
	OutputFunc         func(ctx context.Context, items []T) ([]U, error) `validate:"required"`
	SizeFunc           func(T) int                                       `validate:"required"`
	L                  *zerolog.Logger                                   `validate:"required"`
}

type SemaphoreReleaseOpts

type SemaphoreReleaseOpts struct {
	StepRunId pgtype.UUID
	TenantId  pgtype.UUID
}

type TenantBufManagerOpts

type TenantBufManagerOpts[T any, U any] struct {
	Name       string                                            `validate:"required"`
	OutputFunc func(ctx context.Context, items []T) ([]U, error) `validate:"required"`
	SizeFunc   func(T) int                                       `validate:"required"`
	L          *zerolog.Logger                                   `validate:"required"`
	V          validator.Validator                               `validate:"required"`

	FlushPeriod         *time.Duration
	FlushItemsThreshold int
}

type TenantBufferManager

type TenantBufferManager[T any, U any] struct {
	// contains filtered or unexported fields
}

func NewTenantBufManager

func NewTenantBufManager[T any, U any](opts TenantBufManagerOpts[T, U]) (*TenantBufferManager[T, U], error)

Create a new TenantBufferManager with generic types T for input and U for output

func (*TenantBufferManager[T, U]) BuffItem

func (t *TenantBufferManager[T, U]) BuffItem(tenantKey string, eventOps T) (chan *FlushResponse[U], error)

func (*TenantBufferManager[T, U]) Cleanup

func (t *TenantBufferManager[T, U]) Cleanup() error

cleanup all tenant buffers