diff --git a/trunk/3rdparty/srs-bench/blackbox/srt_test.go b/trunk/3rdparty/srs-bench/blackbox/srt_test.go new file mode 100644 index 000000000..ab524b697 --- /dev/null +++ b/trunk/3rdparty/srs-bench/blackbox/srt_test.go @@ -0,0 +1,116 @@ +// The MIT License (MIT) +// +// # Copyright (c) 2023 Winlin +// +// Permission is hereby granted, free of charge, to any person obtaining a copy of +// this software and associated documentation files (the "Software"), to deal in +// the Software without restriction, including without limitation the rights to +// use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of +// the Software, and to permit persons to whom the Software is furnished to do so, +// subject to the following conditions: +// +// The above copyright notice and this permission notice shall be included in all +// copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS +// FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR +// COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER +// IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN +// CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. +package blackbox + +import ( + "context" + "fmt" + "github.com/ossrs/go-oryx-lib/errors" + "github.com/ossrs/go-oryx-lib/logger" + "math/rand" + "os" + "path" + "sync" + "testing" + "time" +) + +func TestFast_SrtPublish_SrtPlay_Basic(t *testing.T) { + // This case is run in parallel. + t.Parallel() + + // Setup the max timeout for this case. + ctx, cancel := context.WithTimeout(logger.WithContext(context.Background()), time.Duration(*srsTimeout)*time.Millisecond) + defer cancel() + + // Check a set of errors. + var r0, r1, r2, r3, r4, r5, r6, r7 error + defer func(ctx context.Context) { + if err := filterTestError(ctx.Err(), r0, r1, r2, r3, r4, r5, r6, r7); err != nil { + t.Errorf("Fail for err %+v", err) + } else { + logger.Tf(ctx, "test done with err %+v", err) + } + }(ctx) + + var wg sync.WaitGroup + defer wg.Wait() + + // Start SRS server and wait for it to be ready. + svr := NewSRSServer(func(v *srsServer) { + v.envs = []string{ + "SRS_SRT_SERVER_ENABLED=on", + "SRS_VHOST_SRT_ENABLED=on", + } + }) + wg.Add(1) + go func() { + defer wg.Done() + r0 = svr.Run(ctx, cancel) + }() + + // Start FFmpeg to publish stream. + streamID := fmt.Sprintf("stream-%v-%v", os.Getpid(), rand.Int()) + streamURL := fmt.Sprintf("srt://localhost:%v?streamid=#!::r=live/%v,m=publish", svr.SRTPort(), streamID) + ffmpeg := NewFFmpeg(func(v *ffmpegClient) { + v.args = []string{ + "-stream_loop", "-1", "-re", "-i", *srsPublishAvatar, "-c", "copy", + "-pes_payload_size", "0", "-f", "mpegts", streamURL, + } + }) + wg.Add(1) + go func() { + defer wg.Done() + <-svr.ReadyCtx().Done() + r1 = ffmpeg.Run(ctx, cancel) + }() + + // Start FFprobe to detect and verify stream. + duration := time.Duration(*srsFFprobeDuration) * time.Millisecond + ffprobe := NewFFprobe(func(v *ffprobeClient) { + v.dvrFile = path.Join(svr.WorkDir(), "objs", fmt.Sprintf("srs-ffprobe-%v.ts", streamID)) + v.streamURL = fmt.Sprintf("srt://localhost:%v?streamid=#!::r=live/%v,m=request", svr.SRTPort(), streamID) + v.duration, v.timeout = duration, time.Duration(*srsFFprobeTimeout)*time.Millisecond + }) + wg.Add(1) + go func() { + defer wg.Done() + <-svr.ReadyCtx().Done() + r2 = ffprobe.Run(ctx, cancel) + }() + + // Fast quit for probe done. + select { + case <-ctx.Done(): + case <-ffprobe.ProbeDoneCtx().Done(): + defer cancel() + + str, m := ffprobe.Result() + if len(m.Streams) != 2 { + r3 = errors.Errorf("invalid streams=%v, %v, %v", len(m.Streams), m.String(), str) + } + + // Note that SRT score is low, so we only check duration. + if dv := m.Duration(); dv < duration { + r5 = errors.Errorf("short duration=%v < %v, %v, %v", dv, duration, m.String(), str) + } + } +} diff --git a/trunk/3rdparty/srs-bench/blackbox/util.go b/trunk/3rdparty/srs-bench/blackbox/util.go index 16afeabd8..c85508bf2 100644 --- a/trunk/3rdparty/srs-bench/blackbox/util.go +++ b/trunk/3rdparty/srs-bench/blackbox/util.go @@ -163,6 +163,8 @@ func (v *SRSPortAllocator) Allocate() int { if _, ok := v.ports.LoadOrStore(port, true); !ok { return port } + + time.Sleep(time.Duration(rand.Int()%1000) * time.Microsecond) } panic("Allocate port failed") @@ -395,6 +397,8 @@ type SRSServer interface { RTMPPort() int // HTTPPort is the HTTP stream port. HTTPPort() int + // SRTPort is the SRT UDP port. + SRTPort() int } // srsServer is a SRS server instance. @@ -421,6 +425,8 @@ type srsServer struct { apiListen int // HTTP server listen port. httpListen int + // SRT UDP server listen port. + srtListen int // The envs from user. envs []string @@ -446,12 +452,14 @@ func NewSRSServer(opts ...func(v *srsServer)) SRSServer { v.rtmpListen = allocator.Allocate() v.apiListen = allocator.Allocate() v.httpListen = allocator.Allocate() + v.srtListen = allocator.Allocate() // Do cleanup. v.process.onDispose = func(ctx context.Context, bs *backendService) error { allocator.Free(v.rtmpListen) allocator.Free(v.apiListen) allocator.Free(v.httpListen) + allocator.Free(v.srtListen) pidFile := path.Join(v.workDir, v.srsRelativePidFile) if _, err := os.Stat(pidFile); err == nil { @@ -491,6 +499,10 @@ func (v *srsServer) HTTPPort() int { return v.httpListen } +func (v *srsServer) SRTPort() int { + return v.srtListen +} + func (v *srsServer) WorkDir() string { return v.workDir } @@ -531,6 +543,8 @@ func (v *srsServer) Run(ctx context.Context, cancel context.CancelFunc) error { fmt.Sprintf("SRS_LISTEN=%v", v.rtmpListen), // Setup the HTTP sever listen port. fmt.Sprintf("SRS_HTTP_SERVER_LISTEN=%v", v.httpListen), + // Setup the SRT server listen port. + fmt.Sprintf("SRS_SRT_SERVER_LISTEN=%v", v.srtListen), } // Rewrite envs by case. if v.envs != nil {