87 lines
2.3 KiB
Go
87 lines
2.3 KiB
Go
// Package probe maps MediaMTX runtime path health into the Sense actual state.
|
|
package probe
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"time"
|
|
|
|
"yovision/sense/internal/device"
|
|
)
|
|
|
|
const defaultBatchSize = 128
|
|
|
|
type Repository interface {
|
|
ListEnabledVideoDevices(ctx context.Context, limit int) ([]device.Device, error)
|
|
UpdateActualState(ctx context.Context, id string, state device.ActualState, now time.Time) error
|
|
RequestReconcile(ctx context.Context, id string, now time.Time) error
|
|
}
|
|
|
|
type RuntimePaths interface {
|
|
PathReady(ctx context.Context, name string) (bool, error)
|
|
}
|
|
|
|
type Checker struct {
|
|
repository Repository
|
|
media RuntimePaths
|
|
now func() time.Time
|
|
batchSize int
|
|
}
|
|
|
|
func New(repository Repository, media RuntimePaths) *Checker {
|
|
return &Checker{repository: repository, media: media, now: time.Now, batchSize: defaultBatchSize}
|
|
}
|
|
|
|
func (c *Checker) RunOnce(ctx context.Context) error {
|
|
devices, err := c.repository.ListEnabledVideoDevices(ctx, c.batchSize)
|
|
if err != nil {
|
|
return fmt.Errorf("list probe candidates: %w", err)
|
|
}
|
|
var runErrors []error
|
|
for _, value := range devices {
|
|
if err := ctx.Err(); err != nil {
|
|
return err
|
|
}
|
|
ready, probeErr := c.media.PathReady(ctx, value.PathName)
|
|
if ctx.Err() != nil {
|
|
return ctx.Err()
|
|
}
|
|
state := device.ActualOffline
|
|
if probeErr == nil && ready {
|
|
state = device.ActualOnline
|
|
}
|
|
now := c.now().UTC()
|
|
if probeErr != nil {
|
|
if requestErr := c.repository.RequestReconcile(ctx, value.ID, now); requestErr != nil {
|
|
runErrors = append(runErrors, fmt.Errorf("request device %s reconciliation: %w", value.ID, requestErr))
|
|
}
|
|
}
|
|
if updateErr := c.repository.UpdateActualState(ctx, value.ID, state, now); updateErr != nil {
|
|
runErrors = append(runErrors, fmt.Errorf("update device %s health: %w", value.ID, updateErr))
|
|
}
|
|
if probeErr != nil {
|
|
runErrors = append(runErrors, fmt.Errorf("probe device %s: %w", value.ID, probeErr))
|
|
}
|
|
}
|
|
return errors.Join(runErrors...)
|
|
}
|
|
|
|
func (c *Checker) Run(ctx context.Context, interval time.Duration, report func(error)) {
|
|
if err := c.RunOnce(ctx); err != nil && ctx.Err() == nil && report != nil {
|
|
report(err)
|
|
}
|
|
ticker := time.NewTicker(interval)
|
|
defer ticker.Stop()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-ticker.C:
|
|
if err := c.RunOnce(ctx); err != nil && ctx.Err() == nil && report != nil {
|
|
report(err)
|
|
}
|
|
}
|
|
}
|
|
}
|