entities

Documentation

Index

• Variables
• func Collection(name string) string
• func ConsumerId() string
• func ConsumerIdFromTime(t time.Time) string
• func EventId() string
• func EventIdFromTime(t time.Time) string
• func Properties(entity any) []string
• func StreamId() string
• func StreamIdFromTime(t time.Time) string
• func TaskId() string
• func TaskIdFromTime(t time.Time) string
• type AttemptedError
  • func (err *AttemptedError) Scan(value interface{}) error
  • func (err AttemptedError) Value() (driver.Value, error)
• type ConsumerKind
  • func (kind ConsumerKind) String() string
• type ConsumerRegistry
• type Event
  • func NewEvent(subject string, body []byte) *Event
• type Metadata
  • func (m Metadata) Clone() Metadata
  • func (m Metadata) Merge(src Metadata)
  • func (m *Metadata) Scan(value interface{}) error
  • func (m Metadata) Value() (driver.Value, error)
• type StreamRegistry
• type Task
  • func TaskFromEvent(e *Event) *Task
• type TaskState
  • func (state TaskState) String() string

Variables

var (
	DefaultStreamName                string = "default"
	DefaultConsumerName              string = "default"
	DefaultConsumerAttemptMax        int16  = 16
	DefaultConsumerVisibilityTimeout int64  = 1000 * 60 * 15
)

Functions

func Collection

func Collection(name string) string

func ConsumerId

func ConsumerId() string

func ConsumerIdFromTime

func ConsumerIdFromTime(t time.Time) string

func EventId

func EventId() string

func EventIdFromTime

func EventIdFromTime(t time.Time) string

func Properties

func Properties(entity any) []string

func StreamId

func StreamId() string

func StreamIdFromTime

func StreamIdFromTime(t time.Time) string

func TaskId

func TaskId() string

func TaskIdFromTime

func TaskIdFromTime(t time.Time) string

Types

type AttemptedError

type AttemptedError struct {
	// At is the time at which the error occurred.
	At int64 `json:"at" validate:"required,gt=0"`
	// Error contains the stringified error of an error returned from a job or a
	// panic value in case of a panic.
	Error string `json:"error" validate:"required"`
	// Stack contains a stack trace from a job that panicked. The trace is
	// produced by invoking `debug.Stack()`.
	Stack string `json:"stack" validate:"required"`
}

func (*AttemptedError) Scan

func (err *AttemptedError) Scan(value interface{}) error

Scan implements the sql.Scanner interface to scan a value from the database into the Metadata struct

func (AttemptedError) Value

func (err AttemptedError) Value() (driver.Value, error)

Value implements the driver.Valuer interface to convert the Metadata struct to a value that can be stored in the database

type ConsumerKind

type ConsumerKind int16

const (
	KindPrimary ConsumerKind = 1
)

func (ConsumerKind) String

func (kind ConsumerKind) String() string

type ConsumerRegistry

type ConsumerRegistry struct {
	StreamId          string       `json:"stream_id" validate:"required"`
	StreamName        string       `json:"stream_name" validate:"required,is_collection_name"`
	Id                string       `json:"id" validate:"required"`
	Name              string       `json:"name" validate:"required,is_collection_name"`
	Kind              ConsumerKind `json:"kind" validate:"required,is_enum"`
	SubjectIncludes   []string     `json:"subject_includes" validate:"required,gt=0,dive,is_subject_filter"`
	SubjectExcludes   []string     `json:"subject_excludes" validate:"gte=0,dive,is_subject_filter"`
	Cursor            string       `json:"cursor"`
	AttemptMax        int16        `json:"attempt_max"`
	VisibilityTimeout int64        `json:"visibility_timeout" validate:"required,gt=1000"`
	CreatedAt         int64        `json:"created_at"`
	UpdatedAt         int64        `json:"updated_at"`
}

type Event

