kernel

Documentation

Index

• Variables
• type Clock
• type Component
  • func NewComponent(id string, hooks ComponentHooks) *Component
  • func (c *Component) Degrade(err error)
  • func (c *Component) Fault(err error)
  • func (c *Component) Health() Health
  • func (c *Component) Kill(ctx context.Context) error
  • func (c *Component) Start(ctx context.Context) error
  • func (c *Component) State() ComponentState
  • func (c *Component) Stop(ctx context.Context) error
• type ComponentHooks
• type ComponentState
• type EndpointHandler
• type Envelope
• type Health
• type LiveClock
  • func (LiveClock) After(d time.Duration) <-chan time.Time
  • func (LiveClock) Now() time.Time
• type MsgBus
  • func NewMsgBus(cfg MsgBusConfig) *MsgBus
  • func (b *MsgBus) Publish(ctx context.Context, topic string, message any) error
  • func (b *MsgBus) RegisterEndpoint(endpoint string, handler EndpointHandler)
  • func (b *MsgBus) Request(ctx context.Context, endpoint string, payload any) (Response, error)
  • func (b *MsgBus) Stats() MsgBusStats
  • func (b *MsgBus) Subscribe(topic string, buffer int) Subscription
• type MsgBusConfig
• type MsgBusStats
• type Request
• type Response
• type Subscription
• type TestClock
  • func NewTestClock(start time.Time) *TestClock
  • func (c *TestClock) Advance(d time.Duration)
  • func (c *TestClock) After(d time.Duration) <-chan time.Time
  • func (c *TestClock) Now() time.Time

Variables

var (
	ErrBackpressure     = errors.New("msgbus backpressure")
	ErrEndpointNotFound = errors.New("endpoint not found")
)

var ErrComponentFaulted = errors.New("component faulted")

Types

type Clock

type Clock interface {
	Now() time.Time
	After(time.Duration) <-chan time.Time
}

type Component

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

func NewComponent

func NewComponent(id string, hooks ComponentHooks) *Component

func (*Component) Degrade

func (c *Component) Degrade(err error)

func (*Component) Fault

func (c *Component) Fault(err error)

func (*Component) Health

func (c *Component) Health() Health

func (*Component) Kill

func (c *Component) Kill(ctx context.Context) error

func (*Component) Start

func (c *Component) Start(ctx context.Context) error

func (*Component) State

func (c *Component) State() ComponentState

func (*Component) Stop

func (c *Component) Stop(ctx context.Context) error

type ComponentHooks

type ComponentHooks struct {
	Start func(context.Context) error
	Stop  func(context.Context) error
	Kill  func(context.Context) error
}

type ComponentState

type ComponentState string

const (
	ComponentStateInitialized ComponentState = "initialized"
	ComponentStateStarting    ComponentState = "starting"
	ComponentStateRunning     ComponentState = "running"
	ComponentStateStopping    ComponentState = "stopping"
	ComponentStateStopped     ComponentState = "stopped"
	ComponentStateDegraded    ComponentState = "degraded"
	ComponentStateFaulted     ComponentState = "faulted"
)

type EndpointHandler

type EndpointHandler func(context.Context, Request) (Response, error)

type Envelope

type Envelope struct {
	Topic         string
	Message       any
	Timestamp     time.Time
	CorrelationID string
}

type Health

type Health struct {
	ID        string
	State     ComponentState
	LastError string
}

type LiveClock

type LiveClock struct{}

func (LiveClock) After

func (LiveClock) After(d time.Duration) <-chan time.Time

func (LiveClock) Now

func (LiveClock) Now() time.Time

type MsgBus

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

func NewMsgBus

func NewMsgBus(cfg MsgBusConfig) *MsgBus

func (*MsgBus) Publish

func (b *MsgBus) Publish(ctx context.Context, topic string, message any) error

func (*MsgBus) RegisterEndpoint

func (b *MsgBus) RegisterEndpoint(endpoint string, handler EndpointHandler)

func (*MsgBus) Request

func (b *MsgBus) Request(ctx context.Context, endpoint string, payload any) (Response, error)

func (*MsgBus) Stats

func (b *MsgBus) Stats() MsgBusStats

func (*MsgBus) Subscribe

func (b *MsgBus) Subscribe(topic string, buffer int) Subscription

type MsgBusConfig

type MsgBusConfig struct {
	Clock         Clock
	DefaultBuffer int
}

type MsgBusStats

type MsgBusStats struct {
	Published int64
	Dropped   int64
}

type Request

type Request struct {
	Endpoint      string
	Payload       any
	CorrelationID string
	Timestamp     time.Time
}

type Response

type Response struct {
	CorrelationID string
	Payload       any
}

type Subscription

type Subscription interface {
	C() <-chan Envelope
	Close() error
}

type TestClock

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

func NewTestClock

func NewTestClock(start time.Time) *TestClock

func (*TestClock) Advance

func (c *TestClock) Advance(d time.Duration)

func (*TestClock) After

func (c *TestClock) After(d time.Duration) <-chan time.Time

func (*TestClock) Now

func (c *TestClock) Now() time.Time