+2 -1

.github/workflows/build.yaml

··· 33 33 - name: make node 34 34 run: | 35 35 sudo apt install podman -y 36 36 - make in-container DOCKER_OPTS="-e STREAMPLACE_TEST_COUNT=1" IN_CONTAINER_CMD="${{ matrix.cmd }}" 36 36 + make in-container DOCKER_OPTS="-e STREAMPLACE_IGNORE_LEAKS=true -e STREAMPLACE_TEST_COUNT=1 -e CI=true" IN_CONTAINER_CMD="${{ matrix.cmd }}" 37 37 38 38 darwin: 39 39 name: ${{ matrix.name }} ··· 57 57 - name: Build Streamplace 58 58 env: 59 59 STREAMPLACE_TEST_COUNT: "1" 60 60 + STREAMPLACE_IGNORE_LEAKS: "true" 60 61 run: | 61 62 export GOTOOLCHAIN=go1.23.6 \ 62 63 && brew install go \

+6 -1

.gitlab-ci.yml

··· 113 113 interruptible: true 114 114 image: "$CI_REGISTRY_IMAGE:builder-$DOCKERFILE_HASH" 115 115 timeout: 2 hours 116 116 + variables: 117 117 + STREAMPLACE_IGNORE_LEAKS: "true" 118 118 + STREAMPLACE_TEST_COUNT: "1" 116 119 script: 117 120 - git fetch --unshallow || echo 'already unshallow' 118 121 - make ci-test -j$(nproc) ··· 194 197 image: ghcr.io/cirruslabs/macos-sonoma-xcode:16.1 195 198 tags: 196 199 - tart-installed 200 200 + variables: 201 201 + STREAMPLACE_IGNORE_LEAKS: "true" 202 202 + STREAMPLACE_TEST_COUNT: "1" 197 203 timeout: 2 hours 198 204 script: 199 205 # workaround for https://gitlab.com/gitlab-org/gitlab/-/issues/501457 ··· 213 219 - "$(which xcodeproj) --version" 214 220 - > 215 221 export GOTOOLCHAIN=go1.23.6 216 216 - && export STREAMPLACE_TEST_COUNT=10 217 222 && python3 -m pip install virtualenv 218 223 && python3 -m virtualenv ~/venv 219 224 && source ~/venv/bin/activate

+35

hack/sine-time.sh

··· 1 1 + #!/bin/bash 2 2 + 3 3 + # Generates a video with a sine wave that loops 60 times for testing the segmenter and whatnot 4 4 + 5 5 + set -euo pipefail 6 6 + 7 7 + DURATION="${DURATION:-60}" 8 8 + WIDTH="${WIDTH:-1280}" 9 9 + HEIGHT="${HEIGHT:-720}" 10 10 + HALF_HEIGHT=$((HEIGHT / 2)) 11 11 + 12 12 + ffmpeg -y \ 13 13 + -f lavfi -i "aevalsrc=0.125 * sin(2 * PI * (150+(800*mod(t\,1))) * mod(t\,1)):sample_rate=48000" \ 14 14 + -filter_complex " 15 15 + color=c=green:s=${WIDTH}x${HALF_HEIGHT}:r=60:d=${DURATION},format=yuv420p[green]; 16 16 + color=c=blue:s=${WIDTH}x${HALF_HEIGHT}:r=60:d=${DURATION},format=yuv420p[blue]; 17 17 + [green][blue]overlay=x='mod(((n-1)/60),1)*(overlay_w+(overlay_w/60)+1)'[colorfade]; 18 18 + [0:a]asplit[audio][audio2]; 19 19 + [audio2]showwaves=split_channels=1:s=${WIDTH}x${HALF_HEIGHT}:rate=25,fps=60[waveform]; 20 20 + [colorfade][waveform]vstack[video]; 21 21 + " \ 22 22 + -map "[video]" \ 23 23 + -map "[audio]" \ 24 24 + -c:v libx264 \ 25 25 + -preset ultrafast \ 26 26 + -g 60 \ 27 27 + -keyint_min 60 \ 28 28 + -x264-params "keyint=60:scenecut=0:bframes=0" \ 29 29 + -crf 23 \ 30 30 + -c:a libopus \ 31 31 + -b:a 128k \ 32 32 + -t "${DURATION}" \ 33 33 + output_looped.mp4 34 34 + 35 35 + ffplay output_looped.mp4

+7 -8

js/app/components/player/controls.tsx

··· 17 17 import { Animated, Pressable } from "react-native"; 18 18 import { 19 19 Button, 20 20 - Adapt, 21 20 H3, 22 21 ListItem, 23 22 Popover, ··· 31 30 H5, 32 31 Paragraph, 33 32 Slider, 33 33 + Adapt, 34 34 } from "tamagui"; 35 35 import { PlayerProps, PROTOCOL_HLS, PROTOCOL_WEBRTC } from "./props"; 36 36 import { ··· 498 498 onPress={() => setMenu("root")} 499 499 /> 500 500 </YGroup.Item> 501 501 - {/* // #HLS-CAUTERIZATION */} 502 502 - {/* <Separator /> 501 501 + <Separator /> 503 502 <YGroup.Item> 504 503 <ListItem 505 504 hoverTheme ··· 510 509 iconAfter={protocol === PROTOCOL_HLS ? CheckCircle : Circle} 511 510 onPress={() => dispatch(setProtocol(PROTOCOL_HLS))} 512 511 /> 513 513 - </YGroup.Item> */} 514 514 - {/* <Separator /> 515 515 - <YGroup.Item> 512 512 + </YGroup.Item> 513 513 + <Separator /> 514 514 + {/* <YGroup.Item> 516 515 <ListItem 517 516 hoverTheme 518 517 pressTheme ··· 538 537 } 539 538 onPress={() => dispatch(setProtocol(PROTOCOL_PROGRESSIVE_WEBM))} 540 539 /> 541 541 - </YGroup.Item> */} 542 542 - <Separator /> 540 540 + </YGroup.Item> 541 541 + <Separator /> */} 543 542 <YGroup.Item> 544 543 <ListItem 545 544 hoverTheme

