relay

package
v0.11.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 17, 2026 License: AGPL-3.0 Imports: 42 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	DefaultSampleRate           = 48000
	DefaultChannels             = 2
	DefaultFrameDurationMs      = 100 // 10 fps
	DefaultMaxContinuousSilence = 5 * time.Second
)

Default values for SilenceAACGenerator.

Variables

View Source 
var DriftReconnectError = &driftReconnectError{}

DriftReconnectError is returned via cbErr when sustained A/V PTS drift triggers a target reconnect.

Functions

This section is empty.

Types

type BufferAwareSilenceEmitter 

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

BufferAwareSilenceEmitter wraps a SilenceAACGenerator and gates emission via ShouldEmit() based on audio buffer health (drop detection) and a 5 fps hard cap. The goal is to keep the audio bus from being flooded with silent frames when the downstream is saturated.

Use NotifyDrop() to signal that the StreamHub audio consumer dropped a frame (called from the relay engine's OnAudioDrop callback). After a drop, the emitter suppresses all frame emission for dropRecoveryPeriod (3 seconds), then resumes normally — subject to the 5 fps hard cap.

The 5 fps cap corresponds to ≤10% of the 50-frame audio consumer buffer per second, which is well below the 10 fps the generator produces.

func NewBufferAwareSilenceEmitter 

func NewBufferAwareSilenceEmitter(generator *SilenceAACGenerator) *BufferAwareSilenceEmitter

NewBufferAwareSilenceEmitter creates a new emitter wrapping the provided generator. The caller is responsible for lifecycle (Start/Stop).

func (*BufferAwareSilenceEmitter) AudioConfig 

func (e *BufferAwareSilenceEmitter) AudioConfig() []byte

AudioConfig returns the AudioSpecificConfig bytes as configured in the underlying SilenceAACGenerator (delegates to generator.Config()).

func (*BufferAwareSilenceEmitter) Emitted 

func (e *BufferAwareSilenceEmitter) Emitted() int64

Emitted returns the total number of silent frames forwarded since Start.

func (*BufferAwareSilenceEmitter) NotifyDrop 

func (e *BufferAwareSilenceEmitter) NotifyDrop()

NotifyDrop is called by the relay engine when the StreamHub audio consumer reports a dropped frame (channel full). After this, ShouldEmit returns false for dropRecoveryPeriod, then resumes normally.

Thread-safe.

func (*BufferAwareSilenceEmitter) ShouldEmit 

func (e *BufferAwareSilenceEmitter) ShouldEmit() bool

ShouldEmit reports whether a silent frame should be emitted right now.

It returns false when:

  • A drop was detected within the last dropRecoveryPeriod (3 s).
  • The per-second hard cap (maxEmitsPerSecond = 5) has been reached.

Thread-safe.

func (*BufferAwareSilenceEmitter) Start 

func (e *BufferAwareSilenceEmitter) Start(ctx context.Context) <-chan []byte

Start begins the silence generator and returns a channel of silent AAC-LC frames filtered by ShouldEmit(). Frames that pass the gate are forwarded; suppressed frames are counted in totalSuppressed.

The returned channel is closed when the underlying generator stops (context cancelled, Stop() called, or silence timeout expires).

func (*BufferAwareSilenceEmitter) Stop 

func (e *BufferAwareSilenceEmitter) Stop()

Stop delegates to the underlying SilenceAACGenerator.Stop(), which cancels the context and closes the output channel.

func (*BufferAwareSilenceEmitter) Suppressed 

func (e *BufferAwareSilenceEmitter) Suppressed() int64

Suppressed returns the total number of silent frames dropped by the limiter (either due to drop backoff or the per-second hard cap).

type CameraHubProvider 

type CameraHubProvider func(cameraID string) *model.StreamHub

CameraHubProvider returns the StreamHub for a camera id (or nil). Backed by CameraManager.GetHub in production.

type CodecInfoProvider 

type CodecInfoProvider func(cameraID string) model.CodecInfo

CodecInfoProvider returns the source camera's current codec parameters (video SPS/PPS/VPS + audio codec info). The Manager adapts this per-target into the zero-arg codecInfoProvider that PushTarget expects. Backed by CameraManager.GetCodecInfo in production.

type DriftMonitor 

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

DriftMonitor tracks A/V PTS drift between video and audio callbacks and signals a reconnect when the drift exceeds 1s for more than 5 consecutive seconds. All hot-path operations are lock-free atomic reads/writes — the mutex is only taken on sustained-drift state transitions (at most once per second under drift conditions).

func NewDriftMonitor 

func NewDriftMonitor() *DriftMonitor

NewDriftMonitor creates a DriftMonitor with a 5-second sustained-window.

func (*DriftMonitor) DriftMs 

func (m *DriftMonitor) DriftMs() float64

DriftMs returns the absolute A/V PTS drift in milliseconds. Returns 0 when no drift data is available (at least one of video or audio has not been recorded yet, e.g., a no-audio camera).

func (*DriftMonitor) HighWaterMs 

func (m *DriftMonitor) HighWaterMs() float64

HighWaterMs returns the maximum absolute drift ever recorded, in milliseconds. Useful for metrics and debug logging.

func (*DriftMonitor) RecordAudio 

func (m *DriftMonitor) RecordAudio(pts int64)

RecordAudio stores the latest audio PTS (90kHz clock) and re-evaluates the sustained-drift state. Safe to call from any goroutine.

func (*DriftMonitor) RecordVideo 

func (m *DriftMonitor) RecordVideo(pts int64)

RecordVideo stores the latest video PTS (90kHz clock) and re-evaluates the sustained-drift state. Safe to call from any goroutine — the atomic store is O(1) and the mutex is only held for the threshold comparison.

func (*DriftMonitor) Reset 

func (m *DriftMonitor) Reset()

Reset clears all tracked state. Must be called after a reconnect so that the new connection starts with a clean slate.

func (*DriftMonitor) ShouldReconnect 

func (m *DriftMonitor) ShouldReconnect() bool

ShouldReconnect returns true when the A/V drift has exceeded the 1s threshold continuously for more than the sustained duration (default 5s). It is safe to call from any goroutine.

func (*DriftMonitor) StartLogging 

func (m *DriftMonitor) StartLogging(ctx context.Context)

StartLogging spawns a background goroutine that logs the current A/V drift every 60 seconds. The goroutine exits cleanly when ctx is cancelled. Calling StartLogging multiple times creates multiple goroutines — the caller should only call it once per monitor.

type Manager 

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

Manager owns the lifecycle of all push-out targets across all cameras. Each (cameraID, targetID) pair maps to at most one running *PushTarget. Manager is nil-safe at the call sites: when no relays are configured, main.go passes a no-op manager so camera Add/Update/Remove don't need nil checks.

func NewManager 

func NewManager(hubProvider CameraHubProvider, spsProvider SPSCameraProvider) *Manager

NewManager constructs a Manager. hubProvider and spsProvider are required. NewManager constructs a Manager. hubProvider and spsProvider are required.

func (*Manager) CameraStatus 

func (m *Manager) CameraStatus(cameraID string) []TargetStatus

CameraStatus returns the runtime status of every target for a camera.

func (*Manager) CameraStatusJSON 

func (m *Manager) CameraStatusJSON(cameraID string) []any

CameraStatusJSON returns the camera's target statuses as []any so the camera manager (which can't import relay) can pass them to the JSON API.

