storage

Documentation

Index

• type IDGenerator
• type OutboxEntry
• type Storage
  • func New(db *gorm.DB) *Storage
  • func (s *Storage) AutoMigrate() error
  • func (s *Storage) ClaimEntries(ctx context.Context, processorID string, claimDeadline time.Time) error
  • func (s *Storage) DeleteEntries(ctx context.Context, entryIDs ...string) error
  • func (s *Storage) GetClaimedEntries(ctx context.Context, processorID string, batchSize int) ([]outbox.ClaimedEntry, error)
  • func (s *Storage) Publish(ctx context.Context, txn interface{}, messages ...outbox.Message) error
• type UUIDGenerator
  • func (u *UUIDGenerator) GenerateID(clockwork.Clock, outbox.Message) string

Types

type IDGenerator

type IDGenerator interface {
	// GenerateID is called to generate a unique outbox entry ID
	GenerateID(clock clockwork.Clock, message outbox.Message) string
}

IDGenerator allows for customising how outbox entry IDs are generated

type OutboxEntry

type OutboxEntry struct {
	// ID uniquely identifies this outbox entry
	ID string
	// Key that will be passed to the outbox's publisher
	Key []byte
	// Payload that will be passed to the outbox's publisher
	Payload []byte
	// ProcessorID identifies the processor that currently has a claim on this outbox entry
	ProcessorID string
	// ProcessingDeadline is when any outstanding claim to this outbox entry expires
	ProcessingDeadline *time.Time
}

OutboxEntry is the internal representation of an outbox entry

type Storage

type Storage struct {
	IDGenerator IDGenerator
	Clock       clockwork.Clock

	GetTxn func(ctx context.Context) *gorm.DB
	// contains filtered or unexported fields
}

Storage implements outboxen's outbox.ProcessorStorage interface, allowing the outbox.Outbox to process outbox entries and reliably publish them to some external messaging system

func New

func New(db *gorm.DB) *Storage

New constructs a new Storage instance with default settings

func (*Storage) AutoMigrate

func (s *Storage) AutoMigrate() error

AutoMigrate runs GORM's auto migrator on the outbox table

func (*Storage) ClaimEntries

func (s *Storage) ClaimEntries(ctx context.Context, processorID string, claimDeadline time.Time) error

ClaimEntries atomically claims all claimable entries for the specified processor Will claim any unclaimed entries (empty processor ID or lack of processing deadline)

entries previously claimed by this processor (same processor ID)
entries whose claim is now out of date

func (*Storage) DeleteEntries

func (s *Storage) DeleteEntries(ctx context.Context, entryIDs ...string) error

DeleteEntries deletes all entries specified (by their ID)

func (*Storage) GetClaimedEntries

func (s *Storage) GetClaimedEntries(ctx context.Context, processorID string, batchSize int) ([]outbox.ClaimedEntry, error)

GetClaimedEntries retrieves entries claimed by this processor, up to the specified batch size

func (*Storage) Publish

func (s *Storage) Publish(ctx context.Context, txn interface{}, messages ...outbox.Message) error

Publish will write messages to the outbox for later publishing, using the transaction stored in the context

type UUIDGenerator

type UUIDGenerator struct{}

UUIDGenerator generates IDs by producing random UUIDs

func (*UUIDGenerator) GenerateID

func (u *UUIDGenerator) GenerateID(clockwork.Clock, outbox.Message) string

GenerateID implements the IDGenerator interface