+6 -2

pkg/api/playback.go

··· 264 264 265 265 ctx = log.WithLogValues(ctx, "did", did) 266 266 267 267 - mediaSigner, err := media.MakeMediaSignerExt(ctx, a.CLI, did, addrBytes) 268 268 - // mediaSigner, err := media.MakeMediaSigner(ctx, a.CLI, did, signer) 267 267 + var mediaSigner media.MediaSigner 268 268 + if a.CLI.ExternalSigning { 269 269 + mediaSigner, err = media.MakeMediaSignerExt(ctx, a.CLI, did, addrBytes) 270 270 + } else { 271 271 + mediaSigner, err = media.MakeMediaSigner(ctx, a.CLI, did, signer) 272 272 + } 269 273 if err != nil { 270 274 errors.WriteHTTPUnauthorized(w, "invalid authorization key (not valid secp256k1)", err) 271 275 return

+2 -1

pkg/cmd/streamplace.go

··· 146 146 fs.Bool("insecure", false, "DEPRECATED, does nothing.") 147 147 fs.StringVar(&cli.Color, "color", "", "'true' to enable colorized logging, 'false' to disable") 148 148 fs.BoolVar(&cli.Thumbnail, "thumbnail", true, "enable thumbnail generation") 149 149 - 149 149 + fs.BoolVar(&cli.SmearAudio, "smear-audio", true, "enable audio smearing to create 'perfect' segment timestamps") 150 150 + fs.BoolVar(&cli.ExternalSigning, "external-signing", false, "enable external signing via exec (prevents potential memory leak)") 150 151 version := fs.Bool("version", false, "print version and exit") 151 152 152 153 if runtime.GOOS == "linux" {

+1 -1

pkg/cmd/whip.go

··· 88 88 "filesrc name=filesrc ! qtdemux name=demux", 89 89 "demux.video_0 ! tee name=video_tee", 90 90 "demux.audio_0 ! tee name=audio_tee", 91 91 - "video_tee. ! queue ! h264parse ! video/x-h264,stream-format=byte-stream ! appsink name=videoappsink", 91 91 + "video_tee. ! queue ! h264parse config-interval=-1 ! video/x-h264,stream-format=byte-stream ! appsink name=videoappsink", 92 92 "audio_tee. ! queue ! opusparse ! appsink name=audioappsink", 93 93 // "matroskamux name=mux ! fakesink name=fakesink sync=true", 94 94 // "video_tee. ! mux.video_0",

+2

pkg/config/config.go

··· 87 87 LivepeerGatewayURL string 88 88 WHIPTest string 89 89 Thumbnail bool 90 90 + SmearAudio bool 91 91 + ExternalSigning bool 90 92 91 93 dataDirFlags []*string 92 94 }

+52 -45

pkg/director/stream_session.go