func (*Manager) FFmpegAvailable added in v0.8.1 

func (m *Manager) FFmpegAvailable() bool

FFmpegAvailable returns whether the FFmpeg binary is available for relay use.

func (*Manager) GetPreset 

func (m *Manager) GetPreset(name string) (Preset, bool)

GetPreset returns a single preset by name. Returns false when the name is not found or no registry is configured.

func (*Manager) ListAllPresets 

func (m *Manager) ListAllPresets() []Preset

ListAllPresets returns all registered presets sorted by name. Returns nil when no preset registry is configured.

func (*Manager) RemoveCamera 

func (m *Manager) RemoveCamera(cameraID string)

RemoveCamera stops all targets for a camera (called on camera delete).

func (*Manager) SetCameraTargets 

func (m *Manager) SetCameraTargets(cameraID string, cfgs []config.PushTargetConfig)

SetCameraTargets reconciles the running targets for one camera against the given config list: stops removed/changed targets, starts new/changed ones. Idempotent — safe to call with the same config on every camera update. Accepts config.PushTargetConfig (the persisted type) and adapts internally.

func (*Manager) SetCodecInfoProvider 

func (m *Manager) SetCodecInfoProvider(p CodecInfoProvider)

SetCodecInfoProvider wires an optional CodecInfoProvider for use by audio-aware push targets. Should be set before Start.

