Refine the regression test tool, add missing files

ossrs · Mar 4, 2021 · 494c82a · 494c82a
1 parent bf3ac0d
commit 494c82a
Show file tree

Hide file tree

Showing 8 changed files with 1,333 additions and 3 deletions.
diff --git a/trunk/3rdparty/srs-bench/Makefile b/trunk/3rdparty/srs-bench/Makefile
@@ -17,7 +17,7 @@ bench: ./objs/srs_bench
 test: ./objs/srs_test
 
 ./objs/srs_test: .format.txt *.go rtc/*.go srs/*.go Makefile
-	go test ./srs -c -o ./objs/srs_test
+	go test ./srs -mod=vendor -c -o ./objs/srs_test
 
 help:
 	@echo "Usage: make [bench|test]"

diff --git a/trunk/3rdparty/srs-bench/README.md b/trunk/3rdparty/srs-bench/README.md
@@ -113,7 +113,7 @@ fi
 然后运行回归测试用例，如果只跑一次，可以直接运行：
 
 ```bash
-go test ./srs -v
+go test ./srs -mod=vendor -v
 ```
 
 也可以用make编译出重复使用的二进制：
@@ -125,7 +125,7 @@ make test && ./objs/srs_test -test.v
 可以给回归测试传参数，这样可以测试不同的序列，比如：
 
 ```bash
-go test ./srs -v -srs-server=127.0.0.1
+go test ./srs -mod=vendor -v -srs-server=127.0.0.1
 # Or
 make test && ./objs/srs_test -test.v -srs-server=127.0.0.1
 ```

diff --git a/trunk/3rdparty/srs-bench/srs/player.go b/trunk/3rdparty/srs-bench/srs/player.go
@@ -0,0 +1,262 @@
+package srs
+
+import (
+	"context"
+	"fmt"
+	"github.com/ossrs/go-oryx-lib/errors"
+	"github.com/ossrs/go-oryx-lib/logger"
+	"github.com/pion/interceptor"
+	"github.com/pion/rtcp"
+	"github.com/pion/sdp/v3"
+	"github.com/pion/webrtc/v3"
+	"github.com/pion/webrtc/v3/pkg/media"
+	"github.com/pion/webrtc/v3/pkg/media/h264writer"
+	"github.com/pion/webrtc/v3/pkg/media/ivfwriter"
+	"github.com/pion/webrtc/v3/pkg/media/oggwriter"
+	"strings"
+	"sync"
+	"time"
+)
+
+// @see https://github.com/pion/webrtc/blob/master/examples/save-to-disk/main.go
+func StartPlay(ctx context.Context, r, dumpAudio, dumpVideo string, enableAudioLevel, enableTWCC bool, pli int) error {
+	ctx = logger.WithContext(ctx)
+
+	logger.Tf(ctx, "Start play url=%v, audio=%v, video=%v, audio-level=%v, twcc=%v",
+		r, dumpAudio, dumpVideo, enableAudioLevel, enableTWCC)
+
+	// For audio-level.
+	webrtcNewPeerConnection := func(configuration webrtc.Configuration) (*webrtc.PeerConnection, error) {
+		m := &webrtc.MediaEngine{}
+		if err := m.RegisterDefaultCodecs(); err != nil {
+			return nil, err
+		}
+
+		for _, extension := range []string{sdp.SDESMidURI, sdp.SDESRTPStreamIDURI, sdp.TransportCCURI} {
+			if extension == sdp.TransportCCURI && !enableTWCC {
+				continue
+			}
+			if err := m.RegisterHeaderExtension(webrtc.RTPHeaderExtensionCapability{URI: extension}, webrtc.RTPCodecTypeVideo); err != nil {
+				return nil, err
+			}
+		}
+
+		// https://github.com/pion/ion/issues/130
+		// https://github.com/pion/ion-sfu/pull/373/files#diff-6f42c5ac6f8192dd03e5a17e9d109e90cb76b1a4a7973be6ce44a89ffd1b5d18R73
+		for _, extension := range []string{sdp.SDESMidURI, sdp.SDESRTPStreamIDURI, sdp.AudioLevelURI} {
+			if extension == sdp.AudioLevelURI && !enableAudioLevel {
+				continue
+			}
+			if err := m.RegisterHeaderExtension(webrtc.RTPHeaderExtensionCapability{URI: extension}, webrtc.RTPCodecTypeAudio); err != nil {
+				return nil, err
+			}
+		}
+
+		i := &interceptor.Registry{}
+		if err := webrtc.RegisterDefaultInterceptors(m, i); err != nil {
+			return nil, err
+		}
+
+		api := webrtc.NewAPI(webrtc.WithMediaEngine(m), webrtc.WithInterceptorRegistry(i))
+		return api.NewPeerConnection(configuration)
+	}
+
+	pc, err := webrtcNewPeerConnection(webrtc.Configuration{})
+	if err != nil {
+		return errors.Wrapf(err, "Create PC")
+	}
+	defer pc.Close()
+
+	pc.AddTransceiverFromKind(webrtc.RTPCodecTypeAudio, webrtc.RTPTransceiverInit{
+		Direction: webrtc.RTPTransceiverDirectionRecvonly,
+	})
+	pc.AddTransceiverFromKind(webrtc.RTPCodecTypeVideo, webrtc.RTPTransceiverInit{
+		Direction: webrtc.RTPTransceiverDirectionRecvonly,
+	})
+
+	offer, err := pc.CreateOffer(nil)
+	if err != nil {
+		return errors.Wrapf(err, "Create Offer")
+	}
+
+	if err := pc.SetLocalDescription(offer); err != nil {
+		return errors.Wrapf(err, "Set offer %v", offer)
+	}
+
+	answer, err := apiRtcRequest(ctx, "/rtc/v1/play", r, offer.SDP)
+	if err != nil {
+		return errors.Wrapf(err, "Api request offer=%v", offer.SDP)
+	}
+
+	if err := pc.SetRemoteDescription(webrtc.SessionDescription{
+		Type: webrtc.SDPTypeAnswer, SDP: answer,
+	}); err != nil {
+		return errors.Wrapf(err, "Set answer %v", answer)
+	}
+
+	var da media.Writer
+	var dv_vp8 media.Writer
+	var dv_h264 media.Writer
+	defer func() {
+		if da != nil {
+			da.Close()
+		}
+		if dv_vp8 != nil {
+			dv_vp8.Close()
+		}
+		if dv_h264 != nil {
+			dv_h264.Close()
+		}
+	}()
+
+	handleTrack := func(ctx context.Context, track *webrtc.TrackRemote, receiver *webrtc.RTPReceiver) error {
+		// Send a PLI on an interval so that the publisher is pushing a keyframe
+		go func() {
+			if track.Kind() == webrtc.RTPCodecTypeAudio {
+				return
+			}
+
+			if pli <= 0 {
+				return
+			}
+
+			for {
+				select {
+				case <-ctx.Done():
+					return
+				case <-time.After(time.Duration(pli) * time.Second):
+					_ = pc.WriteRTCP([]rtcp.Packet{&rtcp.PictureLossIndication{
+						MediaSSRC: uint32(track.SSRC()),
+					}})
+				}
+			}
+		}()
+
+		codec := track.Codec()
+
+		trackDesc := fmt.Sprintf("channels=%v", codec.Channels)
+		if track.Kind() == webrtc.RTPCodecTypeVideo {
+			trackDesc = fmt.Sprintf("fmtp=%v", codec.SDPFmtpLine)
+		}
+		if headers := receiver.GetParameters().HeaderExtensions; len(headers) > 0 {
+			trackDesc = fmt.Sprintf("%v, header=%v", trackDesc, headers)
+		}
+		logger.Tf(ctx, "Got track %v, pt=%v, tbn=%v, %v",
+			codec.MimeType, codec.PayloadType, codec.ClockRate, trackDesc)
+
+		if codec.MimeType == "audio/opus" {
+			if da == nil && dumpAudio != "" {
+				if da, err = oggwriter.New(dumpAudio, codec.ClockRate, codec.Channels); err != nil {
+					return errors.Wrapf(err, "New audio dumper")
+				}
+				logger.Tf(ctx, "Open ogg writer file=%v, tbn=%v, channels=%v",
+					dumpAudio, codec.ClockRate, codec.Channels)
+			}
+
+			if err = writeTrackToDisk(ctx, da, track); err != nil {
+				return errors.Wrapf(err, "Write audio disk")
+			}
+		} else if codec.MimeType == "video/VP8" {
+			if dumpVideo != "" && !strings.HasSuffix(dumpVideo, ".ivf") {
+				return errors.Errorf("%v should be .ivf for VP8", dumpVideo)
+			}
+
+			if dv_vp8 == nil && dumpVideo != "" {
+				if dv_vp8, err = ivfwriter.New(dumpVideo); err != nil {
+					return errors.Wrapf(err, "New video dumper")
+				}
+				logger.Tf(ctx, "Open ivf writer file=%v", dumpVideo)
+			}
+
+			if err = writeTrackToDisk(ctx, dv_vp8, track); err != nil {
+				return errors.Wrapf(err, "Write video disk")
+			}
+		} else if codec.MimeType == "video/H264" {
+			if dumpVideo != "" && !strings.HasSuffix(dumpVideo, ".h264") {
+				return errors.Errorf("%v should be .h264 for H264", dumpVideo)
+			}
+
+			if dv_h264 == nil && dumpVideo != "" {
+				if dv_h264, err = h264writer.New(dumpVideo); err != nil {
+					return errors.Wrapf(err, "New video dumper")
+				}
+				logger.Tf(ctx, "Open h264 writer file=%v", dumpVideo)
+			}
+
+			if err = writeTrackToDisk(ctx, dv_h264, track); err != nil {
+				return errors.Wrapf(err, "Write video disk")
+			}
+		} else {
+			logger.Wf(ctx, "Ignore track %v pt=%v", codec.MimeType, codec.PayloadType)
+		}
+		return nil
+	}
+
+	ctx, cancel := context.WithCancel(ctx)
+	pc.OnTrack(func(track *webrtc.TrackRemote, receiver *webrtc.RTPReceiver) {
+		err = handleTrack(ctx, track, receiver)
+		if err != nil {
+			codec := track.Codec()
+			err = errors.Wrapf(err, "Handle  track %v, pt=%v", codec.MimeType, codec.PayloadType)
+			cancel()
+		}
+	})
+
+	pc.OnICEConnectionStateChange(func(state webrtc.ICEConnectionState) {
+		logger.If(ctx, "ICE state %v", state)
+
+		if state == webrtc.ICEConnectionStateFailed || state == webrtc.ICEConnectionStateClosed {
+			if ctx.Err() != nil {
+				return
+			}
+
+			logger.Wf(ctx, "Close for ICE state %v", state)
+			cancel()
+		}
+	})
+
+	// Wait for event from context or tracks.
+	var wg sync.WaitGroup
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-time.After(5 * time.Second):
+				StatRTC.PeerConnection = pc.GetStats()
+			}
+		}
+	}()
+
+	wg.Wait()
+	return err
+}
+
+func writeTrackToDisk(ctx context.Context, w media.Writer, track *webrtc.TrackRemote) error {
+	for ctx.Err() == nil {
+		pkt, _, err := track.ReadRTP()
+		if err != nil {
+			if ctx.Err() != nil {
+				return nil
+			}
+			return errors.Wrapf(err, "Read RTP")
+		}
+
+		if w == nil {
+			continue
+		}
+
+		if err := w.WriteRTP(pkt); err != nil {
+			if len(pkt.Payload) <= 2 {
+				continue
+			}
+			logger.Wf(ctx, "Ignore write RTP %vB err %+v", len(pkt.Payload), err)
+		}
+	}
+
+	return ctx.Err()
+}