··· 70 70 71 71 g, ctx := errgroup.WithContext(ctx) 72 72 73 73 - for _, r := range allRenditions { 74 74 - g.Go(func() error { 75 75 - // #HLS-CAUTERIZATION 76 76 - return nil 77 77 - for { 78 78 - if ctx.Err() != nil { 79 79 - return nil 80 80 - } 81 81 - err := ss.mm.ToHLS(ctx, spseg.Creator, r.Name, ss.hls) 82 82 - if ctx.Err() != nil { 83 83 - return nil 84 84 - } 85 85 - log.Warn(ctx, "hls failed, retrying in 5 seconds", "error", err) 86 86 - time.Sleep(time.Second * 5) 87 87 - } 88 88 - }) 89 89 - } 73 73 + // for _, r := range allRenditions { 74 74 + // g.Go(func() error { 75 75 + // for { 76 76 + // if ctx.Err() != nil { 77 77 + // return nil 78 78 + // } 79 79 + // err := ss.mm.ToHLS(ctx, spseg.Creator, r.Name, ss.hls) 80 80 + // if ctx.Err() != nil { 81 81 + // return nil 82 82 + // } 83 83 + // log.Warn(ctx, "hls failed, retrying in 5 seconds", "error", err) 84 84 + // time.Sleep(time.Second * 5) 85 85 + // } 86 86 + // }) 87 87 + // } 90 88 91 89 for { 92 90 select { ··· 119 117 } 120 118 121 119 ss.bus.Publish(spseg.Creator, spseg) 120 120 + go ss.TryAddToHLS(ctx, spseg, "source", not.Data) 122 121 123 122 if ss.cli.Thumbnail { 124 123 go func() { ··· 217 216 } 218 217 defer fd.Close() 219 218 fd.Write(seg) 220 220 - // go ss.TryAddToHLS(ctx, spseg, rs[i].Name, seg) 219 219 + go ss.TryAddToHLS(ctx, spseg, rs[i].Name, seg) 221 220 go ss.mm.PublishSegment(ctx, spseg.Creator, rs[i].Name, &segchanman.Seg{ 222 221 Filepath: fd.Name(), 223 222 Data: seg, ··· 226 225 return nil 227 226 } 228 227 229 229 - // func (ss *StreamSession) TryAddToHLS(ctx context.Context, spseg *streamplace.Segment, rendition string, data []byte) { 230 230 - // ctx = log.WithLogValues(ctx, "rendition", rendition) 231 231 - // err := ss.AddToHLS(ctx, spseg, rendition, data) 232 232 - // if err != nil { 233 233 - // log.Error(ctx, "could not add to hls", "error", err) 234 234 - // } 235 235 - // } 228 228 + func (ss *StreamSession) TryAddToHLS(ctx context.Context, spseg *streamplace.Segment, rendition string, data []byte) { 229 229 + ctx = log.WithLogValues(ctx, "rendition", rendition) 230 230 + err := ss.AddToHLS(ctx, spseg, rendition, data) 231 231 + if err != nil { 232 232 + log.Error(ctx, "could not add to hls", "error", err) 233 233 + } 234 234 + } 236 235 237 237 - // func (ss *StreamSession) AddToHLS(ctx context.Context, spseg *streamplace.Segment, rendition string, data []byte) error { 238 238 - // buf := bytes.Buffer{} 239 239 - // dur, err := media.MP4ToMPEGTS(ctx, bytes.NewReader(data), &buf) 240 240 - // if err != nil { 241 241 - // return err 242 242 - // } 243 243 - // newSeg := &streamplace.Segment{ 244 244 - // LexiconTypeID: "place.stream.segment", 245 245 - // Id: spseg.Id, 246 246 - // Creator: spseg.Creator, 247 247 - // StartTime: spseg.StartTime, 248 248 - // Duration: &dur, 249 249 - // Audio: spseg.Audio, 250 250 - // Video: spseg.Video, 251 251 - // SigningKey: spseg.SigningKey, 252 252 - // } 253 253 - // log.Debug(ctx, "transmuxed to mpegts, adding to hls", "rendition", rendition, "size", buf.Len()) 254 254 - // ss.hls.NewSegment(newSeg, rendition, buf.Bytes()) 255 255 - // return nil 256 256 - // } 236 236 + func (ss *StreamSession) AddToHLS(ctx context.Context, spseg *streamplace.Segment, rendition string, data []byte) error { 237 237 + buf := bytes.Buffer{} 238 238 + dur, err := media.MP4ToMPEGTS(ctx, bytes.NewReader(data), &buf) 239 239 + if err != nil { 240 240 + return err 241 241 + } 242 242 + // newSeg := &streamplace.Segment{ 243 243 + // LexiconTypeID: "place.stream.segment", 244 244 + // Id: spseg.Id, 245 245 + // Creator: spseg.Creator, 246 246 + // StartTime: spseg.StartTime, 247 247 + // Duration: &dur, 248 248 + // Audio: spseg.Audio, 249 249 + // Video: spseg.Video, 250 250 + // SigningKey: spseg.SigningKey, 251 251 + // } 252 252 + aqt, err := aqtime.FromString(spseg.StartTime) 253 253 + if err != nil { 254 254 + return err 255 255 + } 256 256 + log.Debug(ctx, "transmuxed to mpegts, adding to hls", "rendition", rendition, "size", buf.Len()) 257 257 + ss.hls.GetRendition(rendition).NewSegment(&media.Segment{ 258 258 + Buf: &buf, 259 259 + Duration: time.Duration(dur), 260 260 + Time: aqt.Time(), 261 261 + }) 262 262 + return nil 263 263 + }

+357

pkg/media/audio_smear.go

··· 1 1 + package media 2 2 + 3 3 + import ( 4 4 + "bytes" 5 5 + "context" 6 6 + "fmt" 7 7 + "io" 8 8 + "os" 9 9 + "strings" 10 10 + "time" 11 11 + 12 12 + "github.com/go-gst/go-gst/gst" 13 13 + "github.com/go-gst/go-gst/gst/app" 14 14 + "github.com/google/uuid" 15 15 + "stream.place/streamplace/pkg/log" 16 16 + ) 17 17 + 18 18 + type SegmentBuffer struct { 19 19 + bytes []byte 20 20 + pts *time.Duration 21 21 + dur *time.Duration 22 22 + } 23 23 + 24 24 + type SegmentData struct { 25 25 + Audio []SegmentBuffer 26 26 + AudioCaps string 27 27 + Video []SegmentBuffer 28 28 + VideoCaps string 29 29 + } 30 30 + 31 31 + func SmearAudioTimestamps(ctx context.Context, input io.Reader, output io.Writer) error { 32 32 + bs, err := io.ReadAll(input) 33 33 + if err != nil { 34 34 + return err 35 35 + } 36 36 + seg, err := ToBuffers(ctx, bytes.NewReader(bs)) 37 37 + if err != nil { 38 38 + // Write the input bytes to a file for debugging 39 39 + debugFile := fmt.Sprintf("audio_smear_debug_%s.mp4", uuid.New().String()) 40 40 + err = os.WriteFile(debugFile, bs, 0644) 41 41 + if err != nil { 42 42 + log.Log(ctx, "failed to write debug file", "error", err, "path", debugFile) 43 43 + } else { 44 44 + log.Log(ctx, "wrote debug file", "path", debugFile) 45 45 + } 46 46 + return err 47 47 + } 48 48 + 49 49 + err = seg.Normalize(ctx) 50 50 + if err != nil { 51 51 + return err 52 52 + } 53 53 + 54 54 + return JoinAudioVideo(ctx, seg, output) 55 55 + } 56 56 + 57 57 + func (s *SegmentData) Normalize(ctx context.Context) error { 58 58 + if len(s.Video) == 0 { 59 59 + return fmt.Errorf("no video segments") 60 60 + } 61 61 + if len(s.Audio) == 0 { 62 62 + return fmt.Errorf("no audio segments") 63 63 + } 64 64 + 65 65 + lastVideo := s.Video[len(s.Video)-1] 66 66 + lastAudio := s.Audio[len(s.Audio)-1] 67 67 + 68 68 + if lastVideo.pts == nil { 69 69 + return fmt.Errorf("last video segment has no pts") 70 70 + } 71 71 + if lastAudio.pts == nil { 72 72 + return fmt.Errorf("last audio segment has no pts") 73 73 + } 74 74 + 75 75 + videoEnd := lastVideo.pts.Nanoseconds() + lastVideo.dur.Nanoseconds() 76 76 + audioEnd := lastAudio.pts.Nanoseconds() + lastAudio.dur.Nanoseconds() 77 77 + 78 78 + diff := videoEnd - audioEnd 79 79 + diffPerAudio := diff / int64(len(s.Audio)-1) 80 80 + for i, audio := range s.Audio { 81 81 + newPts := time.Duration(audio.pts.Nanoseconds() + (diffPerAudio * int64(i))) 82 82 + audio.pts = &newPts 83 83 + s.Audio[i] = audio 84 84 + } 85 85 + 86 86 + lastVideo = s.Video[len(s.Video)-1] 87 87 + lastAudio = s.Audio[len(s.Audio)-1] 88 88 + videoEnd = lastVideo.pts.Nanoseconds() + lastVideo.dur.Nanoseconds() 89 89 + audioEnd = lastAudio.pts.Nanoseconds() + lastAudio.dur.Nanoseconds() 90 90 + 91 91 + return nil 92 92 + } 93 93 + 94 94 + func ToBuffers(ctx context.Context, input io.Reader) (*SegmentData, error) { 95 95 + ctx = log.WithLogValues(ctx, "func", "SplitAudioVideo") 96 96 + 97 97 + pipelineSlice := []string{ 98 98 + "appsrc name=mp4src ! qtdemux name=demux", 99 99 + "demux.video_0 ! queue ! h264parse name=videoparse disable-passthrough=true config-interval=-1 ! appsink sync=false name=videoappsink", 100 100 + "demux.audio_0 ! queue ! opusparse name=audioparse ! appsink sync=false name=audioappsink", 101 101 + } 102 102 + 103 103 + ctx, cancel := context.WithCancel(ctx) 104 104 + 105 105 + pipeline, err := gst.NewPipelineFromString(strings.Join(pipelineSlice, "\n")) 106 106 + if err != nil { 107 107 + return nil, fmt.Errorf("failed to create GStreamer pipeline: %w", err) 108 108 + } 109 109 + 110 110 + errCh := make(chan error) 111 111 + go func() { 112 112 + err := HandleBusMessages(ctx, pipeline) 113 113 + cancel() 114 114 + errCh <- err 115 115 + close(errCh) 116 116 + }() 117 117 + 118 118 + defer func() { 119 119 + cancel() 120 120 + err := <-errCh 121 121 + if err != nil { 122 122 + log.Error(ctx, "bus handler error", "error", err) 123 123 + } 124 124 + err = pipeline.BlockSetState(gst.StateNull) 125 125 + if err != nil { 126 126 + log.Error(ctx, "failed to set pipeline to null state", "error", err) 127 127 + } 128 128 + }() 129 129 + 130 130 + mp4src, err := pipeline.GetElementByName("mp4src") 131 131 + if err != nil { 132 132 + return nil, fmt.Errorf("failed to get mp4src element: %w", err) 133 133 + } 134 134 + src := app.SrcFromElement(mp4src) 135 135 + if src == nil { 136 136 + return nil, fmt.Errorf("failed to get mp4src element: %w", err) 137 137 + } 138 138 + src.SetCallbacks(&app.SourceCallbacks{ 139 139 + NeedDataFunc: ReaderNeedData(ctx, input), 140 140 + }) 141 141 + 142 142 + audioSinkElem, err := pipeline.GetElementByName("audioappsink") 143 143 + if err != nil { 144 144 + return nil, fmt.Errorf("failed to get audioappsink element: %w", err) 145 145 + } 146 146 + audioSink := app.SinkFromElement(audioSinkElem) 147 147 + if audioSink == nil { 148 148 + return nil, fmt.Errorf("failed to get audioappsink element: %w", err) 149 149 + } 150 150 + 151 151 + seg := SegmentData{ 152 152 + Audio: []SegmentBuffer{}, 153 153 + Video: []SegmentBuffer{}, 154 154 + } 155 155 + 156 156 + audioSink.SetCallbacks(&app.SinkCallbacks{ 157 157 + NewSampleFunc: func(sink *app.Sink) gst.FlowReturn { 158 158 + sample := sink.PullSample() 159 159 + if sample == nil { 160 160 + return gst.FlowOK 161 161 + } 162 162 + 163 163 + // Retrieve the buffer from the sample. 164 164 + buffer := sample.GetBuffer() 165 165 + // log.Log(ctx, "audio buffer", "presentation_timestamp", buffer.PresentationTimestamp(), "duration", buffer.Duration(), "dts", buffer.DecodingTimestamp()) 166 166 + bs := buffer.Map(gst.MapRead).Bytes() 167 167 + defer buffer.Unmap() 168 168 + sinkPads, err := sink.GetSinkPads() 169 169 + if err != nil { 170 170 + src.Error("could not get sink pads", err) 171 171 + return gst.FlowError 172 172 + } 173 173 + caps := sinkPads[0].GetCurrentCaps() 174 174 + if caps != nil { 175 175 + seg.AudioCaps = caps.String() 176 176 + } 177 177 + 178 178 + seg.Audio = append(seg.Audio, SegmentBuffer{ 179 179 + bytes: bs, 180 180 + pts: buffer.PresentationTimestamp().AsDuration(), 181 181 + dur: buffer.Duration().AsDuration(), 182 182 + }) 183 183 + 184 184 + if err != nil { 185 185 + src.Error("could not get sink pads", err) 186 186 + return gst.FlowError 187 187 + } 188 188 + 189 189 + return gst.FlowOK 190 190 + }, 191 191 + }) 192 192 + 193 193 + videoSinkElem, err := pipeline.GetElementByName("videoappsink") 194 194 + if err != nil { 195 195 + return nil, fmt.Errorf("failed to get videoappsink element: %w", err) 196 196 + } 197 197 + videoSink := app.SinkFromElement(videoSinkElem) 198 198 + if videoSink == nil { 199 199 + return nil, fmt.Errorf("failed to get videoappsink element: %w", err) 200 200 + } 201 201 + videoSink.SetCallbacks(&app.SinkCallbacks{ 202 202 + NewSampleFunc: func(sink *app.Sink) gst.FlowReturn { 203 203 + sample := sink.PullSample() 204 204 + if sample == nil { 205 205 + return gst.FlowOK 206 206 + } 207 207 + 208 208 + // Retrieve the buffer from the sample. 209 209 + buffer := sample.GetBuffer() 210 210 + // log.Log(ctx, "video buffer", "presentation_timestamp", buffer.PresentationTimestamp(), "duration", buffer.Duration()) 211 211 + bs := buffer.Map(gst.MapRead).Bytes() 212 212 + defer buffer.Unmap() 213 213 + sinkPads, err := sink.GetSinkPads() 214 214 + if err != nil { 215 215 + src.Error("could not get sink pads", err) 216 216 + return gst.FlowError 217 217 + } 218 218 + caps := sinkPads[0].GetCurrentCaps() 219 219 + if caps != nil { 220 220 + seg.VideoCaps = caps.String() 221 221 + } 222 222 + 223 223 + sb := SegmentBuffer{ 224 224 + bytes: bs, 225 225 + pts: buffer.PresentationTimestamp().AsDuration(), 226 226 + dur: buffer.Duration().AsDuration(), 227 227 + } 228 228 + 229 229 + // log.Log(ctx, "video buffer", "presentation_timestamp", sb.pts, "duration", sb.dur) 230 230 + if sb.pts == nil { 231 231 + sink.Error("no video pts", fmt.Errorf("no video pts")) 232 232 + return gst.FlowError 233 233 + } 234 234 + 235 235 + seg.Video = append(seg.Video, sb) 236 236 + 237 237 + return gst.FlowOK 238 238 + }, 239 239 + }) 240 240 + 241 241 + pipeline.SetState(gst.StatePlaying) 242 242 + 243 243 + <-ctx.Done() 244 244 + 245 245 + return &seg, <-errCh 246 246 + } 247 247 + 248 248 + func JoinAudioVideo(ctx context.Context, seg *SegmentData, output io.Writer) error { 249 249 + uu, _ := uuid.NewV7() 250 250 + ctx = log.WithLogValues(ctx, "func", "JoinAudioVideo", "uuid", uu.String()) 251 251 + 252 252 + pipelineSlice := []string{ 253 253 + "mp4mux name=mux ! appsink sync=false name=mp4sink", 254 254 + "appsrc name=videosrc format=time ! queue ! mux.video_0", 255 255 + "appsrc name=audiosrc format=time ! queue ! mux.audio_0", 256 256 + } 257 257 + 258 258 + ctx, cancel := context.WithCancel(ctx) 259 259 + 260 260 + pipeline, err := gst.NewPipelineFromString(strings.Join(pipelineSlice, "\n")) 261 261 + if err != nil { 262 262 + return fmt.Errorf("failed to create GStreamer pipeline: %w", err) 263 263 + } 264 264 + 265 265 + errCh := make(chan error) 266 266 + go func() { 267 267 + err := HandleBusMessages(ctx, pipeline) 268 268 + cancel() 269 269 + errCh <- err 270 270 + close(errCh) 271 271 + }() 272 272 + 273 273 + defer func() { 274 274 + cancel() 275 275 + err := <-errCh 276 276 + if err != nil { 277 277 + log.Error(ctx, "bus handler error", "error", err) 278 278 + } 279 279 + err = pipeline.BlockSetState(gst.StateNull) 280 280 + if err != nil { 281 281 + log.Error(ctx, "failed to set pipeline to null state", "error", err) 282 282 + } 283 283 + }() 284 284 + 285 285 + videoSrcElem, err := pipeline.GetElementByName("videosrc") 286 286 + if err != nil { 287 287 + return fmt.Errorf("failed to get videosrc element: %w", err) 288 288 + } 289 289 + videoSrc := app.SrcFromElement(videoSrcElem) 290 290 + if videoSrc == nil { 291 291 + return fmt.Errorf("failed to get videosrc element: %w", err) 292 292 + } 293 293 + videoSrc.SetCaps(gst.NewCapsFromString(seg.VideoCaps)) 294 294 + for _, seg := range seg.Video { 295 295 + buf := gst.NewBufferFromBytes(seg.bytes) 296 296 + if seg.pts != nil { 297 297 + buf.SetPresentationTimestamp(gst.ClockTime(uint64(seg.pts.Nanoseconds()))) 298 298 + } else { 299 299 + videoSrc.Error("no video pts", fmt.Errorf("no video pts")) 300 300 + return fmt.Errorf("no video pts") 301 301 + } 302 302 + if seg.dur != nil { 303 303 + buf.SetDuration(gst.ClockTime(uint64(seg.dur.Nanoseconds()))) 304 304 + } 305 305 + ret := videoSrc.PushBuffer(buf) 306 306 + if ret != gst.FlowOK { 307 307 + return fmt.Errorf("failed to push video buffer: %s", ret) 308 308 + } else { 309 309 + // log.Log(ctx, "pushed video buffer", "presentation_timestamp", seg.pts, "duration", seg.dur) 310 310 + } 311 311 + } 312 312 + 313 313 + audioSrcElem, err := pipeline.GetElementByName("audiosrc") 314 314 + if err != nil { 315 315 + return fmt.Errorf("failed to get audiosrc element: %w", err) 316 316 + } 317 317 + audioSrc := app.SrcFromElement(audioSrcElem) 318 318 + if audioSrc == nil { 319 319 + return fmt.Errorf("failed to get audiosrc element: %w", err) 320 320 + } 321 321 + audioSrc.SetCaps(gst.NewCapsFromString(seg.AudioCaps)) 322 322 + for _, seg := range seg.Audio { 323 323 + buf := gst.NewBufferFromBytes(seg.bytes) 324 324 + if seg.pts != nil { 325 325 + buf.SetPresentationTimestamp(gst.ClockTime(uint64(seg.pts.Nanoseconds()))) 326 326 + } 327 327 + if seg.dur != nil { 328 328 + buf.SetDuration(gst.ClockTime(uint64(seg.dur.Nanoseconds()))) 329 329 + } 330 330 + ret := audioSrc.PushBuffer(buf) 331 331 + if ret != gst.FlowOK { 332 332 + return fmt.Errorf("failed to push audio buffer: %s", ret) 333 333 + } else { 334 334 + // log.Log(ctx, "pushed audio buffer", "presentation_timestamp", seg.pts, "duration", seg.dur) 335 335 + } 336 336 + } 337 337 + 338 338 + videoSrc.EndStream() 339 339 + audioSrc.EndStream() 340 340 + mp4sinkElem, err := pipeline.GetElementByName("mp4sink") 341 341 + if err != nil { 342 342 + return fmt.Errorf("failed to get mp4sink element: %w", err) 343 343 + } 344 344 + mp4sink := app.SinkFromElement(mp4sinkElem) 345 345 + if mp4sink == nil { 346 346 + return fmt.Errorf("failed to get mp4sink element: %w", err) 347 347 + } 348 348 + mp4sink.SetCallbacks(&app.SinkCallbacks{ 349 349 + NewSampleFunc: WriterNewSample(ctx, output), 350 350 + }) 351 351 + 352 352 + pipeline.SetState(gst.StatePlaying) 353 353 + 354 354 + <-ctx.Done() 355 355 + 356 356 + return <-errCh 357 357 + }

+164

pkg/media/audio_smear_test.go

··· 1 1 + package media 2 2 + 3 3 + import ( 4 4 + "bytes" 5 5 + "context" 6 6 + "fmt" 7 7 + "os" 8 8 + "testing" 9 9 + "time" 10 10 + 11 11 + "github.com/go-gst/go-gst/gst" 12 12 + "github.com/go-gst/go-gst/gst/pbutils" 13 13 + "github.com/stretchr/testify/require" 14 14 + "go.uber.org/goleak" 15 15 + "golang.org/x/sync/errgroup" 16 16 + "stream.place/streamplace/pkg/gstinit" 17 17 + ) 18 18 + 19 19 + func TestAudioSmear(t *testing.T) { 20 20 + gstinit.InitGST() 21 21 + before := getLeakCount(t) 22 22 + defer checkGStreamerLeaks(t, before) 23 23 + ignore := goleak.IgnoreCurrent() 24 24 + defer goleak.VerifyNone(t, ignore) 25 25 + 26 26 + g, _ := errgroup.WithContext(context.Background()) 27 27 + for i := 0; i < streamplaceTestCount; i++ { 28 28 + g.Go(func() error { 29 29 + return testAudioSmearInner(t) 30 30 + }) 31 31 + } 32 32 + err := g.Wait() 33 33 + require.NoError(t, err) 34 34 + } 35 35 + 36 36 + func testAudioSmearInner(t *testing.T) error { 37 37 + uri := getFixture("duration-mismatch.mp4") 38 38 + 39 39 + // info, err := discoverer.DiscoverURI(fmt.Sprintf("file://%s", uri)) 40 40 + // if err != nil { 41 41 + // return err 42 42 + // } 43 43 + 44 44 + f, err := os.Open(uri) 45 45 + if err != nil { 46 46 + return err 47 47 + } 48 48 + defer f.Close() 49 49 + 50 50 + // audioBs := bytes.Buffer{} 51 51 + // videoBs := bytes.Buffer{} 52 52 + 53 53 + seg, err := ToBuffers(context.Background(), f) 54 54 + if err != nil { 55 55 + return err 56 56 + } 57 57 + 58 58 + err = seg.Normalize(context.Background()) 59 59 + if err != nil { 60 60 + return err 61 61 + } 62 62 + 63 63 + buf := bytes.Buffer{} 64 64 + err = JoinAudioVideo(context.Background(), seg, &buf) 65 65 + if err != nil { 66 66 + return err 67 67 + } 68 68 + 69 69 + require.Equal(t, 1191255, buf.Len()) 70 70 + 71 71 + // // Write audio and video buffers to temporary files for further analysis 72 72 + // tempDir := t.TempDir() 73 73 + 74 74 + // audioFilePath := fmt.Sprintf("%s/audio.mp4", tempDir) 75 75 + // videoFilePath := fmt.Sprintf("%s/video.mp4", tempDir) 76 76 + 77 77 + // // Write audio buffer to file 78 78 + // audioFile, err := os.Create(audioFilePath) 79 79 + // if err != nil { 80 80 + // return err 81 81 + // } 82 82 + // _, err = io.Copy(audioFile, bytes.NewReader(audioBs.Bytes())) 83 83 + // if err != nil { 84 84 + // return err 85 85 + // } 86 86 + // err = audioFile.Close() 87 87 + // if err != nil { 88 88 + // return err 89 89 + // } 90 90 + 91 91 + // // Write video buffer to file 92 92 + // videoFile, err := os.Create(videoFilePath) 93 93 + // if err != nil { 94 94 + // return err 95 95 + // } 96 96 + // _, err = io.Copy(videoFile, bytes.NewReader(videoBs.Bytes())) 97 97 + // if err != nil { 98 98 + // return err 99 99 + // } 100 100 + // err = videoFile.Close() 101 101 + // if err != nil { 102 102 + // return err 103 103 + // } 104 104 + 105 105 + // SmearAudioTimestamps(context.Background(), bytes.NewReader(audioBs.Bytes()), &bytes.Buffer{}) 106 106 + 107 107 + // checkSame(t, videoFile.Name(), getFixture("duration-mismatch-video.mp4")) 108 108 + // checkSame(t, audioFile.Name(), getFixture("duration-mismatch-audio.mp4")) 109 109 + // printDiscovererInfo(info) 110 110 + return nil 111 111 + 112 112 + } 113 113 + 114 114 + func checkSame(t *testing.T, v1, v2 string) { 115 115 + discoverer, err := pbutils.NewDiscoverer(gst.ClockTime(time.Second * 15)) 116 116 + if err != nil { 117 117 + panic(err) 118 118 + } 119 119 + 120 120 + info, err := discoverer.DiscoverURI(fmt.Sprintf("file://%s", v1)) 121 121 + require.NoError(t, err) 122 122 + dur1 := info.GetDuration().AsDuration() 123 123 + 124 124 + info, err = discoverer.DiscoverURI(fmt.Sprintf("file://%s", v2)) 125 125 + require.NoError(t, err) 126 126 + dur2 := info.GetDuration().AsDuration() 127 127 + 128 128 + require.Equal(t, *dur2, *dur1) 129 129 + } 130 130 + 131 131 + func printDiscovererInfo(info *pbutils.DiscovererInfo) { 132 132 + fmt.Println("URI:", info.GetURI()) 133 133 + fmt.Println("Duration:", info.GetDuration()) 134 134 + 135 135 + printTags(info) 136 136 + printStreamInfo(info.GetStreamInfo()) 137 137 + 138 138 + children := info.GetStreamList() 139 139 + fmt.Println("Children streams:") 140 140 + for _, child := range children { 141 141 + printStreamInfo(child) 142 142 + } 143 143 + } 144 144 + 145 145 + func printTags(info *pbutils.DiscovererInfo) { 146 146 + fmt.Println("Tags:") 147 147 + tags := info.GetTags() 148 148 + if tags != nil { 149 149 + fmt.Println(" ", tags) 150 150 + return 151 151 + } 152 152 + fmt.Println(" no tags") 153 153 + } 154 154 + 155 155 + func printStreamInfo(info *pbutils.DiscovererStreamInfo) { 156 156 + if info == nil { 157 157 + return 158 158 + } 159 159 + fmt.Println("Stream: ") 160 160 + fmt.Println(" Stream id:", info.GetStreamID()) 161 161 + if caps := info.GetCaps(); caps != nil { 162 162 + fmt.Println(" Format:", caps) 163 163 + } 164 164 + }

+1 -1

pkg/media/io_helpers_test.go

··· 41 41 fileSize := fileInfo.Size() 42 42 t.Logf("Test file size: %d bytes", fileSize) 43 43 g, ctx := errgroup.WithContext(context.Background()) 44 44 - ctx = log.WithDebugValue(ctx, map[string]map[string]int{"func": {"TestWriterNewSample": 9}}) 44 44 + // ctx = log.WithDebugValue(ctx, map[string]map[string]int{"func": {"TestWriterNewSample": 9}}) 45 45 for i := 0; i < streamplaceTestCount; i++ { 46 46 g.Go(func() error { 47 47 bs := bytes.Buffer{}

+40 -21

pkg/media/leak_test.go

··· 19 19 "stream.place/streamplace/pkg/gstinit" 20 20 ) 21 21 22 22 + const IGNORE_LEAKS = "STREAMPLACE_IGNORE_LEAKS" 22 23 const GST_DEBUG_NEEDED = "leaks:9,GST_TRACER:9" 23 24 const LEAK_LINE = "GST_TRACER :0:: object-alive" 24 25 ··· 29 30 var LeakDoneCh = make(chan struct{}) 30 31 31 32 func TestMain(m *testing.M) { 33 33 + if os.Getenv(IGNORE_LEAKS) != "" { 34 34 + gstinit.InitGST() 35 35 + os.Exit(m.Run()) 36 36 + return 37 37 + } 32 38 gstDebug := os.Getenv("GST_DEBUG") 33 39 if gstDebug == "" { 34 40 gstDebug = GST_DEBUG_NEEDED ··· 82 88 } 83 89 84 90 func getLeakCount(t *testing.T) int { 91 91 + if os.Getenv(IGNORE_LEAKS) != "" { 92 92 + return 0 93 93 + } 85 94 process, err := os.FindProcess(os.Getpid()) 86 95 LeakReportMutex.Lock() 87 96 LeakReport = []string{} 88 97 LeakReportMutex.Unlock() 89 98 90 90 - ch := make(chan struct{}) 91 91 - done := false 99 99 + // we want CI to be extra reliable here and a little slower is okay 100 100 + flushes := 2 101 101 + if os.Getenv("CI") != "" { 102 102 + flushes = 5 103 103 + } 92 104 93 93 - go func() { 94 94 - thing := &[]byte{} 95 95 - runtime.SetFinalizer(thing, func(thing *[]byte) { 96 96 - done = true 97 97 - ch <- struct{}{} 98 98 - }) 99 99 - }() 105 105 + for i := 0; i < flushes; i++ { 106 106 + ch := make(chan struct{}) 107 107 + done := false 108 108 + go func() { 109 109 + thing := &[]byte{} 110 110 + runtime.SetFinalizer(thing, func(thing *[]byte) { 111 111 + done = true 112 112 + ch <- struct{}{} 113 113 + }) 114 114 + }() 100 115 101 101 - go func() { 102 102 - runtime.GC() 103 103 - runtime.GC() 104 104 - for { 105 105 - if done { 106 106 - break 107 107 - } 116 116 + go func() { 108 117 runtime.GC() 109 118 runtime.GC() 110 110 - time.Sleep(500 * time.Millisecond) 111 111 - } 112 112 - }() 119 119 + for { 120 120 + if done { 121 121 + break 122 122 + } 123 123 + runtime.GC() 124 124 + runtime.GC() 125 125 + time.Sleep(500 * time.Millisecond) 126 126 + } 127 127 + <-ch 128 128 + }() 129 129 + } 113 130 114 114 - <-ch 115 115 - time.Sleep(1 * time.Second) 131 131 + time.Sleep(time.Duration(flushes) * time.Second) 116 132 117 133 err = process.Signal(os.Signal(syscall.SIGUSR1)) 118 134 require.NoError(t, err) ··· 126 142 } 127 143 128 144 func checkGStreamerLeaks(t *testing.T, expected int) { 145 145 + if os.Getenv(IGNORE_LEAKS) != "" { 146 146 + return 147 147 + } 129 148 leaks := getLeakCount(t) 130 149 if leaks > expected { 131 150 LeakReportMutex.Lock()

+2 -1

pkg/media/segment_conv.go

··· 20 20 "appsrc name=appsrc ! qtdemux name=demux", 21 21 "mpegtsmux name=mux ! appsink name=appsink sync=false", 22 22 "demux.video_0 ! h264parse ! video/x-h264,stream-format=byte-stream ! queue name=videoqueue", 23 23 - "demux.audio_0 ! opusparse ! queue name=audioqueue", 23 23 + "demux.audio_0 ! opusdec name=audioparse ! audioresample ! audiorate ! fdkaacenc name=audioenc ! queue name=audioqueue", 24 24 }, " ") 25 25 26 26 pipeline, err := gst.NewPipelineFromString(pipelineStr) ··· 506 506 onPadAdded := func(element *gst.Element, pad *gst.Pad) { 507 507 if pad.GetDirection() == gst.PadDirectionSource { 508 508 ok := pad.Link(videoParseSinkPad) 509 509 + defer func() { videoParseSinkPad = nil }() 509 510 if ok != gst.PadLinkOK { 510 511 log.Error(ctx, "failed to link video parse sink pad to video demux pad", "error", ok) 511 512 cancel()

+6 -4

pkg/media/segment_conv_test.go

··· 7 7 "testing" 8 8 "time" 9 9 10 10 - "github.com/go-gst/go-gst/gst" 11 10 "github.com/stretchr/testify/require" 12 11 "go.uber.org/goleak" 13 12 ) ··· 15 14 func TestMP4ToMPEGTS(t *testing.T) { 16 15 ignore := goleak.IgnoreCurrent() 17 16 defer goleak.VerifyNone(t, ignore) 18 18 - gst.Init(nil) 17 17 + before := getLeakCount(t) 18 18 + defer checkGStreamerLeaks(t, before) 19 19 20 20 // Open input file 21 21 inputFile, err := os.Open(getFixture("sample-segment.mp4")) ··· 67 67 func TestMPEGTSToMP4(t *testing.T) { 68 68 ignore := goleak.IgnoreCurrent() 69 69 defer goleak.VerifyNone(t, ignore) 70 70 - gst.Init(nil) 70 70 + before := getLeakCount(t) 71 71 + defer checkGStreamerLeaks(t, before) 71 72 72 73 // Open input file 73 74 inputFile, err := os.Open(getFixture("sample-segment.mpegts")) ··· 86 87 func TestMP4ToMPEGTSVideoMP4Audio(t *testing.T) { 87 88 ignore := goleak.IgnoreCurrent() 88 89 defer goleak.VerifyNone(t, ignore) 89 89 - gst.Init(nil) 90 90 + before := getLeakCount(t) 91 91 + defer checkGStreamerLeaks(t, before) 90 92 91 93 // Open input file 92 94 inputFile, err := os.Open(getFixture("5sec.mp4"))

+12 -2

pkg/media/segmenter.go

··· 18 18 "name": "signer", 19 19 "async-finalize": true, 20 20 "sink-factory": "appsink", 21 21 - "muxer-factory": "mp4mux", 21 21 + "muxer-factory": "qtmux", 22 22 "max-size-bytes": 1, 23 23 }) 24 24 if err != nil { ··· 63 63 EOSFunc: func(sink *app.Sink) { 64 64 resetTimer <- struct{}{} 65 65 now := time.Now().UnixMilli() 66 66 - bs, err := ms.SignMP4(ctx, bytes.NewReader(buf.Bytes()), now) 66 66 + bs := buf.Bytes() 67 67 + if mm.cli.SmearAudio { 68 68 + smearedBuf := &bytes.Buffer{} 69 69 + err := SmearAudioTimestamps(ctx, bytes.NewReader(buf.Bytes()), smearedBuf) 70 70 + if err != nil { 71 71 + log.Error(ctx, "error smearing audio timestamps", "error", err) 72 72 + return 73 73 + } 74 74 + bs = smearedBuf.Bytes() 75 75 + } 76 76 + bs, err := ms.SignMP4(ctx, bytes.NewReader(bs), now) 67 77 if err != nil { 68 78 log.Error(ctx, "error signing segment", "error", err) 69 79 return

+2 -2

pkg/media/thumbnail_test.go

··· 12 12 "go.uber.org/goleak" 13 13 "golang.org/x/sync/errgroup" 14 14 "stream.place/streamplace/pkg/gstinit" 15 15 - "stream.place/streamplace/pkg/log" 16 15 ) 17 16 18 17 func TestThumbnail(t *testing.T) { ··· 35 34 for i := 0; i < streamplaceTestCount; i++ { 36 35 g.Go(func() error { 37 36 thumbnail := bytes.Buffer{} 38 38 - thumbnailCtx := log.WithDebugValue(context.Background(), map[string]map[string]int{"function": {"Thumbnail": 9}}) 37 37 + thumbnailCtx := context.Background() 38 38 + // thumbnailCtx = log.WithDebugValue(ctx, map[string]map[string]int{"function": {"Thumbnail": 9}}) 39 39 err := Thumbnail(thumbnailCtx, bytes.NewReader(bs), &thumbnail) 40 40 if err != nil { 41 41 return err

+3 -3

pkg/media/webrtc_ingest.go

··· 42 42 43 43 pipelineSlice := []string{ 44 44 "multiqueue name=queue", 45 45 - "appsrc format=time is-live=true do-timestamp=true name=videosrc ! capsfilter caps=application/x-rtp ! rtph264depay ! capsfilter caps=video/x-h264,stream-format=byte-stream,alignment=nal ! h264parse ! h264timestamper ! identity ! queue.sink_0", 46 46 - "appsrc format=time is-live=true do-timestamp=true name=audiosrc ! capsfilter caps=application/x-rtp,media=audio,encoding-name=OPUS,payload=111 ! rtpopusdepay ! opusdec use-inband-fec=true ! audiorate ! opusenc ! queue.sink_1", 45 45 + "appsrc format=time is-live=true do-timestamp=true name=videosrc ! capsfilter caps=application/x-rtp ! rtph264depay ! capsfilter caps=video/x-h264,stream-format=byte-stream,alignment=nal ! h264parse disable-passthrough=true config-interval=-1 ! h264timestamper ! identity ! queue.sink_0", 46 46 + "appsrc format=time do-timestamp=true name=audiosrc ! capsfilter caps=application/x-rtp,media=audio,encoding-name=OPUS,payload=111 ! rtpopusdepay ! queue.sink_1", 47 47 } 48 48 49 49 pipeline, err := gst.NewPipelineFromString(strings.Join(pipelineSlice, "\n")) ··· 190 190 if track.Kind() == webrtc.RTPCodecTypeVideo { 191 191 // Send a PLI on an interval so that the publisher is pushing a keyframe every rtcpPLIInterval 192 192 go func() { 193 193 - ticker := time.NewTicker(time.Second * 5) 193 193 + ticker := time.NewTicker(time.Second * 1) 194 194 for { 195 195 select { 196 196 case <-ctx.Done():

test/fixtures/duration-mismatch-audio.mp4

This is a binary file and will not be displayed.

test/fixtures/duration-mismatch-video.mp4

This is a binary file and will not be displayed.

test/fixtures/duration-mismatch.mp4

This is a binary file and will not be displayed.