engine

README

lumen

A Go media engine for RTSP cameras: pull streams, fan out frames to viewers, record continuously to HLS, remux recordings to MP4, and deliver live video over WebRTC.

Features

• RTSP pull with exponential-backoff reconnect, a no-keyframe watchdog, TCP transport, and per-session epochs so downstream consumers can always tell which session a frame belongs to
• Frame hub fan-out: one bounded queue per viewer, slow viewers drop frames instead of blocking the pull loop
• HLS recording (MPEG-TS, archive mode): time-windowed directories with self-contained segments and final playlists, segment rotation at keyframes, and a session-aware recorder that starts a new window after every reconnect
• MP4 remux: lossless TS to MP4 conversion of a recorded window, with movie-time alignment that preserves the original audio/video start offset
• WebRTC output (webrtcview subpackage): H.264 track delivery driven by source PTS, with frame-rate adaptation, gap handling, and peer-connection lifecycle management
• Unified timestamps: audio and video share one synchronized clock (via gortsplib), with NTP fallback when RTCP sender reports are unavailable

Requirements

• Go 1.26 or newer

Install

go get github.com/LanceAdd/lumen

Quick start

package main

import (
    "context"
    "log"
    "time"

    "github.com/LanceAdd/lumen"
)

func main() {
    eng := engine.New(engine.Config{
        Context:        context.Background(),
        Log:            stdLogger{},
        RecordDir:      "/var/lib/lumen/record",
        ConnectTimeout: 3 * time.Second,
        ReadTimeout:    5 * time.Second,
    }, engine.Events{
        OnChannelStatus: func(streamID string, online bool) { /* update your online state */ },
        OnRecordClosed:  func(ev engine.RecordEvent) { /* index the finished window */ },
    })
    defer eng.Shutdown()

    err := eng.ApplyChannel(engine.ChannelConfig{
        Id:             "cam-01",
        Url:            "rtsp://user:pass@192.168.1.10/stream1",
        RecordEnabled:  true,
        SegmentSeconds: 300,
    })
    if err != nil {
        log.Fatal(err)
    }
}

Config.Context and Config.Log are required; New panics if they are missing.

Watching frames

sub, err := eng.Subscribe("cam-01")
if err != nil {
    return err
}
defer sub.Cancel()

for {
    select {
    case <-sub.Done:
        return
    case frame := <-sub.Frames:
        // frame.AU, frame.PTS, frame.NTP, frame.SessionID, frame.Key ...
    }
}

See the Subscription type documentation for lifecycle rules.

WebRTC

answer, err := webrtcview.Offer(ctx, webrtcview.OfferInput{
    Engine:        eng,
    Logger:        stdLogger{},
    StreamID:      streamID,
    SPS:           info.Video.SPS,
    OfferSDP:      offerSDP,
    ICEServerURLs: []string{"stun:stun.example.org:3478"},
})

Recording layout and MP4

Each recording window is a directory:

{recordDir}/{streamId}/{20060102_150405}/hls/   segN.ts + index.m3u8 + playlists

Use the layout helpers and the remux helpers instead of hardcoding paths:

hlsDir  := engine.WindowHLSDir(recordDir, rel)
mp4Path := engine.WindowMP4Path(recordDir, rel)

// One-shot remux into memory:
data, err := engine.RemuxWindowTS(hlsDir)

// Or remux and atomically write the MP4 (tmp file + rename):
size, err := engine.WriteWindowMP4(hlsDir, mp4Path)

License

MIT

Documentation

Index