type Event struct {
	Id      string `json:"id" validate:"required"`
	Subject string `json:"subject" validate:"required,is_subject"`
	Body    []byte `json:"body" validate:"required"`

	// Metadata carry additional information about the event.
	Metadata Metadata `json:"metadata" validate:"required"`
	// CreatedAt is when the event record was created.
	CreatedAt int64 `json:"created_at" validate:"required,gt=0"`
}

func NewEvent

func NewEvent(subject string, body []byte) *Event

type Metadata

type Metadata map[string]interface{}

func (Metadata) Clone

func (m Metadata) Clone() Metadata

func (Metadata) Merge

func (m Metadata) Merge(src Metadata)

func (*Metadata) Scan

func (m *Metadata) Scan(value interface{}) error

Scan implements the sql.Scanner interface to scan a value from the database into the Metadata struct

func (Metadata) Value

func (m Metadata) Value() (driver.Value, error)

Value implements the driver.Valuer interface to convert the Metadata struct to a value that can be stored in the database

type StreamRegistry

type StreamRegistry struct {
	Id        string `json:"id" validate:"required"`
	Name      string `json:"name" validate:"required,is_collection_name"`
	CreatedAt int64  `json:"created_at"`
	UpdatedAt int64  `json:"updated_at"`
}

type Task

type Task struct {
	EventId string `json:"event_id" validate:"required"`
	Subject string `json:"subject" validate:"required,is_subject"`

	// State is the state of task like `available` or `completed`.
	State TaskState `json:"state" validate:"is_enum"`
	// ScheduledAt is when the task is scheduled to become available to be
	// worked. Tasks default to running immediately, but may be scheduled
	// for the future when they're inserted. They may also be scheduled for
	// later because they were snoozed or because they errored and have
	// additional retry attempts remaining.
	ScheduleAt int64 `json:"schedule_at" validate:"required,gt=0"`

	// AttemptCount is the attempt number of the task. Tasks are inserted at 0, the
	// number is incremented to 1 the first time work its worked, and may
	// increment further if it's either snoozed or errors.
	AttemptCount int16 `json:"attempt_count" validate:"gte=0"`
	// AttemptedAt is the time that the task was last worked. Starts out as NOW()
	// on a new insert.
	AttemptedAt int64 `json:"attempted_at" validate:"gte=0"`
	// AttemptedBy is the set of client IDs that have worked this task.
	AttemptedError []AttemptedError `json:"attempted_error"`
	// FinalizedAt is the time at which the task was "finalized", meaning it was
	// either completed successfully or errored for the last time such that
	// it'll no longer be retried.
	FinalizedAt int64 `json:"finalized_at" validate:"gte=0"`

	// Metadata carry additional information about the task.
	Metadata Metadata `json:"metadata" validate:"required"`
	// CreatedAt is when the task record was created.
	CreatedAt int64 `json:"created_at" validate:"required,gt=0"`
	// CreatedAt is when the task record was updated.
	UpdatedAt int64 `json:"updated_at" validate:"gte=0"`
}

func TaskFromEvent

func TaskFromEvent(e *Event) *Task

type TaskState

type TaskState int16

const (
	// StateDiscarded is the state for tasks that have errored enough times
	// that they're no longer eligible to be retried. Manual user invention
	// is required for them to be tried again.
	StateDiscarded TaskState = -102
	// StateCancelled is the state for tasks that have been manually cancelled
	// by user request.
	StateCancelled TaskState = -101
	// StatePending is a state for tasks to be parked while waiting for some
	// external action before they can be worked. Tasks in pending will never be
	// worked or deleted unless moved out of this state by the user.
	StatePending TaskState = 0
	// StateAvailable is the state for tasks that are immediately eligible to
	// be worked.
	StateAvailable TaskState = 1
	// StateRunning is the state for tasks tasks which are actively running.
	StateRunning TaskState = 2
	// Completed is the state for tasks that have successfully run to
	// completion.
	StateCompleted TaskState = 101
	// StateRetryable is the state for tasks that have errored, but will be
	// retried.
	StateRetryable TaskState = 102
)

func (TaskState) String

func (state TaskState) String() string