func (*Manager) SetFFmpegPath 

func (m *Manager) SetFFmpegPath(path string)

SetFFmpegPath sets an explicit FFmpeg binary path for the transcoder. If empty, the PushTarget will probe via exec.LookPath at runtime.

func (*Manager) SetHardwareCap 

func (m *Manager) SetHardwareCap(hwCap *transcoding.HardwareCapabilities)

SetHardwareCap wires HardwareCapabilities for transcoder encoder selection. Should be set before Start.

func (*Manager) SetPresetRegistry 

func (m *Manager) SetPresetRegistry(r *PresetRegistry)

SetPresetRegistry wires an optional PresetRegistry for transcode resolution. Should be set before Start (targets created after this call will use it).

func (*Manager) SetStreamURLProvider added in v0.8.1 

func (m *Manager) SetStreamURLProvider(p StreamURLProvider)

SetStreamURLProvider wires a function that resolves a camera's stream URL (e.g. rtsp://...) for FFmpeg relay mode.

func (*Manager) Start 

func (m *Manager) Start(ctx context.Context)

Start sets the root context used for all target goroutines.

func (*Manager) Stop 

func (m *Manager) Stop()

Stop cancels every running target and waits for them to exit.

type Preset 

type Preset struct {
	Name               string `json:"name"`
	Description        string `json:"description"`
	URLHint            string `json:"url_hint"`
	GopSeconds         int    `json:"gop_seconds"`
	VideoBitrateKbps   int    `json:"video_bitrate_kbps"`
	AudioBitrateKbps   int    `json:"audio_bitrate_kbps"`
	Resolution         string `json:"resolution"`
	Framerate          int    `json:"framerate"`
	Profile            string `json:"profile"`
	Bframes            int    `json:"bframes"`
	AudioCodecRequired string `json:"audio_codec_required"`
}

Preset defines encoding parameters for a live streaming platform. Fields are populated either from YAML or from built-in defaults.

type PresetRegistry 

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

PresetRegistry loads platform presets from YAML and provides 5 built-in defaults. On Load failure the registry falls back to the built-in set rather than returning an error.

func NewPresetRegistry 

func NewPresetRegistry() *PresetRegistry

NewPresetRegistry creates a registry initialised with the 5 built-in presets.

func (*PresetRegistry) Get 

func (r *PresetRegistry) Get(name string) (Preset, bool)

Get returns the preset with the given name and a boolean indicating whether it exists.

func (*PresetRegistry) List 

func (r *PresetRegistry) List() []string

List returns all preset names in the registry (order is not guaranteed).

func (*PresetRegistry) Load 

func (r *PresetRegistry) Load(path string) error

Load reads platform presets from a YAML file. On any failure (read, parse, validation) it logs a warning and falls back to built-in defaults.

func (*PresetRegistry) Resolve 

func (r *PresetRegistry) Resolve(cfg PushTargetConfig) ResolvedPreset

Resolve merges a PushTargetConfig with the matching platform preset to produce a fully-populated ResolvedPreset. Merge priority: 

per-target override > preset value > generic default

If the target's Platform field is empty or does not match any known preset, the "generic" preset is used as the base.

type PushTarget 

type PushTarget struct {
	CameraID string
	Config   PushTargetConfig
	// contains filtered or unexported fields
}

PushTarget is one push-out destination: it subscribes to a camera's StreamHub and forwards each access unit to the target (RTMP or RTSP) over a dedicated connection. Each target runs in its own goroutine with independent reconnect.

func NewPushTarget 

func NewPushTarget(cameraID string, cfg PushTargetConfig, hub *model.StreamHub, sps SPSProvider) *PushTarget

NewPushTarget constructs an idle target. It does not connect until Run.

func (*PushTarget) Run 

func (t *PushTarget) Run(ctx context.Context)

Run starts the target and blocks until ctx is canceled or the hub is gone. It owns the reconnect loop. Safe to call via `go t.Run(ctx)`.

func (*PushTarget) SetCodecInfoProvider 

func (t *PushTarget) SetCodecInfoProvider(p func() model.CodecInfo)