• Variables
• func RemuxWindowTS(hlsDir string) ([]byte, error)
• func WindowHLSDir(recordDir, rel string) string
• func WindowMP4Path(recordDir, rel string) string
• func WriteWindowMP4(hlsDir, mp4Path string) (int64, error)
• type AACSpec
• type ChannelConfig
• type Config
• type Engine
  • func New(cfg Config, events Events) *Engine
  • func (e *Engine) ApplyChannel(cfg ChannelConfig) error
  • func (e *Engine) RemoveChannel(id string) error
  • func (e *Engine) RunAll(fn func() ([]ChannelConfig, error)) error
  • func (e *Engine) SetRecord(id string, on bool) error
  • func (e *Engine) SetSegment(id string, seconds int) error
  • func (e *Engine) Shutdown()
  • func (e *Engine) Snapshot(id string) (online bool, recording bool)
  • func (e *Engine) Subscribe(id string) (*Subscription, error)
  • func (e *Engine) TrackInfoContext(ctx context.Context, id string) (*TrackSpec, error)
• type Events
• type Frame
• type H264Spec
• type Kind
• type Logger
• type RecordEvent
• type Subscription
• type TrackSpec

Variables

var (
	ErrChannelNotFound = errors.New("channel not found")
	ErrChannelNoCodec  = errors.New("channel codec not ready")
	ErrEngineClosed    = errors.New("engine already shut down")
)

Functions

func RemuxWindowTS

func RemuxWindowTS(hlsDir string) ([]byte, error)

func WindowHLSDir

func WindowHLSDir(recordDir, rel string) string

func WindowMP4Path

func WindowMP4Path(recordDir, rel string) string

func WriteWindowMP4

func WriteWindowMP4(hlsDir, mp4Path string) (int64, error)

Types

type AACSpec

type AACSpec struct {
	SampleRate   int
	ChannelCount int
}

type ChannelConfig

type ChannelConfig struct {
	Id             string
	Name           string
	Url            string
	OnDemand       bool
	RecordEnabled  bool
	SegmentSeconds int
}

type Config

type Config struct {
	Context         context.Context
	Log             Logger
	RecordDir       string
	ConnectTimeout  time.Duration
	ReadTimeout     time.Duration
	NoKeyframeAfter time.Duration
}

type Engine

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

func New

func New(cfg Config, events Events) *Engine

func (*Engine) ApplyChannel

func (e *Engine) ApplyChannel(cfg ChannelConfig) error

func (*Engine) RemoveChannel

func (e *Engine) RemoveChannel(id string) error

func (*Engine) RunAll

func (e *Engine) RunAll(fn func() ([]ChannelConfig, error)) error

func (*Engine) SetRecord

func (e *Engine) SetRecord(id string, on bool) error

func (*Engine) SetSegment

func (e *Engine) SetSegment(id string, seconds int) error

func (*Engine) Shutdown

func (e *Engine) Shutdown()

func (*Engine) Snapshot

func (e *Engine) Snapshot(id string) (online bool, recording bool)

func (*Engine) Subscribe

func (e *Engine) Subscribe(id string) (*Subscription, error)

func (*Engine) TrackInfoContext

func (e *Engine) TrackInfoContext(ctx context.Context, id string) (*TrackSpec, error)

type Events

type Events struct {
	OnChannelStatus func(streamId string, online bool)
	OnRecordClosed  func(ev RecordEvent)
}

type Frame

type Frame struct {
	SessionID uint64
	Kind      Kind
	PTS       time.Duration
	NTP       time.Time
	Key       bool
	AU        [][]byte
}

type H264Spec

type H264Spec struct {
	SPS []byte
	PPS []byte
}

type Kind

type Kind uint8

const (
	KindVideo Kind = iota + 1
	KindAudio
)

type Logger

type Logger interface {
	Debugf(format string, args ...interface{})
	Infof(format string, args ...interface{})
	Warnf(format string, args ...interface{})
	Errorf(format string, args ...interface{})
}

type RecordEvent

type RecordEvent struct {
	StreamId string
	FilePath string
	Start    time.Time
	End      time.Time
	Size     int64
	Segments int
}

type Subscription

type Subscription struct {
	Frames <-chan *Frame
	Done   <-chan struct{}
	Cancel func()
}

type TrackSpec

type TrackSpec struct {
	Video *H264Spec
	Audio *AACSpec
}