storm/skald
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Checkpoint audio-only Skald fork work

Browse Source
This commit is contained in:
Storm Dragon
2026-05-18 13:06:57 -04:00
parent a8ada950d5
commit 965347cad4
48 changed files with 1080 additions and 3651 deletions
Download Patch File Download Diff File Expand all files Collapse all files
diskwriter/diskwriter.go
+39 -191
View File
@@ -22,7 +22,6 @@ import (
"github.com/jech/samplebuilder"
gcodecs "git.stormux.org/storm/skald/codecs"
"git.stormux.org/storm/skald/conn"
"git.stormux.org/storm/skald/hall"
"git.stormux.org/storm/skald/rtptime"
@@ -30,7 +29,6 @@ import (
const (
audioMaxLate = 32
videoMaxLate = 256
)
var Directory string
@@ -82,7 +80,7 @@ func (client *Client) Data() map[string]interface{} {
return nil
}
func (client *Client) PushClient(group, kind, id, username string, perms []string, data map[string]interface{}) error {
func (client *Client) PushClient(hall, kind, id, username string, perms []string, data map[string]interface{}) error {
return nil
}
@@ -112,7 +110,7 @@ func (client *Client) Addr() net.Addr {
return nil
}
func (client *Client) Joined(group, kind string) error {
func (client *Client) Joined(hall, kind string) error {
return nil
}
@@ -173,16 +171,14 @@ type diskConn struct {
client *Client
directory string
username string
hasVideo bool
mu sync.Mutex
file *os.File
remote conn.Up
tracks []*diskTrack
width, height uint32
lastWarning time.Time
originLocal time.Time
originRemote uint64
mu sync.Mutex
file *os.File
remote conn.Up
tracks []*diskTrack
lastWarning time.Time
originLocal time.Time
originRemote uint64
}
// called locked
@@ -315,14 +311,10 @@ type diskTrack struct {
remoteNTP uint64
remoteRTP uint32
kfRequested time.Time
lastKf time.Time
savedKf *rtp.Packet
}
func newDiskConn(client *Client, directory string, up conn.Up, remoteTracks []conn.UpTrack) (*diskConn, error) {
var audio, video conn.UpTrack
var audio conn.UpTrack
for _, remote := range remoteTracks {
codec := remote.Codec().MimeType
@@ -332,30 +324,16 @@ func newDiskConn(client *Client, directory string, up conn.Up, remoteTracks []co
} else {
client.hall.WallOps("Multiple audio tracks, recording just one")
}
} else if strings.EqualFold(codec, "video/vp8") ||
strings.EqualFold(codec, "video/vp9") ||
strings.EqualFold(codec, "video/h264") {
if video == nil || video.Label() == "l" {
video = remote
} else if remote.Label() != "l" {
client.hall.WallOps("Multiple video tracks, recording just one")
}
} else {
client.hall.WallOps("Unknown codec, " + codec + ", not recording")
}
}
if video == nil && audio == nil {
if audio == nil {
return nil, errors.New("no usable tracks found")
}
tracks := make([]conn.UpTrack, 0, 2)
if audio != nil {
tracks = append(tracks, audio)
}
if video != nil {
tracks = append(tracks, video)
}
tracks := []conn.UpTrack{audio}
_, username := up.User()
conn := diskConn{
@@ -374,24 +352,6 @@ func newDiskConn(client *Client, directory string, up conn.Up, remoteTracks []co
audioMaxLate,
&codecs.OpusPacket{}, codec.ClockRate,
)
} else if strings.EqualFold(codec.MimeType, "video/vp8") {
builder = samplebuilder.New(
videoMaxLate,
&codecs.VP8Packet{}, codec.ClockRate,
)
conn.hasVideo = true
} else if strings.EqualFold(codec.MimeType, "video/vp9") {
builder = samplebuilder.New(
videoMaxLate, &codecs.VP9Packet{},
codec.ClockRate,
)
conn.hasVideo = true
} else if strings.EqualFold(codec.MimeType, "video/h264") {
builder = samplebuilder.New(
videoMaxLate, &codecs.H264Packet{},
codec.ClockRate,
)
conn.hasVideo = true
} else {
// this shouldn't happen
return nil, errors.New(
@@ -406,8 +366,6 @@ func newDiskConn(client *Client, directory string, up conn.Up, remoteTracks []co
conn.tracks = append(conn.tracks, track)
}
// Only do this after all tracks have been added to conn, to avoid
// racing on hasVideo.
for _, t := range conn.tracks {
err := t.remote.AddLocal(t)
if err != nil {
@@ -454,7 +412,7 @@ func (t *diskTrack) Write(buf []byte) (int, error) {
fetch(t, lastSeqno+i)
}
} else {
requestKeyframe(t)
}
t.lastSeqno = some(uint32(p.SequenceNumber))
} else {
@@ -462,7 +420,7 @@ func (t *diskTrack) Write(buf []byte) (int, error) {
count := lastSeqno - p.SequenceNumber
if count >= 512 {
t.lastSeqno = none
requestKeyframe(t)
}
}
} else {
@@ -491,41 +449,14 @@ func fetch(t *diskTrack, seqno uint16) {
t.writeRTP(p)
}
func requestKeyframe(t *diskTrack) {
now := time.Now()
if now.Sub(t.kfRequested) > 500*time.Millisecond {
t.remote.RequestKeyframe()
t.kfRequested = now
}
}
// writeRTP writes the packet without fetching lost packets
// Called locked.
func (t *diskTrack) writeRTP(p *rtp.Packet) error {
codec := t.remote.Codec().MimeType
if len(codec) > 6 && strings.EqualFold(codec[:6], "video/") {
kf, _ := gcodecs.Keyframe(codec, p)
if kf {
t.savedKf = p
t.lastKf = time.Now()
if !valid(t.origin) {
t.setOrigin(
p.Timestamp, time.Now(),
t.remote.Codec().ClockRate,
)
}
} else if time.Since(t.lastKf) > 4*time.Second {
requestKeyframe(t)
}
}
if !valid(t.origin) {
if !t.conn.hasVideo || !t.conn.originLocal.Equal(time.Time{}) {
t.setOrigin(
p.Timestamp, time.Now(),
t.remote.Codec().ClockRate,
)
}
t.setOrigin(
p.Timestamp, time.Now(),
t.remote.Codec().ClockRate,
)
}
t.builder.Push(p)
@@ -537,8 +468,6 @@ func (t *diskTrack) writeRTP(p *rtp.Packet) error {
// samples will be flushed even if they are preceded by incomplete
// samples.
func (t *diskTrack) writeBuffered(force bool) error {
codec := t.remote.Codec().MimeType
for {
var sample *media.Sample
var ts uint32
@@ -561,39 +490,13 @@ func (t *diskTrack) writeBuffered(force bool) error {
t.conn.close()
}
var keyframe bool
if len(codec) > 6 && strings.EqualFold(codec[:6], "video/") {
if t.savedKf == nil {
keyframe = false
} else {
keyframe = (ts == t.savedKf.Timestamp)
}
if keyframe {
w, h := gcodecs.KeyframeDimensions(
codec, t.savedKf,
if t.writer == nil {
err := t.conn.initWriter(t, ts)
if err != nil {
t.conn.warn(
"Write to disk " + err.Error(),
)
err := t.conn.initWriter(w, h, t, ts)
if err != nil {
t.conn.warn(
"Write to disk " + err.Error(),
)
return err
}
}
} else {
keyframe = true
if t.writer == nil {
if !t.conn.hasVideo {
err := t.conn.initWriter(0, 0, t, ts)
if err != nil {
t.conn.warn(
"Write to disk " +
err.Error(),
)
return err
}
}
return err
}
}
@@ -608,6 +511,7 @@ func (t *diskTrack) writeBuffered(force bool) error {
tm := (ts - value(t.origin)) /
(t.remote.Codec().ClockRate / 1000)
keyframe := true
_, err := t.writer.Write(keyframe, int64(tm), sample.Data)
if err != nil {
return err
@@ -722,67 +626,23 @@ func (t *diskTrack) adjustOrigin(ts uint32) {
}
// called locked
func (conn *diskConn) initWriter(width, height uint32, track *diskTrack, ts uint32) error {
func (conn *diskConn) initWriter(track *diskTrack, ts uint32) error {
if conn.file != nil {
if width == conn.width && height == conn.height {
return nil
} else {
conn.close()
}
conn.close()
}
isWebm := true
var desc []mkvcore.TrackDescription
for i, t := range conn.tracks {
var entry webm.TrackEntry
codec := t.remote.Codec()
if strings.EqualFold(codec.MimeType, "audio/opus") {
entry = webm.TrackEntry{
Name: "Audio",
TrackNumber: uint64(i + 1),
CodecID: "A_OPUS",
TrackType: 2,
Audio: &webm.Audio{
SamplingFrequency: float64(codec.ClockRate),
Channels: uint64(codec.Channels),
},
}
} else if strings.EqualFold(codec.MimeType, "video/vp8") {
entry = webm.TrackEntry{
Name: "Video",
TrackNumber: uint64(i + 1),
CodecID: "V_VP8",
TrackType: 1,
Video: &webm.Video{
PixelWidth: uint64(width),
PixelHeight: uint64(height),
},
}
} else if strings.EqualFold(codec.MimeType, "video/vp9") {
entry = webm.TrackEntry{
Name: "Video",
TrackNumber: uint64(i + 1),
CodecID: "V_VP9",
TrackType: 1,
Video: &webm.Video{
PixelWidth: uint64(width),
PixelHeight: uint64(height),
},
}
} else if strings.EqualFold(codec.MimeType, "video/h264") {
entry = webm.TrackEntry{
Name: "Video",
TrackNumber: uint64(i + 1),
CodecID: "V_MPEG4/ISO/AVC",
TrackType: 1,
Video: &webm.Video{
PixelWidth: uint64(width),
PixelHeight: uint64(height),
},
}
isWebm = false
} else {
return errors.New("unknown track type")
entry := webm.TrackEntry{
Name: "Audio",
TrackNumber: uint64(i + 1),
CodecID: "A_OPUS",
TrackType: 2,
Audio: &webm.Audio{
SamplingFrequency: float64(codec.ClockRate),
Channels: uint64(codec.Channels),
},
}
desc = append(desc,
mkvcore.TrackDescription{
@@ -792,27 +652,18 @@ func (conn *diskConn) initWriter(width, height uint32, track *diskTrack, ts uint
)
}
extension := "webm"
header := webm.DefaultEBMLHeader
if !isWebm {
extension = "mkv"
h := *header
h.DocType = "matroska"
header = &h
}
if track != nil {
track.adjustOrigin(ts)
}
err := conn.open(extension)
err := conn.open("webm")
if err != nil {
return err
}
interceptor, err := mkvcore.NewMultiTrackBlockSorter(
// must be larger than the samplebuilder's MaxLate.
mkvcore.WithMaxDelayedPackets(videoMaxLate+16),
mkvcore.WithMaxDelayedPackets(audioMaxLate+16),
mkvcore.WithSortRule(mkvcore.BlockSorterWriteOutdated),
)
if err != nil {
@@ -823,7 +674,7 @@ func (conn *diskConn) initWriter(width, height uint32, track *diskTrack, ts uint
ws, err := mkvcore.NewSimpleBlockWriter(
conn.file, desc,
mkvcore.WithEBMLHeader(header),
mkvcore.WithEBMLHeader(webm.DefaultEBMLHeader),
mkvcore.WithSegmentInfo(webm.DefaultSegmentInfo),
mkvcore.WithBlockInterceptor(interceptor),
)
@@ -839,9 +690,6 @@ func (conn *diskConn) initWriter(width, height uint32, track *diskTrack, ts uint
return errors.New("unexpected number of writers")
}
conn.width = width
conn.height = height
for i, t := range conn.tracks {
t.writer = ws[i]
}