core

Documentation

Index

• Variables
• type AbstractSystem
• type Broker
• type BrokerEvent
• type Event
  • func NewEvent(topic string, data []byte) Event
• type Option
• type Resource
• type ResourceDB
• type System
  • func NewSystem(broker Broker, resources ResourceDB, tasks TaskDB, logger *slog.Logger) *System
  • func (system System) Events() Broker
  • func (system System) Resources() ResourceDB
  • func (system System) Spawn(ctx context.Context, newTask Task) (initializedTask Task, err error)
  • func (system System) Task(ctx context.Context, id string) (Task, error)
  • func (system System) Tasks() TaskDB
• type Task
  • func New(options ...Option) Task
  • func (task Task) AsDoc() storage.Document
  • func (task Task) ContextID() string
• type TaskCodec
  • func (c TaskCodec) Decode(rec map[string]any) (Task, error)
  • func (c TaskCodec) Encode(task Task) (map[string]any, error)
• type TaskDB

Variables

var OnTask struct {
	Created   string
	Ready     string
	Received  string
	Finished  string
	Cancelled string
}

var ResourceStatuses struct {
	DoesNotExist string
	Allocated    string
	Ready        string
}

var TaskStatuses struct {
	Initialized string
	Ready       string
	Cancelled   string
	Received    string
	Finished    string
}

var TaskSubStatuses struct {
	OK     string
	Failed string
}

Types

type AbstractSystem

type AbstractSystem interface {
	Tasks() TaskDB
	Resources() ResourceDB
	Events() Broker

	Spawn(ctx context.Context, t Task) (Task, error)
	Task(ctx context.Context, id string) (Task, error)
}

type Broker

type Broker interface {
	Send(ctx context.Context, event Event) error
	Consume(ctx context.Context, events []string, consumerSettings broker.ConsumerSettings) (<-chan BrokerEvent, error)
}

type BrokerEvent

type BrokerEvent = broker.Message[Event]

type Event

type Event struct {
	ID        string
	Data      []byte
	Topic     string
	Timestamp int64
}

func NewEvent

func NewEvent(topic string, data []byte) Event

type Option

type Option func(t *Task)

type Resource

type Resource struct {
	Data   []byte
	ID     string
	Status string
}

type ResourceDB

type ResourceDB interface {
	Load(ctx context.Context, ids ...string) ([]Resource, error)
	Alloc(ctx context.Context, amount int) (ids []string, err error)
	Init(ctx context.Context, resources []Resource) error
	Dealloc(ctx context.Context, ids []string) error
}

type System

type System struct {
	// contains filtered or unexported fields
}

func NewSystem

func NewSystem(broker Broker, resources ResourceDB, tasks TaskDB, logger *slog.Logger) *System

func (System) Events

func (system System) Events() Broker

func (System) Resources

func (system System) Resources() ResourceDB

func (System) Spawn

func (system System) Spawn(ctx context.Context, newTask Task) (initializedTask Task, err error)

func (System) Task

func (system System) Task(ctx context.Context, id string) (Task, error)

func (System) Tasks

func (system System) Tasks() TaskDB

type Task

type Task struct {
	Inputs  []string
	Outputs []string
	ID      string
	Info    map[string]any
}

func New

func New(options ...Option) Task

func (Task) AsDoc

func (task Task) AsDoc() storage.Document

func (Task) ContextID

func (task Task) ContextID() string

type TaskCodec

type TaskCodec struct{}

func (TaskCodec) Decode

func (c TaskCodec) Decode(rec map[string]any) (Task, error)

func (TaskCodec) Encode

func (c TaskCodec) Encode(task Task) (map[string]any, error)

type TaskDB

type TaskDB interface {
	storage.Storage

	Insert(ctx context.Context, tasks []Task) error
	Delete(ctx context.Context, ids []string) error
	ByIDs(ctx context.Context, ids []string) ([]Task, error)
	UpdateByIDs(ctx context.Context, ids []string, properties map[string]any) error
	GetWithProperties(ctx context.Context, propertiesToValues map[string][]any) ([]Task, error)
	UpdateWithProperties(ctx context.Context, propertiesToValues map[string][]any, newProperties map[string]any) (updatedDocs int, err error)
}