Merge branch v5.0.112 into develop

1. SRT: Fix srt to rtmp crash when sps or pps empty. v5.0.112 (#3323)
2. GB28181: Fix memory overlap for small packets. v5.0.111 (#3315)
3. FLV: Support set default has_av and disable guessing. v5.0.110 (#3311)
4. FLV: Drop packet if header flag is not matched. v5.0.109 (#3306)
5. FLV: Reset has_audio or has_video if only sequence header. (#3310)

.run/regression-test.run.xml

@@ -0,0 +1,7 @@
+<component name="ProjectRunConfigurationManager">
+  <configuration default="false" name="regression-test" type="CMakeRunConfiguration" factoryName="Application" PROGRAM_PARAMS="-c conf/regression-test-for-clion.conf" REDIRECT_INPUT="false" ELEVATE="false" USE_EXTERNAL_CONSOLE="false" WORKING_DIR="file://$CMakeCurrentBuildDir$/../../../" PASS_PARENT_ENVS_2="true" PROJECT_NAME="srs" TARGET_NAME="srs" CONFIG_NAME="Debug" RUN_TARGET_PROJECT_NAME="srs" RUN_TARGET_NAME="srs">
+    <method v="2">
+      <option name="com.jetbrains.cidr.execution.CidrBuildBeforeRunTaskProvider$BuildBeforeRunTask" enabled="true" />
+    </method>
+  </configuration>
+</component>

README.md

@@ -137,6 +137,7 @@ but some third-party libraries are distributed using their [own licenses](https:
 
 ## Releases
 
+* 2022-12-18, [Release v5.0-a2](https://github.com/ossrs/srs/releases/tag/v5.0-a2), v5.0-a2, 5.0 alpha2, v5.0.112, 161233 lines.
 * 2022-12-01, [Release v5.0-a1](https://github.com/ossrs/srs/releases/tag/v5.0-a1), v5.0-a1, 5.0 alpha1, v5.0.100, 160817 lines.
 * 2022-11-25, [Release v5.0-a0](https://github.com/ossrs/srs/releases/tag/v5.0-a0), v5.0-a0, 5.0 alpha0, v5.0.98, 159813 lines.
 * 2022-11-22, Release [v4.0-r4](https://github.com/ossrs/srs/releases/tag/v4.0-r4), v4.0-r4, 4.0 release4, v4.0.268, 145482 lines.

trunk/3rdparty/srs-bench/srs/rtc_test.go

@@ -2022,79 +2022,44 @@ func TestRtcPublish_FlvPlay(t *testing.T) {
 
 		select {
 		case <-ctx.Done():
+			return
 		case <-publishReady.Done():
-			var url string = "http://127.0.0.1:8080" + *srsStream + "-" + streamSuffix + ".flv"
-			logger.Tf(ctx, "Run play flv url=%v", url)
+		}
 
-			req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
-			if err != nil {
-				logger.Tf(ctx, "New request for flv %v failed, err=%v", url, err)
-				return
-			}
+		player := NewFLVPlayer()
+		defer player.Close()
 
-			client := http.Client{}
-			resp, err := client.Do(req)
-			if err != nil {
-				logger.Tf(ctx, "Http get flv %v failed, err=%v", url, err)
-				return
-			}
-
-			var f flv.Demuxer
-			if f, err = flv.NewDemuxer(resp.Body); err != nil {
-				logger.Tf(ctx, "Create flv demuxer for %v failed, err=%v", url, err)
-				return
+		r3 = func() error {
+			flvUrl := fmt.Sprintf("http://%v%v-%v.flv", *srsHttpServer, *srsStream, streamSuffix)
+			if err := player.Play(ctx, flvUrl); err != nil {
+				return err
 			}
-			defer f.Close()
 
-			var version uint8
+			var nnVideo, nnAudio int
 			var hasVideo, hasAudio bool
-			if version, hasVideo, hasAudio, err = f.ReadHeader(); err != nil {
-				logger.Tf(ctx, "Flv demuxer read header failed, err=%v", err)
-				return
+			player.onRecvHeader = func(ha, hv bool) error {
+				hasAudio, hasVideo = ha, hv
+				return nil
 			}
-
-			// Optional, user can check the header.
-			_ = version
-			_ = hasAudio
-			_ = hasVideo
-
-			var nnVideo, nnAudio int
-			var prevVideoTimestamp, prevAudioTimestamp int64
-
-			for {
-				var tagType flv.TagType
-				var tagSize, timestamp uint32
-				if tagType, tagSize, timestamp, err = f.ReadTagHeader(); err != nil {
-					logger.Tf(ctx, "Flv demuxer read tag header failed, err=%v", err)
-					return
-				}
-
-				var tag []byte
-				if tag, err = f.ReadTag(tagSize); err != nil {
-					logger.Tf(ctx, "Flv demuxer read tag failed, err=%v", err)
-					return
-				}
-
+			player.onRecvTag = func(tagType flv.TagType, size, timestamp uint32, tag []byte) error {
 				if tagType == flv.TagTypeAudio {
 					nnAudio++
-					prevAudioTimestamp = (int64)(timestamp)
 				} else if tagType == flv.TagTypeVideo {
 					nnVideo++
-					prevVideoTimestamp = (int64)(timestamp)
 				}
+				logger.Tf(ctx, "got %v tag, %v %vms %vB", nnVideo+nnAudio, tagType, timestamp, len(tag))
 
-				if nnAudio >= 10 && nnVideo >= 10 {
-					avDiff := prevVideoTimestamp - prevAudioTimestamp
-					// Check timestamp gap between video and audio, make sure audio timestamp align to video timestamp.
-					if avDiff <= 50 && avDiff >= -50 {
-						logger.Tf(ctx, "Flv recv %v audio, %v video, timestamp gap=%v", nnAudio, nnVideo, avDiff)
-						cancel()
-						break
-					}
+				if audioPacketsOK, videoPacketsOK := !hasAudio || nnAudio >= 10, !hasVideo || nnVideo >= 10; audioPacketsOK && videoPacketsOK {
+					logger.Tf(ctx, "Flv recv %v/%v audio, %v/%v video", hasAudio, nnAudio, hasVideo, nnVideo)
+					cancel()
 				}
-
-				_ = tag
+				return nil
 			}
-		}
+			if err := player.Consume(ctx); err != nil {
+				return err
+			}
+
+			return nil
+		}()
 	}()
 }

trunk/3rdparty/srs-bench/srs/rtmp_test.go

@@ -24,6 +24,7 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"github.com/pkg/errors"
 	"math/rand"
 	"os"
 	"sync"
@@ -393,3 +394,246 @@ func TestRtmpPublish_MultipleSequences_RtcPlay(t *testing.T) {
 		t.Errorf("err %+v", err)
 	}
 }
+
+func TestRtmpPublish_FlvPlay(t *testing.T) {
+	ctx := logger.WithContext(context.Background())
+	ctx, cancel := context.WithTimeout(ctx, time.Duration(*srsTimeout)*time.Millisecond)
+
+	var r0, r1 error
+	err := func() error {
+		publisher := NewRTMPPublisher()
+		defer publisher.Close()
+
+		player := NewFLVPlayer()
+		defer player.Close()
+
+		// Connect to RTMP URL.
+		streamSuffix := fmt.Sprintf("rtmp-regression-%v-%v", os.Getpid(), rand.Int())
+		rtmpUrl := fmt.Sprintf("rtmp://%v/live/%v", *srsServer, streamSuffix)
+		flvUrl := fmt.Sprintf("http://%v/live/%v.flv", *srsHttpServer, streamSuffix)
+
+		if err := publisher.Publish(ctx, rtmpUrl); err != nil {
+			return err
+		}
+
+		if err := player.Play(ctx, flvUrl); err != nil {
+			return err
+		}
+
+		// Check packets.
+		var wg sync.WaitGroup
+		defer wg.Wait()
+
+		publisherReady, publisherReadyCancel := context.WithCancel(context.Background())
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			time.Sleep(30 * time.Millisecond) // Wait for publisher to push sequence header.
+			publisherReadyCancel()
+		}()
+
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			<-publisherReady.Done()
+
+			var nnPackets int
+			player.onRecvHeader = func(hasAudio, hasVideo bool) error {
+				return nil
+			}
+			player.onRecvTag = func(tp flv.TagType, size, ts uint32, tag []byte) error {
+				logger.Tf(ctx, "got %v tag, %v %vms %vB", nnPackets, tp, ts, len(tag))
+				if nnPackets += 1; nnPackets > 50 {
+					cancel()
+				}
+				return nil
+			}
+			if r1 = player.Consume(ctx); r1 != nil {
+				cancel()
+			}
+		}()
+
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			publisher.onSendPacket = func(m *rtmp.Message) error {
+				time.Sleep(1 * time.Millisecond)
+				return nil
+			}
+			if r0 = publisher.Ingest(ctx, *srsPublishAvatar); r0 != nil {
+				cancel()
+			}
+		}()
+
+		return nil
+	}()
+	if err := filterTestError(ctx.Err(), err, r0, r1); err != nil {
+		t.Errorf("err %+v", err)
+	}
+}
+
+func TestRtmpPublish_FlvPlayNoAudio(t *testing.T) {
+	ctx := logger.WithContext(context.Background())
+	ctx, cancel := context.WithTimeout(ctx, time.Duration(*srsTimeout)*time.Millisecond)
+
+	var r0, r1 error
+	err := func() error {
+		publisher := NewRTMPPublisher()
+		defer publisher.Close()
+
+		// Set publisher to drop audio.
+		publisher.hasAudio = false
+
+		player := NewFLVPlayer()
+		defer player.Close()
+
+		// Connect to RTMP URL.
+		streamSuffix := fmt.Sprintf("rtmp-regression-%v-%v", os.Getpid(), rand.Int())
+		rtmpUrl := fmt.Sprintf("rtmp://%v/live/%v", *srsServer, streamSuffix)
+		flvUrl := fmt.Sprintf("http://%v/live/%v.flv", *srsHttpServer, streamSuffix)
+
+		if err := publisher.Publish(ctx, rtmpUrl); err != nil {
+			return err
+		}
+
+		if err := player.Play(ctx, flvUrl); err != nil {
+			return err
+		}
+
+		// Check packets.
+		var wg sync.WaitGroup
+		defer wg.Wait()
+
+		publisherReady, publisherReadyCancel := context.WithCancel(context.Background())
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			time.Sleep(30 * time.Millisecond) // Wait for publisher to push sequence header.
+			publisherReadyCancel()
+		}()
+
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			<-publisherReady.Done()
+
+			var nnPackets int
+			player.onRecvHeader = func(hasAudio, hasVideo bool) error {
+				return nil
+			}
+			player.onRecvTag = func(tp flv.TagType, size, ts uint32, tag []byte) error {
+				if tp == flv.TagTypeAudio {
+					return errors.New("should no audio")
+				}
+				logger.Tf(ctx, "got %v tag, %v %vms %vB", nnPackets, tp, ts, len(tag))
+				if nnPackets += 1; nnPackets > 50 {
+					cancel()
+				}
+				return nil
+			}
+			if r1 = player.Consume(ctx); r1 != nil {
+				cancel()
+			}
+		}()
+
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			publisher.onSendPacket = func(m *rtmp.Message) error {
+				time.Sleep(1 * time.Millisecond)
+				return nil
+			}
+			if r0 = publisher.Ingest(ctx, *srsPublishAvatar); r0 != nil {
+				cancel()
+			}
+		}()
+
+		return nil
+	}()
+	if err := filterTestError(ctx.Err(), err, r0, r1); err != nil {
+		t.Errorf("err %+v", err)
+	}
+}
+
+func TestRtmpPublish_FlvPlayNoVideo(t *testing.T) {
+	ctx := logger.WithContext(context.Background())
+	ctx, cancel := context.WithTimeout(ctx, time.Duration(*srsTimeout)*time.Millisecond)
+
+	var r0, r1 error
+	err := func() error {
+		publisher := NewRTMPPublisher()
+		defer publisher.Close()
+
+		// Set publisher to drop video.
+		publisher.hasVideo = false
+
+		player := NewFLVPlayer()
+		defer player.Close()
+
+		// Connect to RTMP URL.
+		streamSuffix := fmt.Sprintf("rtmp-regression-%v-%v", os.Getpid(), rand.Int())
+		rtmpUrl := fmt.Sprintf("rtmp://%v/live/%v", *srsServer, streamSuffix)
+		flvUrl := fmt.Sprintf("http://%v/live/%v.flv", *srsHttpServer, streamSuffix)
+
+		if err := publisher.Publish(ctx, rtmpUrl); err != nil {
+			return err
+		}
+
+		if err := player.Play(ctx, flvUrl); err != nil {
+			return err
+		}
+
+		// Check packets.
+		var wg sync.WaitGroup
+		defer wg.Wait()
+
+		publisherReady, publisherReadyCancel := context.WithCancel(context.Background())
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			time.Sleep(30 * time.Millisecond) // Wait for publisher to push sequence header.
+			publisherReadyCancel()
+		}()
+
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			<-publisherReady.Done()
+
+			var nnPackets int
+			player.onRecvHeader = func(hasAudio, hasVideo bool) error {
+				return nil
+			}
+			player.onRecvTag = func(tp flv.TagType, size, ts uint32, tag []byte) error {
+				if tp == flv.TagTypeVideo {
+					return errors.New("should no video")
+				}
+				logger.Tf(ctx, "got %v tag, %v %vms %vB", nnPackets, tp, ts, len(tag))
+				if nnPackets += 1; nnPackets > 50 {
+					cancel()
+				}
+				return nil
+			}
+			if r1 = player.Consume(ctx); r1 != nil {
+				cancel()
+			}
+		}()
+
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			publisher.onSendPacket = func(m *rtmp.Message) error {
+				time.Sleep(1 * time.Millisecond)
+				return nil
+			}
+			if r0 = publisher.Ingest(ctx, *srsPublishAvatar); r0 != nil {
+				cancel()
+			}
+		}()
+
+		return nil
+	}()
+	if err := filterTestError(ctx.Err(), err, r0, r1); err != nil {
+		t.Errorf("err %+v", err)
+	}
+}