health

Documentation

Index

• Variables
• type Core
  • func NewCore(opts CoreOptions, consumer event.StreamConsumer, reducer Reducer) *Core
  • func (c *Core) GetPastEvents(manifestID string, from, to *time.Time) ([]data.Event, error)
  • func (c *Core) GetStatus(manifestID string) (*data.HealthStatus, error)
  • func (c *Core) HandleMessage(msg event.StreamMessage)
  • func (c *Core) IsHealthy() bool
  • func (c *Core) Start(ctx context.Context) error
  • func (c *Core) SubscribeEvents(ctx context.Context, manifestID string, lastEvtID *uuid.UUID, from *time.Time) ([]data.Event, <-chan data.Event, error)
• type CoreOptions
• type Record
  • func NewRecord(id string, conditionTypes []data.ConditionType) *Record
  • func (r *Record) SubscribeLocked(ctx context.Context, subs chan data.Event) chan data.Event
• type RecordStorage
  • func (s *RecordStorage) Get(id string) (*Record, bool)
  • func (s *RecordStorage) GetOrCreate(id string, conditions []data.ConditionType) *Record
  • func (s *RecordStorage) StartCleanupLoop(ctx context.Context, ttl time.Duration)
• type Reducer
• type ReducerFunc
  • func (f ReducerFunc) Bindings() []event.BindingArgs
  • func (f ReducerFunc) Conditions() []data.ConditionType
  • func (f ReducerFunc) Reduce(current *data.HealthStatus, state interface{}, evt data.Event) (*data.HealthStatus, interface{})
• type StreamingOptions

Variables

var (
	ErrStreamNotFound = errors.New("stream not found")
	ErrEventNotFound  = errors.New("event not found")
)

Types

type Core

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

func NewCore

func NewCore(opts CoreOptions, consumer event.StreamConsumer, reducer Reducer) *Core

func (*Core) GetPastEvents

func (c *Core) GetPastEvents(manifestID string, from, to *time.Time) ([]data.Event, error)

func (*Core) GetStatus

func (c *Core) GetStatus(manifestID string) (*data.HealthStatus, error)

func (*Core) HandleMessage

func (c *Core) HandleMessage(msg event.StreamMessage)

func (*Core) IsHealthy

func (c *Core) IsHealthy() bool

func (*Core) Start

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

func (*Core) SubscribeEvents

func (c *Core) SubscribeEvents(ctx context.Context, manifestID string, lastEvtID *uuid.UUID, from *time.Time) ([]data.Event, <-chan data.Event, error)

type CoreOptions

type CoreOptions struct {
	Streaming        StreamingOptions
	StartTimeOffset  time.Duration
	MemoryRecordsTtl time.Duration
}

type Record

type Record struct {
	ID         string
	Conditions []data.ConditionType

	sync.RWMutex

	PastEvents []data.Event
	EventsByID map[uuid.UUID]data.Event
	EventSubs  []chan<- data.Event

	ReducerState interface{}
	LastStatus   *data.HealthStatus
	// contains filtered or unexported fields
}

func NewRecord

func NewRecord(id string, conditionTypes []data.ConditionType) *Record

func (*Record) SubscribeLocked

func (r *Record) SubscribeLocked(ctx context.Context, subs chan data.Event) chan data.Event

type RecordStorage

type RecordStorage struct {
	SizeGauge prometheus.Gauge
	// contains filtered or unexported fields
}

func (*RecordStorage) Get

func (s *RecordStorage) Get(id string) (*Record, bool)

func (*RecordStorage) GetOrCreate

func (s *RecordStorage) GetOrCreate(id string, conditions []data.ConditionType) *Record

func (*RecordStorage) StartCleanupLoop

func (s *RecordStorage) StartCleanupLoop(ctx context.Context, ttl time.Duration)

type Reducer

type Reducer interface {
	Bindings() []event.BindingArgs
	Conditions() []data.ConditionType
	Reduce(current *data.HealthStatus, state interface{}, evt data.Event) (*data.HealthStatus, interface{})
}

type ReducerFunc

type ReducerFunc func(*data.HealthStatus, interface{}, data.Event) (*data.HealthStatus, interface{})

func (ReducerFunc) Bindings

func (f ReducerFunc) Bindings() []event.BindingArgs

func (ReducerFunc) Conditions

func (f ReducerFunc) Conditions() []data.ConditionType

func (ReducerFunc) Reduce

func (f ReducerFunc) Reduce(current *data.HealthStatus, state interface{}, evt data.Event) (*data.HealthStatus, interface{})

type StreamingOptions

type StreamingOptions struct {
	Stream, ConsumerName string

	event.RawStreamOptions
}

Purposedly made of built-in types only to bind directly to cli flags.