repository

Documentation

Index

• Variables
• type MessageEnvelope
  • func (msgEnv *MessageEnvelope) String() string
• type Repository
  • func NewPostgresRepository(db *sql.DB, log logrus.FieldLogger) Repository

Variables

var (
	ErrMessageNoID       = errors.New("Message cannot be written without a new UUID")
	ErrNegativeBatchSize = errors.New("Batch size cannot be negative")
)

var (
	ErrInvalidSubscriberID       = errors.New("Subscriber ID cannot be blank")
	ErrInvalidStreamName         = errors.New("Stream Name cannot be blank")
	ErrBlankCategory             = errors.New("Category cannot be blank")
	ErrInvalidCategory           = errors.New("Category cannot contain a hyphen")
	ErrInvalidSubscriberPosition = errors.New("Subscriber position must be greater than or equal to -1")
	ErrNilMessage                = errors.New("Message cannot be nil")
	ErrInvalidPosition           = errors.New("position must be greater than equal to -1")
)

Errors

Types

type MessageEnvelope

type MessageEnvelope struct {
	ID             uuid.UUID `db:"id"`
	StreamName     string    `db:"stream_name"`
	StreamCategory string    `db:"stream_category"`
	MessageType    string    `db:"type"`
	Version        int64     `db:"position"`
	GlobalPosition int64     `db:"global_position"`
	Data           []byte    `db:"data"`
	Metadata       []byte    `db:"metadata"`
	Time           time.Time `db:"time"`
}

Actual values that come out of the database

func (*MessageEnvelope) String

func (msgEnv *MessageEnvelope) String() string

type Repository

type Repository interface {
	// writes
	WriteMessage(ctx context.Context, message *MessageEnvelope) error
	WriteMessageWithExpectedPosition(ctx context.Context, message *MessageEnvelope, position int64) error
	// reads from stream
	GetAllMessagesInStream(ctx context.Context, streamName string, batchSize int) ([]*MessageEnvelope, error)
	GetAllMessagesInStreamSince(ctx context.Context, streamName string, globalPosition int64, batchSize int) ([]*MessageEnvelope, error)
	GetLastMessageInStream(ctx context.Context, streamName string) (*MessageEnvelope, error)
	// reads from category
	GetAllMessagesInCategory(ctx context.Context, category string, batchSize int) ([]*MessageEnvelope, error)
	GetAllMessagesInCategorySince(ctx context.Context, category string, globalPosition int64, batchSize int) ([]*MessageEnvelope, error)
}

Repository the storage implementation for messagestore

func NewPostgresRepository

func NewPostgresRepository(db *sql.DB, log logrus.FieldLogger) Repository

NewPostgresRepository creates a new in memory implementation for the messagestore reop