SetCodecInfoProvider wires an optional codec info provider for audio-aware targets. Should be set before Run.

func (*PushTarget) SetFFmpegPath 

func (t *PushTarget) SetFFmpegPath(path string)

SetFFmpegPath sets an explicit FFmpeg binary path for the transcoder. When empty, the transcoder auto-detects from HardwareCap or PATH.

func (*PushTarget) SetHardwareCap 

func (t *PushTarget) SetHardwareCap(hwCap *transcoding.HardwareCapabilities)

SetHardwareCap wires hardware capabilities for transcoder encoder selection. Should be set before Run if transcode may be used.

func (*PushTarget) SetPresetRegistry 

func (t *PushTarget) SetPresetRegistry(r *PresetRegistry)

SetPresetRegistry wires the preset registry for transcode path resolution. Should be set before Run if transcode may be used.

func (*PushTarget) SetStreamURLProvider added in v0.8.1 

func (t *PushTarget) SetStreamURLProvider(p StreamURLProvider)

SetStreamURLProvider wires a function that resolves a camera's stream URL (e.g. rtsp://...) for FFmpeg relay mode.

func (*PushTarget) Status 

func (t *PushTarget) Status() TargetStatus

Status returns a snapshot of the target's runtime status for the API/UI.

type PushTargetConfig 

type PushTargetConfig struct {
	ID                  string // stable id within the camera
	Name                string
	Protocol            string // "rtmp" or "rtsp"
	URL                 string // rtmp://host[:port]/app/key  |  rtsp://host[:port]/path
	Enabled             bool
	Platform            string                // preset: bilibili/douyin/youtube/kuaishou/generic/empty
	TranscodePolicy     string                // auto/force_sw/off
	VideoPresetOverride *VideoPresetOverrides // optional override
	UseFFmpeg           bool                  // if true, use FFmpeg subprocess for relay (compatibility mode)
	SourceURL           string                // optional: override auto-resolved source URL for FFmpeg relay
}

PushTargetConfig is the user-facing configuration for one push-out target. Mirrors config.PushTargetConfig but kept local to avoid a config<->relay import cycle (the manager maps between them).

type RelayStatus 

type RelayStatus string

RelayStatus is the lifecycle state of a single push-out target. It is a distinct type from model.RecorderStatus on purpose: a target reporting "streaming" must not be confused with a camera "recording" to disk, and the camera status UI/health system keys off RecorderStatus. 

const (
	// StatusIdle means the target exists but is disabled (not running).
	StatusIdle RelayStatus = "idle"
	// StatusConnecting means the target is attempting to establish a connection.
	StatusConnecting RelayStatus = "connecting"
	// StatusStreaming means frames are actively being pushed to the target.
	StatusStreaming RelayStatus = "streaming"
	// StatusReconnecting means the connection dropped and a retry is scheduled.
	StatusReconnecting RelayStatus = "reconnecting"
	// StatusError means the target is in a persistent error state (e.g. the
	// source camera codec is incompatible with the target protocol).
	StatusError RelayStatus = "error"
)

type ResolvedPreset 

type ResolvedPreset struct {
	Name             string
	GopSeconds       int
	VideoBitrateKbps int
	AudioBitrateKbps int
	Resolution       string
	Framerate        int
	Profile          string
	Bframes          int
}

ResolvedPreset is the fully-resolved set of encoding parameters after merging a platform preset with per-target overrides.

type SPSCameraProvider 

type SPSCameraProvider func(cameraID string) (sps, pps []byte, isH264 bool)

SPSCameraProvider returns the source camera's current SPS/PPS + H.264 flag, looked up by camera id. The Manager adapts this per-target into the zero-arg SPSProvider that PushTarget expects.

type SPSProvider 

type SPSProvider func() (sps, pps []byte, isH264 bool)

SPSProvider returns the source camera's current SPS/PPS (raw NALUs, no start code) so an RTMP target can initialize its track, and reports whether the source is H.264 (RTMP targets require H.264; H.265 sources are rejected).

type SilenceAACConfig 

type SilenceAACConfig struct {
	SampleRate           int
	Channels             int
	FrameDurationMs      int
	MaxContinuousSilence time.Duration
}

SilenceAACConfig configures a SilenceAACGenerator. Zero values get defaults.

