diff --git a/.run/srs-ingest.run.xml b/.run/srs-ingest.run.xml new file mode 100644 index 000000000..1ef1dba30 --- /dev/null +++ b/.run/srs-ingest.run.xml @@ -0,0 +1,7 @@ + + + + + + \ No newline at end of file diff --git a/trunk/3rdparty/srs-bench/srs/rtc_test.go b/trunk/3rdparty/srs-bench/srs/rtc_test.go index 941b1ccbe..4d2360120 100644 --- a/trunk/3rdparty/srs-bench/srs/rtc_test.go +++ b/trunk/3rdparty/srs-bench/srs/rtc_test.go @@ -25,11 +25,13 @@ import ( "encoding/json" "fmt" "github.com/pion/transport/vnet" + "github.com/pion/webrtc/v3" "io" "io/ioutil" "math/rand" "net/http" "os" + "strings" "sync" "testing" "time" @@ -62,7 +64,7 @@ func TestMain(m *testing.M) { func TestPR2483_RtcStatApi_PublisherOnly(t *testing.T) { if err := filterTestError(func() error { streamSuffix := fmt.Sprintf("publish-only-%v-%v", os.Getpid(), rand.Int()) - p, err := newTestPublisher(createApiForPublisher, func(p *testPublisher) error { + p, err := newTestPublisher(registerDefaultCodecs, func(p *testPublisher) error { p.streamSuffix = streamSuffix return nil }) @@ -96,11 +98,346 @@ func TestPR2483_RtcStatApi_PublisherOnly(t *testing.T) { } } +// Veirfy https://github.com/ossrs/srs/issues/2371 +func TestBugfix2371_PublishWithNack(t *testing.T) { + ctx, cancel := context.WithTimeout(logger.WithContext(context.Background()), time.Duration(*srsTimeout)*time.Millisecond) + if err := filterTestError(func() error { + streamSuffix := fmt.Sprintf("bugfix-2371-%v-%v", os.Getpid(), rand.Int()) + p, err := newTestPublisher(registerMiniCodecs, func(p *testPublisher) error { + p.streamSuffix = streamSuffix + p.onOffer = func(s *webrtc.SessionDescription) error { + if n := strings.Count(s.SDP, "nack"); n != 2 { + return errors.Errorf("invalid %v nack", n) + } + return nil + } + p.onAnswer = func(s *webrtc.SessionDescription) error { + if n := strings.Count(s.SDP, "nack"); n != 2 { + return errors.Errorf("invalid %v nack", n) + } + cancel() + return nil + } + return nil + }) + if err != nil { + return err + } + defer p.Close() + + if err := p.Setup(*srsVnetClientIP); err != nil { + return err + } + + return p.Run(ctx, cancel) + }()); err != nil { + t.Errorf("err %+v", err) + } +} + +// Veirfy https://github.com/ossrs/srs/issues/2371 +func TestBugfix2371_PublishWithoutNack(t *testing.T) { + ctx, cancel := context.WithTimeout(logger.WithContext(context.Background()), time.Duration(*srsTimeout)*time.Millisecond) + if err := filterTestError(func() error { + streamSuffix := fmt.Sprintf("bugfix-2371-%v-%v", os.Getpid(), rand.Int()) + p, err := newTestPublisher(registerMiniCodecsWithoutNack, func(p *testPublisher) error { + p.streamSuffix = streamSuffix + p.onOffer = func(s *webrtc.SessionDescription) error { + if n := strings.Count(s.SDP, "nack"); n != 0 { + return errors.Errorf("invalid %v nack", n) + } + return nil + } + p.onAnswer = func(s *webrtc.SessionDescription) error { + if n := strings.Count(s.SDP, "nack"); n != 0 { + return errors.Errorf("invalid %v nack", n) + } + cancel() + return nil + } + return nil + }) + if err != nil { + return err + } + defer p.Close() + + if err := p.Setup(*srsVnetClientIP); err != nil { + return err + } + + return p.Run(ctx, cancel) + }()); err != nil { + t.Errorf("err %+v", err) + } +} + +// Veirfy https://github.com/ossrs/srs/issues/2371 +func TestBugfix2371_PlayWithNack(t *testing.T) { + ctx := logger.WithContext(context.Background()) + ctx, cancel := context.WithTimeout(ctx, time.Duration(*srsTimeout)*time.Millisecond) + + var r0, r1, r2, r3 error + defer func(ctx context.Context) { + if err := filterTestError(ctx.Err(), r0, r1, r2, r3); err != nil { + t.Errorf("Fail for err %+v", err) + } else { + logger.Tf(ctx, "test done with err %+v", err) + } + }(ctx) + + var resources []io.Closer + defer func() { + for _, resource := range resources { + _ = resource.Close() + } + }() + + var wg sync.WaitGroup + defer wg.Wait() + + // The event notify. + var thePublisher *testPublisher + var thePlayer *testPlayer + + mainReady, mainReadyCancel := context.WithCancel(context.Background()) + publishReady, publishReadyCancel := context.WithCancel(context.Background()) + + // Objects init. + wg.Add(1) + go func() { + defer wg.Done() + defer cancel() + + doInit := func() (err error) { + streamSuffix := fmt.Sprintf("basic-publish-play-%v-%v", os.Getpid(), rand.Int()) + + // Initialize player with private api. + if thePlayer, err = newTestPlayer(registerMiniCodecs, func(play *testPlayer) error { + play.streamSuffix = streamSuffix + play.onOffer = func(s *webrtc.SessionDescription) error { + if n := strings.Count(s.SDP, "nack"); n != 2 { + return errors.Errorf("invalid %v nack", n) + } + return nil + } + play.onAnswer = func(s *webrtc.SessionDescription) error { + if n := strings.Count(s.SDP, "nack"); n != 2 { + return errors.Errorf("invalid %v nack", n) + } + cancel() + return nil + } + resources = append(resources, play) + return play.Setup(*srsVnetClientIP) + }); err != nil { + return err + } + + // Initialize publisher with private api. + if thePublisher, err = newTestPublisher(registerMiniCodecs, func(pub *testPublisher) error { + pub.streamSuffix = streamSuffix + pub.iceReadyCancel = publishReadyCancel + resources = append(resources, pub) + return pub.Setup(*srsVnetClientIP) + }); err != nil { + return err + } + + // Init done. + mainReadyCancel() + + <-ctx.Done() + return nil + } + + if err := doInit(); err != nil { + r1 = err + } + }() + + // Run publisher. + wg.Add(1) + go func() { + defer wg.Done() + defer cancel() + + select { + case <-ctx.Done(): + case <-mainReady.Done(): + r2 = thePublisher.Run(logger.WithContext(ctx), cancel) + logger.Tf(ctx, "pub done") + } + }() + + // Run player. + wg.Add(1) + go func() { + defer wg.Done() + defer cancel() + + select { + case <-ctx.Done(): + case <-publishReady.Done(): + r3 = thePlayer.Run(logger.WithContext(ctx), cancel) + logger.Tf(ctx, "play done") + } + }() +} + +// Veirfy https://github.com/ossrs/srs/issues/2371 +func TestBugfix2371_PlayWithoutNack(t *testing.T) { + ctx := logger.WithContext(context.Background()) + ctx, cancel := context.WithTimeout(ctx, time.Duration(*srsTimeout)*time.Millisecond) + + var r0, r1, r2, r3 error + defer func(ctx context.Context) { + if err := filterTestError(ctx.Err(), r0, r1, r2, r3); err != nil { + t.Errorf("Fail for err %+v", err) + } else { + logger.Tf(ctx, "test done with err %+v", err) + } + }(ctx) + + var resources []io.Closer + defer func() { + for _, resource := range resources { + _ = resource.Close() + } + }() + + var wg sync.WaitGroup + defer wg.Wait() + + // The event notify. + var thePublisher *testPublisher + var thePlayer *testPlayer + + mainReady, mainReadyCancel := context.WithCancel(context.Background()) + publishReady, publishReadyCancel := context.WithCancel(context.Background()) + + // Objects init. + wg.Add(1) + go func() { + defer wg.Done() + defer cancel() + + doInit := func() (err error) { + streamSuffix := fmt.Sprintf("basic-publish-play-%v-%v", os.Getpid(), rand.Int()) + + // Initialize player with private api. + if thePlayer, err = newTestPlayer(registerMiniCodecsWithoutNack, func(play *testPlayer) error { + play.streamSuffix = streamSuffix + play.onOffer = func(s *webrtc.SessionDescription) error { + if n := strings.Count(s.SDP, "nack"); n != 0 { + return errors.Errorf("invalid %v nack", n) + } + return nil + } + play.onAnswer = func(s *webrtc.SessionDescription) error { + if n := strings.Count(s.SDP, "nack"); n != 0 { + return errors.Errorf("invalid %v nack", n) + } + cancel() + return nil + } + resources = append(resources, play) + return play.Setup(*srsVnetClientIP) + }); err != nil { + return err + } + + // Initialize publisher with private api. + if thePublisher, err = newTestPublisher(registerMiniCodecs, func(pub *testPublisher) error { + pub.streamSuffix = streamSuffix + pub.iceReadyCancel = publishReadyCancel + resources = append(resources, pub) + return pub.Setup(*srsVnetClientIP) + }); err != nil { + return err + } + + // Init done. + mainReadyCancel() + + <-ctx.Done() + return nil + } + + if err := doInit(); err != nil { + r1 = err + } + }() + + // Run publisher. + wg.Add(1) + go func() { + defer wg.Done() + defer cancel() + + select { + case <-ctx.Done(): + case <-mainReady.Done(): + r2 = thePublisher.Run(logger.WithContext(ctx), cancel) + logger.Tf(ctx, "pub done") + } + }() + + // Run player. + wg.Add(1) + go func() { + defer wg.Done() + defer cancel() + + select { + case <-ctx.Done(): + case <-publishReady.Done(): + r3 = thePlayer.Run(logger.WithContext(ctx), cancel) + logger.Tf(ctx, "play done") + } + }() +} + +// Veirfy https://github.com/ossrs/srs/issues/2371 +func TestBugfix2371_RTMP2RTC_PlayWithNack(t *testing.T) { + if err := filterTestError(func() error { + ctx, cancel := context.WithTimeout(logger.WithContext(context.Background()), time.Duration(*srsTimeout)*time.Millisecond) + p, err := newTestPlayer(registerMiniCodecs, func(play *testPlayer) error { + play.onOffer = func(s *webrtc.SessionDescription) error { + if n := strings.Count(s.SDP, "nack"); n != 2 { + return errors.Errorf("invalid %v nack", n) + } + return nil + } + play.onAnswer = func(s *webrtc.SessionDescription) error { + if n := strings.Count(s.SDP, "nack"); n != 2 { + return errors.Errorf("invalid %v nack", n) + } + cancel() + return nil + } + return nil + }) + if err != nil { + return err + } + defer p.Close() + + if err := p.Setup(*srsVnetClientIP); err != nil { + return err + } + + return p.Run(ctx, cancel) + }()); err != nil { + t.Errorf("err %+v", err) + } +} + // Basic use scenario, publish a stream. func TestRtcBasic_PublishOnly(t *testing.T) { if err := filterTestError(func() error { streamSuffix := fmt.Sprintf("publish-only-%v-%v", os.Getpid(), rand.Int()) - p, err := newTestPublisher(createApiForPublisher, func(p *testPublisher) error { + p, err := newTestPublisher(registerDefaultCodecs, func(p *testPublisher) error { p.streamSuffix = streamSuffix return nil }) @@ -187,7 +524,7 @@ func TestRtcBasic_PublishPlay(t *testing.T) { streamSuffix := fmt.Sprintf("basic-publish-play-%v-%v", os.Getpid(), rand.Int()) // Initialize player with private api. - if thePlayer, err = newTestPlayer(createApiForPlayer, func(play *testPlayer) error { + if thePlayer, err = newTestPlayer(registerDefaultCodecs, func(play *testPlayer) error { play.streamSuffix = streamSuffix resources = append(resources, play) @@ -221,7 +558,7 @@ func TestRtcBasic_PublishPlay(t *testing.T) { } // Initialize publisher with private api. - if thePublisher, err = newTestPublisher(createApiForPublisher, func(pub *testPublisher) error { + if thePublisher, err = newTestPublisher(registerDefaultCodecs, func(pub *testPublisher) error { pub.streamSuffix = streamSuffix pub.iceReadyCancel = publishReadyCancel resources = append(resources, pub) @@ -342,7 +679,7 @@ func TestRtcBasic_Republish(t *testing.T) { streamSuffix := fmt.Sprintf("basic-publish-play-%v-%v", os.Getpid(), rand.Int()) // Initialize player with private api. - if thePlayer, err = newTestPlayer(createApiForPlayer, func(play *testPlayer) error { + if thePlayer, err = newTestPlayer(registerDefaultCodecs, func(play *testPlayer) error { play.streamSuffix = streamSuffix resources = append(resources, play) @@ -368,7 +705,7 @@ func TestRtcBasic_Republish(t *testing.T) { } // Initialize publisher with private api. - if thePublisher, err = newTestPublisher(createApiForPublisher, func(pub *testPublisher) error { + if thePublisher, err = newTestPublisher(registerDefaultCodecs, func(pub *testPublisher) error { pub.streamSuffix = streamSuffix pub.iceReadyCancel = publishReadyCancel resources = append(resources, pub) @@ -391,7 +728,7 @@ func TestRtcBasic_Republish(t *testing.T) { } // Initialize re-publisher with private api. - if theRepublisher, err = newTestPublisher(createApiForPublisher, func(pub *testPublisher) error { + if theRepublisher, err = newTestPublisher(registerDefaultCodecs, func(pub *testPublisher) error { pub.streamSuffix = streamSuffix pub.iceReadyCancel = republishReadyCancel resources = append(resources, pub) @@ -457,7 +794,7 @@ func TestRtcBasic_Republish(t *testing.T) { func TestRtcDTLS_ClientActive_Default(t *testing.T) { if err := filterTestError(func() error { streamSuffix := fmt.Sprintf("dtls-passive-no-arq-%v-%v", os.Getpid(), rand.Int()) - p, err := newTestPublisher(createApiForPublisher, func(p *testPublisher) error { + p, err := newTestPublisher(registerDefaultCodecs, func(p *testPublisher) error { p.streamSuffix = streamSuffix p.onOffer = testUtilSetupActive return nil @@ -512,7 +849,7 @@ func TestRtcDTLS_ClientActive_Default(t *testing.T) { func TestRtcDTLS_ClientPassive_Default(t *testing.T) { if err := filterTestError(func() error { streamSuffix := fmt.Sprintf("dtls-active-no-arq-%v-%v", os.Getpid(), rand.Int()) - p, err := newTestPublisher(createApiForPublisher, func(p *testPublisher) error { + p, err := newTestPublisher(registerDefaultCodecs, func(p *testPublisher) error { p.streamSuffix = streamSuffix p.onOffer = testUtilSetupPassive return nil @@ -564,7 +901,7 @@ func TestRtcDTLS_ClientPassive_Default(t *testing.T) { func TestRtcDTLS_ClientActive_Duplicated_Alert(t *testing.T) { if err := filterTestError(func() error { streamSuffix := fmt.Sprintf("dtls-active-no-arq-%v-%v", os.Getpid(), rand.Int()) - p, err := newTestPublisher(createApiForPublisher, func(p *testPublisher) error { + p, err := newTestPublisher(registerDefaultCodecs, func(p *testPublisher) error { p.streamSuffix = streamSuffix p.onOffer = testUtilSetupActive return nil @@ -623,7 +960,7 @@ func TestRtcDTLS_ClientActive_Duplicated_Alert(t *testing.T) { func TestRtcDTLS_ClientPassive_Duplicated_Alert(t *testing.T) { if err := filterTestError(func() error { streamSuffix := fmt.Sprintf("dtls-active-no-arq-%v-%v", os.Getpid(), rand.Int()) - p, err := newTestPublisher(createApiForPublisher, func(p *testPublisher) error { + p, err := newTestPublisher(registerDefaultCodecs, func(p *testPublisher) error { p.streamSuffix = streamSuffix p.onOffer = testUtilSetupPassive return nil @@ -689,7 +1026,7 @@ func TestRtcDTLS_ClientActive_ARQ_ClientHello_ByDropped_ClientHello(t *testing.T var r0 error err := func() error { streamSuffix := fmt.Sprintf("dtls-active-arq-client-hello-%v-%v", os.Getpid(), rand.Int()) - p, err := newTestPublisher(createApiForPublisher, func(p *testPublisher) error { + p, err := newTestPublisher(registerDefaultCodecs, func(p *testPublisher) error { p.streamSuffix = streamSuffix p.onOffer = testUtilSetupActive return nil @@ -767,7 +1104,7 @@ func TestRtcDTLS_ClientPassive_ARQ_ClientHello_ByDropped_ClientHello(t *testing. var r0 error err := func() error { streamSuffix := fmt.Sprintf("dtls-passive-arq-client-hello-%v-%v", os.Getpid(), rand.Int()) - p, err := newTestPublisher(createApiForPublisher, func(p *testPublisher) error { + p, err := newTestPublisher(registerDefaultCodecs, func(p *testPublisher) error { p.streamSuffix = streamSuffix p.onOffer = testUtilSetupPassive return nil @@ -844,7 +1181,7 @@ func TestRtcDTLS_ClientActive_ARQ_ClientHello_ByDropped_ServerHello(t *testing.T var r0, r1 error err := func() error { streamSuffix := fmt.Sprintf("dtls-active-arq-client-hello-%v-%v", os.Getpid(), rand.Int()) - p, err := newTestPublisher(createApiForPublisher, func(p *testPublisher) error { + p, err := newTestPublisher(registerDefaultCodecs, func(p *testPublisher) error { p.streamSuffix = streamSuffix p.onOffer = testUtilSetupActive return nil @@ -932,7 +1269,7 @@ func TestRtcDTLS_ClientPassive_ARQ_ClientHello_ByDropped_ServerHello(t *testing. var r0, r1 error err := func() error { streamSuffix := fmt.Sprintf("dtls-passive-arq-client-hello-%v-%v", os.Getpid(), rand.Int()) - p, err := newTestPublisher(createApiForPublisher, func(p *testPublisher) error { + p, err := newTestPublisher(registerDefaultCodecs, func(p *testPublisher) error { p.streamSuffix = streamSuffix p.onOffer = testUtilSetupPassive return nil @@ -1017,7 +1354,7 @@ func TestRtcDTLS_ClientActive_ARQ_Certificate_ByDropped_Certificate(t *testing.T var r0 error err := func() error { streamSuffix := fmt.Sprintf("dtls-active-arq-certificate-%v-%v", os.Getpid(), rand.Int()) - p, err := newTestPublisher(createApiForPublisher, func(p *testPublisher) error { + p, err := newTestPublisher(registerDefaultCodecs, func(p *testPublisher) error { p.streamSuffix = streamSuffix p.onOffer = testUtilSetupActive return nil @@ -1094,7 +1431,7 @@ func TestRtcDTLS_ClientPassive_ARQ_Certificate_ByDropped_Certificate(t *testing. var r0 error err := func() error { streamSuffix := fmt.Sprintf("dtls-passive-arq-certificate-%v-%v", os.Getpid(), rand.Int()) - p, err := newTestPublisher(createApiForPublisher, func(p *testPublisher) error { + p, err := newTestPublisher(registerDefaultCodecs, func(p *testPublisher) error { p.streamSuffix = streamSuffix p.onOffer = testUtilSetupPassive return nil @@ -1171,7 +1508,7 @@ func TestRtcDTLS_ClientActive_ARQ_Certificate_ByDropped_ChangeCipherSpec(t *test var r0, r1 error err := func() error { streamSuffix := fmt.Sprintf("dtls-active-arq-certificate-%v-%v", os.Getpid(), rand.Int()) - p, err := newTestPublisher(createApiForPublisher, func(p *testPublisher) error { + p, err := newTestPublisher(registerDefaultCodecs, func(p *testPublisher) error { p.streamSuffix = streamSuffix p.onOffer = testUtilSetupActive return nil @@ -1257,7 +1594,7 @@ func TestRtcDTLS_ClientPassive_ARQ_Certificate_ByDropped_ChangeCipherSpec(t *tes var r0, r1 error err := func() error { streamSuffix := fmt.Sprintf("dtls-passive-arq-certificate-%v-%v", os.Getpid(), rand.Int()) - p, err := newTestPublisher(createApiForPublisher, func(p *testPublisher) error { + p, err := newTestPublisher(registerDefaultCodecs, func(p *testPublisher) error { p.streamSuffix = streamSuffix p.onOffer = testUtilSetupPassive return nil @@ -1334,7 +1671,7 @@ func TestRtcDTLS_ClientPassive_ARQ_Certificate_ByDropped_ChangeCipherSpec(t *tes func TestRtcDTLS_ClientPassive_ARQ_DropAllAfter_ClientHello(t *testing.T) { if err := filterTestError(func() error { streamSuffix := fmt.Sprintf("dtls-passive-no-arq-%v-%v", os.Getpid(), rand.Int()) - p, err := newTestPublisher(createApiForPublisher, func(p *testPublisher) error { + p, err := newTestPublisher(registerDefaultCodecs, func(p *testPublisher) error { p.streamSuffix = streamSuffix p.onOffer = testUtilSetupPassive return nil @@ -1387,7 +1724,7 @@ func TestRtcDTLS_ClientPassive_ARQ_DropAllAfter_ClientHello(t *testing.T) { func TestRtcDTLS_ClientPassive_ARQ_DropAllAfter_ServerHello(t *testing.T) { if err := filterTestError(func() error { streamSuffix := fmt.Sprintf("dtls-passive-no-arq-%v-%v", os.Getpid(), rand.Int()) - p, err := newTestPublisher(createApiForPublisher, func(p *testPublisher) error { + p, err := newTestPublisher(registerDefaultCodecs, func(p *testPublisher) error { p.streamSuffix = streamSuffix p.onOffer = testUtilSetupPassive return nil @@ -1440,7 +1777,7 @@ func TestRtcDTLS_ClientPassive_ARQ_DropAllAfter_ServerHello(t *testing.T) { func TestRtcDTLS_ClientPassive_ARQ_DropAllAfter_Certificate(t *testing.T) { if err := filterTestError(func() error { streamSuffix := fmt.Sprintf("dtls-passive-no-arq-%v-%v", os.Getpid(), rand.Int()) - p, err := newTestPublisher(createApiForPublisher, func(p *testPublisher) error { + p, err := newTestPublisher(registerDefaultCodecs, func(p *testPublisher) error { p.streamSuffix = streamSuffix p.onOffer = testUtilSetupPassive return nil @@ -1493,7 +1830,7 @@ func TestRtcDTLS_ClientPassive_ARQ_DropAllAfter_Certificate(t *testing.T) { func TestRtcDTLS_ClientPassive_ARQ_DropAllAfter_ChangeCipherSpec(t *testing.T) { if err := filterTestError(func() error { streamSuffix := fmt.Sprintf("dtls-passive-no-arq-%v-%v", os.Getpid(), rand.Int()) - p, err := newTestPublisher(createApiForPublisher, func(p *testPublisher) error { + p, err := newTestPublisher(registerDefaultCodecs, func(p *testPublisher) error { p.streamSuffix = streamSuffix p.onOffer = testUtilSetupPassive return nil @@ -1547,7 +1884,7 @@ func TestRtcDTLS_ClientPassive_ARQ_DropAllAfter_ChangeCipherSpec(t *testing.T) { func TestRtcDTLS_ClientPassive_ARQ_VeryBadNetwork(t *testing.T) { if err := filterTestError(func() error { streamSuffix := fmt.Sprintf("dtls-passive-no-arq-%v-%v", os.Getpid(), rand.Int()) - p, err := newTestPublisher(createApiForPublisher, func(p *testPublisher) error { + p, err := newTestPublisher(registerDefaultCodecs, func(p *testPublisher) error { p.streamSuffix = streamSuffix p.onOffer = testUtilSetupPassive return nil @@ -1625,7 +1962,7 @@ func TestRtcDTLS_ClientPassive_ARQ_Certificate_After_ClientHello(t *testing.T) { var r0 error err := func() error { streamSuffix := fmt.Sprintf("dtls-passive-no-arq-%v-%v", os.Getpid(), rand.Int()) - p, err := newTestPublisher(createApiForPublisher, func(p *testPublisher) error { + p, err := newTestPublisher(registerDefaultCodecs, func(p *testPublisher) error { p.streamSuffix = streamSuffix p.onOffer = testUtilSetupPassive return nil diff --git a/trunk/3rdparty/srs-bench/srs/util.go b/trunk/3rdparty/srs-bench/srs/util.go index ed8692523..eb88e8c95 100644 --- a/trunk/3rdparty/srs-bench/srs/util.go +++ b/trunk/3rdparty/srs-bench/srs/util.go @@ -59,6 +59,7 @@ var srsDTLSDropPackets *int var srsSchema string var srsServer *string var srsStream *string +var srsLiveStream *string var srsPublishAudio *string var srsPublishVideo *string var srsVnetClientIP *string @@ -68,7 +69,8 @@ func prepareTest() error { srsHttps = flag.Bool("srs-https", false, "Whther connect to HTTPS-API") srsServer = flag.String("srs-server", "127.0.0.1", "The RTC server to connect to") - srsStream = flag.String("srs-stream", "/rtc/regression", "The RTC stream to play") + srsStream = flag.String("srs-stream", "/rtc/regression", "The RTC app/stream to play") + srsLiveStream = flag.String("srs-live-stream", "/live/livestream", "The LIVE app/stream to play") srsLog = flag.Bool("srs-log", false, "Whether enable the detail log") srsTimeout = flag.Int("srs-timeout", 5000, "For each case, the timeout in ms") srsPlayPLI = flag.Int("srs-play-pli", 5000, "The PLI interval in seconds for player.") @@ -234,6 +236,11 @@ func (v *wallClock) Tick(d time.Duration) time.Duration { return 0 } +// Do nothing for SDP. +func testUtilPassBy(s *webrtc.SessionDescription) error { + return nil +} + // Set to active, as DTLS client, to start ClientHello. func testUtilSetupActive(s *webrtc.SessionDescription) error { if strings.Contains(s.SDP, "setup:passive") { @@ -548,6 +555,7 @@ func (v *dtlsRecord) Unmarshal(b []byte) error { return nil } +// The func to setup testWebRTCAPI type testWebRTCAPIOptionFunc func(api *testWebRTCAPI) type testWebRTCAPI struct { @@ -566,25 +574,88 @@ type testWebRTCAPI struct { proxy *vnet_proxy.UDPProxy } -func newTestWebRTCAPI(options ...testWebRTCAPIOptionFunc) (*testWebRTCAPI, error) { - v := &testWebRTCAPI{} +// The func to initialize testWebRTCAPI +type testWebRTCAPIInitFunc func(api *testWebRTCAPI) error + +// Implements interface testWebRTCAPIInitFunc to init testWebRTCAPI +func registerDefaultCodecs(api *testWebRTCAPI) error { + v := api - v.mediaEngine = &webrtc.MediaEngine{} if err := v.mediaEngine.RegisterDefaultCodecs(); err != nil { - return nil, err + return err } - v.registry = &interceptor.Registry{} if err := webrtc.RegisterDefaultInterceptors(v.mediaEngine, v.registry); err != nil { - return nil, err + return err } - for _, setup := range options { - setup(v) + return nil +} + +// Implements interface testWebRTCAPIInitFunc to init testWebRTCAPI +func registerMiniCodecs(api *testWebRTCAPI) error { + v := api + + if err := v.mediaEngine.RegisterCodec(webrtc.RTPCodecParameters{ + RTPCodecCapability: webrtc.RTPCodecCapability{webrtc.MimeTypeOpus, 48000, 2, "minptime=10;useinbandfec=1", nil}, + PayloadType: 111, + }, webrtc.RTPCodecTypeAudio); err != nil { + return err } + videoRTCPFeedback := []webrtc.RTCPFeedback{{"goog-remb", ""}, {"ccm", "fir"}, {"nack", ""}, {"nack", "pli"}} + if err := v.mediaEngine.RegisterCodec(webrtc.RTPCodecParameters{ + RTPCodecCapability: webrtc.RTPCodecCapability{webrtc.MimeTypeH264, 90000, 0, "level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=42e01f", videoRTCPFeedback}, + PayloadType: 108, + }, webrtc.RTPCodecTypeVideo); err != nil { + return err + } + + // Interceptors for NACK??? @see webrtc.ConfigureNack(v.mediaEngine, v.registry) + return nil +} + +// Implements interface testWebRTCAPIInitFunc to init testWebRTCAPI +func registerMiniCodecsWithoutNack(api *testWebRTCAPI) error { + v := api + + if err := v.mediaEngine.RegisterCodec(webrtc.RTPCodecParameters{ + RTPCodecCapability: webrtc.RTPCodecCapability{webrtc.MimeTypeOpus, 48000, 2, "minptime=10;useinbandfec=1", nil}, + PayloadType: 111, + }, webrtc.RTPCodecTypeAudio); err != nil { + return err + } + + videoRTCPFeedback := []webrtc.RTCPFeedback{{"goog-remb", ""}, {"ccm", "fir"}} + if err := v.mediaEngine.RegisterCodec(webrtc.RTPCodecParameters{ + RTPCodecCapability: webrtc.RTPCodecCapability{webrtc.MimeTypeH264, 90000, 0, "level-asymmetry-allowed=1;packetization-mode=0;profile-level-id=42e01f", videoRTCPFeedback}, + PayloadType: 108, + }, webrtc.RTPCodecTypeVideo); err != nil { + return err + } + + // Interceptors for NACK??? @see webrtc.ConfigureNack(v.mediaEngine, v.registry) + return nil +} + +func newTestWebRTCAPI(inits ...testWebRTCAPIInitFunc) (*testWebRTCAPI, error) { + v := &testWebRTCAPI{} + + v.mediaEngine = &webrtc.MediaEngine{} + v.registry = &interceptor.Registry{} v.settingEngine = &webrtc.SettingEngine{} + // Apply initialize filter, for example, register default codecs when create publisher/player. + for _, setup := range inits { + if setup == nil { + continue + } + + if err := setup(v); err != nil { + return nil, err + } + } + return v, nil } @@ -635,10 +706,12 @@ func (v *testWebRTCAPI) Setup(vnetClientIP string, options ...testWebRTCAPIOptio return err } + // Apply options from params, for example, tester to register vnet filter. for _, setup := range options { setup(v) } + // Apply options in api, for example, publisher register audio-level interceptor. for _, setup := range v.options { setup(v) } @@ -659,26 +732,28 @@ func (v *testWebRTCAPI) NewPeerConnection(configuration webrtc.Configuration) (* type testPlayerOptionFunc func(p *testPlayer) error type testPlayer struct { + onOffer func(s *webrtc.SessionDescription) error + onAnswer func(s *webrtc.SessionDescription) error pc *webrtc.PeerConnection receivers []*webrtc.RTPReceiver // We should dispose it. api *testWebRTCAPI // Optional suffix for stream url. streamSuffix string + // Optional app/stream to play, use srsStream by default. + defaultStream string } -func createApiForPlayer(play *testPlayer) error { - api, err := newTestWebRTCAPI() +// Create test player, the init is used to initialize api which maybe nil, +// and the options is used to setup the player itself. +func newTestPlayer(init testWebRTCAPIInitFunc, options ...testPlayerOptionFunc) (*testPlayer, error) { + v := &testPlayer{} + + api, err := newTestWebRTCAPI(init) if err != nil { - return err + return nil, err } - - play.api = api - return nil -} - -func newTestPlayer(options ...testPlayerOptionFunc) (*testPlayer, error) { - v := &testPlayer{} + v.api = api for _, opt := range options { if err := opt(v); err != nil { @@ -711,6 +786,9 @@ func (v *testPlayer) Close() error { func (v *testPlayer) Run(ctx context.Context, cancel context.CancelFunc) error { r := fmt.Sprintf("%v://%v%v", srsSchema, *srsServer, *srsStream) + if v.defaultStream != "" { + r = fmt.Sprintf("%v://%v%v", srsSchema, *srsServer, v.defaultStream) + } if v.streamSuffix != "" { r = fmt.Sprintf("%v-%v", r, v.streamSuffix) } @@ -743,22 +821,35 @@ func (v *testPlayer) Run(ctx context.Context, cancel context.CancelFunc) error { return errors.Wrapf(err, "Set offer %v", offer) } - answer, err := apiRtcRequest(ctx, "/rtc/v1/play", r, offer.SDP) + if v.onOffer != nil { + if err := v.onOffer(&offer); err != nil { + return errors.Wrapf(err, "sdp %v %v", offer.Type, offer.SDP) + } + } + + answerSDP, err := apiRtcRequest(ctx, "/rtc/v1/play", r, offer.SDP) if err != nil { return errors.Wrapf(err, "Api request offer=%v", offer.SDP) } // Run a proxy for real server and vnet. - if address, err := parseAddressOfCandidate(answer); err != nil { - return errors.Wrapf(err, "parse address of %v", answer) + if address, err := parseAddressOfCandidate(answerSDP); err != nil { + return errors.Wrapf(err, "parse address of %v", answerSDP) } else if err := v.api.proxy.Proxy(v.api.network, address); err != nil { return errors.Wrapf(err, "proxy %v to %v", v.api.network, address) } - if err := pc.SetRemoteDescription(webrtc.SessionDescription{ - Type: webrtc.SDPTypeAnswer, SDP: answer, - }); err != nil { - return errors.Wrapf(err, "Set answer %v", answer) + answer := &webrtc.SessionDescription{ + Type: webrtc.SDPTypeAnswer, SDP: answerSDP, + } + if v.onAnswer != nil { + if err := v.onAnswer(answer); err != nil { + return errors.Wrapf(err, "on answerSDP") + } + } + + if err := pc.SetRemoteDescription(*answer); err != nil { + return errors.Wrapf(err, "Set answerSDP %v", answerSDP) } handleTrack := func(ctx context.Context, track *webrtc.TrackRemote, receiver *webrtc.RTPReceiver) error { @@ -830,21 +921,19 @@ type testPublisher struct { cancel context.CancelFunc } -func createApiForPublisher(pub *testPublisher) error { - api, err := newTestWebRTCAPI() - if err != nil { - return err - } - - pub.api = api - return nil -} - -func newTestPublisher(options ...testPublisherOptionFunc) (*testPublisher, error) { +// Create test publisher, the init is used to initialize api which maybe nil, +// and the options is used to setup the publisher itself. +func newTestPublisher(init testWebRTCAPIInitFunc, options ...testPublisherOptionFunc) (*testPublisher, error) { sourceVideo, sourceAudio := *srsPublishVideo, *srsPublishAudio v := &testPublisher{} + api, err := newTestWebRTCAPI(init) + if err != nil { + return nil, err + } + v.api = api + for _, opt := range options { if err := opt(v); err != nil { return nil, err @@ -860,7 +949,6 @@ func newTestPublisher(options ...testPublisherOptionFunc) (*testPublisher, error } // Setup the interceptors for packets. - api := v.api api.options = append(api.options, func(api *testWebRTCAPI) { // Filter for RTCP packets. rtcpInterceptor := &rtcpInterceptor{} diff --git a/trunk/conf/clion-ingest.conf b/trunk/conf/clion-ingest.conf new file mode 100644 index 000000000..76fab5f41 --- /dev/null +++ b/trunk/conf/clion-ingest.conf @@ -0,0 +1,53 @@ + +listen 1935; +max_connections 1000; +daemon off; +srs_log_tank console; + +http_server { + enabled on; + listen 8080; + dir ./objs/nginx/html; +} + +http_api { + enabled on; + listen 1985; +} +stats { + network 0; +} +rtc_server { + enabled on; + # Listen at udp://8000 + listen 8000; + # + # The $CANDIDATE means fetch from env, if not configed, use * as default. + # + # The * means retrieving server IP automatically, from all network interfaces, + # @see https://github.com/ossrs/srs/wiki/v4_CN_RTCWiki#config-candidate + candidate $CANDIDATE; +} + +vhost __defaultVhost__ { + rtc { + enabled on; + } + http_remux { + enabled on; + mount [vhost]/[app]/[stream].flv; + } + ingest livestream { + enabled on; + input { + type file; + url ./doc/source.200kbps.768x320.flv; + } + ffmpeg ./objs/ffmpeg/bin/ffmpeg; + engine { + enabled off; + output rtmp://127.0.0.1:[port]/live/livestream; + } + } +} + diff --git a/trunk/src/app/srs_app_rtc_conn.cpp b/trunk/src/app/srs_app_rtc_conn.cpp index 0fce6926b..775e1ece8 100644 --- a/trunk/src/app/srs_app_rtc_conn.cpp +++ b/trunk/src/app/srs_app_rtc_conn.cpp @@ -2909,7 +2909,7 @@ srs_error_t SrsRtcConnection::negotiate_publish_capability(SrsRtcUserConfig* ruc //local_media_desc.payload_types_.back().rtcp_fb_.push_back("rrtr"); } - // TODO: FIXME: use one parse paylod from sdp. + // TODO: FIXME: use one parse payload from sdp. track_desc->create_auxiliary_payload(remote_media_desc.find_media_with_encoding_name("red")); track_desc->create_auxiliary_payload(remote_media_desc.find_media_with_encoding_name("rtx"));