type SilenceAACGenerator 

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

SilenceAACGenerator emits pre-computed silent AAC-LC frames at a configurable rate via a channel. It auto-stops after MaxContinuousSilence (default 5s) unless SourceActive() is called to reset the silence timer. Channel sends are non-blocking — the producer never blocks on a full channel.

Typical use: provide a silent audio track when the source camera has no audio, so that downstream consumers (RTMP targets, HLS) always have an audio stream.

func NewSilenceAACGenerator 

func NewSilenceAACGenerator() *SilenceAACGenerator

NewSilenceAACGenerator creates a generator with default configuration (48 kHz stereo, 10 fps, 5 s auto-stop timeout).

func NewSilenceAACGeneratorWith 

func NewSilenceAACGeneratorWith(cfg SilenceAACConfig) *SilenceAACGenerator

NewSilenceAACGeneratorWith creates a generator with the provided config. Zero-valued fields are replaced by defaults.

func (*SilenceAACGenerator) Config 

func (g *SilenceAACGenerator) Config() []byte

Config returns the AudioSpecificConfig marshaled bytes for the generator's configuration (AAC-LC, configured sample rate, configured channels).

func (*SilenceAACGenerator) SourceActive 

func (g *SilenceAACGenerator) SourceActive()

SourceActive resets the silence timeout. Call this when the audio source is known to be active (e.g. when video frames arrive) to prevent the generator from auto-stopping.

func (*SilenceAACGenerator) Start 

func (g *SilenceAACGenerator) Start(ctx context.Context) <-chan []byte

Start begins emitting one silent AAC-LC frame every FrameDurationMs on the returned channel. The generator runs until one of:

  • ctx is cancelled
  • Stop() is called
  • MaxContinuousSilence elapses without a SourceActive() call

The channel is closed when emission stops. Sends are non-blocking — the producer goroutine drops frames if the channel buffer is full.

func (*SilenceAACGenerator) Stop 

func (g *SilenceAACGenerator) Stop()

Stop cancels generation and closes the channel. Safe to call multiple times.

type StreamURLProvider added in v0.8.1 

type StreamURLProvider func(cameraID string) string

StreamURLProvider returns the source camera's stream URL (e.g. rtsp://...) for FFmpeg relay mode. If it returns empty, SourceURL from config is used.

type TargetStatus 

type TargetStatus struct {
	ID        string      `json:"id"`
	Name      string      `json:"name"`
	Protocol  string      `json:"protocol"`        // "rtmp" or "rtsp"
	URL       string      `json:"url"`             // target URL (masked in UI as needed)
	Status    RelayStatus `json:"status"`          // idle/connecting/streaming/reconnecting/error
	Kbps      float64     `json:"kbps"`            // recent outbound bitrate (kbps)
	Enabled   bool        `json:"enabled"`         // whether the target is active
	Uptime    string      `json:"uptime"`          // human duration since streaming started
	Error     string      `json:"error,omitempty"` // last error message (empty when healthy)
	UpdatedAt time.Time   `json:"updated_at"`      // last status change / sample time

	// Extended runtime status fields (populated when target is streaming).
	Platform            string  `json:"platform"`             // platform preset name
	TranscodePolicy     string  `json:"transcode_policy"`     // auto/force_sw/off
	TranscodeStatus     string  `json:"transcode_status"`     // "idle"/"active_hw"/"active_sw"/"throttled"/"error"
	TranscodeResolution string  `json:"transcode_resolution"` // current output resolution
	AudioCodec          string  `json:"audio_codec"`          // "aac"/"g711_mu"/"g711_a"/"silent"
	TemperatureC        int     `json:"temperature_c"`        // 0 when not transcoding
	RestartCount        int     `json:"restart_count"`        // transcoder restart count
	AVDriftMs           float64 `json:"av_drift_ms"`          // current A/V PTS drift in ms
}

TargetStatus is the JSON-serializable runtime status of one push-out target, returned by GET /api/cameras/{id}/push-status and surfaced in the camera card.

type VideoPresetOverrides 

type VideoPresetOverrides struct {
	Resolution       string
	Framerate        int
	VideoBitrateKbps int
	GopSeconds       int
	Profile          string
	Bframes          int
}

VideoPresetOverrides mirrors config.VideoPresetOverrides to avoid an import cycle between config and relay.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.