SquashSRS4: Update srs-bench.

trunk/3rdparty/srs-bench/LICENSE

@@ -1,6 +1,6 @@
 The MIT License (MIT)
 
-Copyright (c) 2021 srs-bench(ossrs)
+Copyright (c) 2021 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

trunk/3rdparty/srs-bench/README.md

@@ -142,6 +142,12 @@ go test ./srs -mod=vendor -v -srs-server=127.0.0.1
 make && ./objs/srs_test -test.v -srs-server=127.0.0.1
 ```
 
+可以只运行某个用例，并打印详细日志，比如：
+
+```bash
+make && ./objs/srs_test -test.v -srs-log -test.run TestRtcBasic_PublishPlay
+```
+
 支持的参数如下：
 
 * `-srs-server`，RTC服务器地址。默认值：`127.0.0.1`

trunk/3rdparty/srs-bench/auto/sync_vnet.sh

@@ -0,0 +1,33 @@
+#!/bin/bash
+
+FILES=(udpproxy.go udpproxy_test.go)
+for file in ${FILES[@]}; do
+  echo "cp vnet/udpproxy.go ~/git/transport/vnet/" &&
+  cp vnet/udpproxy.go ~/git/transport/vnet/
+done
+
+# https://github.com/pion/webrtc/wiki/Contributing#run-all-automated-tests-and-checks-before-submitting
+cd ~/git/transport/
+
+echo ".github/lint-commit-message.sh" &&
+.github/lint-commit-message.sh &&
+echo ".github/assert-contributors.sh" &&
+.github/assert-contributors.sh &&
+echo ".github/lint-disallowed-functions-in-library.sh" &&
+.github/lint-disallowed-functions-in-library.sh &&
+echo ".github/lint-filename.sh" &&
+.github/lint-filename.sh
+if [[ $? -ne 0 ]]; then echo "fail"; exit -1; fi
+
+# https://github.com/pion/webrtc/wiki/Contributing#run-all-automated-tests-and-checks-before-submitting
+cd ~/git/transport/vnet/
+
+echo "go test -race ./..." &&
+go test -race ./...
+if [[ $? -ne 0 ]]; then echo "fail"; exit -1; fi
+
+echo "golangci-lint run --skip-files conn_map_test.go" &&
+golangci-lint run --skip-files conn_map_test.go
+if [[ $? -ne 0 ]]; then echo "fail"; exit -1; fi
+
+echo "OK"

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

@@ -44,16 +44,20 @@ type videoIngester struct {
 	markerInterceptor *RTPInterceptor
 	sVideoTrack       *webrtc.TrackLocalStaticSample
 	sVideoSender      *webrtc.RTPSender
+	ready             context.Context
+	readyCancel       context.CancelFunc
 }
 
 func NewVideoIngester(sourceVideo string) *videoIngester {
-	return &videoIngester{markerInterceptor: &RTPInterceptor{}, sourceVideo: sourceVideo}
+	v := &videoIngester{markerInterceptor: &RTPInterceptor{}, sourceVideo: sourceVideo}
+	v.ready, v.readyCancel = context.WithCancel(context.Background())
+	return v
 }
 
 func (v *videoIngester) Close() error {
+	v.readyCancel()
 	if v.sVideoSender != nil {
-		v.sVideoSender.Stop()
+		_ = v.sVideoSender.Stop()
-		v.sVideoSender = nil
 	}
 	return nil
 }
@@ -102,6 +106,9 @@ func (v *videoIngester) Ingest(ctx context.Context) error {
 	logger.Tf(ctx, "Video %v, tbn=%v, fps=%v, ssrc=%v, pt=%v, header=%v",
 		codec.MimeType, codec.ClockRate, fps, enc.SSRC, codec.PayloadType, headers)
 
+	// OK, we are ready.
+	v.readyCancel()
+
 	clock := newWallClock()
 	sampleDuration := time.Duration(uint64(time.Millisecond) * 1000 / uint64(fps))
 	for ctx.Err() == nil {
@@ -179,16 +186,21 @@ type audioIngester struct {
 	audioLevelInterceptor *RTPInterceptor
 	sAudioTrack           *webrtc.TrackLocalStaticSample
 	sAudioSender          *webrtc.RTPSender
+	ready                 context.Context
+	readyCancel           context.CancelFunc
 }
 
 func NewAudioIngester(sourceAudio string) *audioIngester {
-	return &audioIngester{audioLevelInterceptor: &RTPInterceptor{}, sourceAudio: sourceAudio}
+	v := &audioIngester{audioLevelInterceptor: &RTPInterceptor{}, sourceAudio: sourceAudio}
+	v.ready, v.readyCancel = context.WithCancel(context.Background())
+	return v
 }
 
 func (v *audioIngester) Close() error {
+	v.readyCancel() // OK we are closed, also ready.
+
 	if v.sAudioSender != nil {
-		v.sAudioSender.Stop()
+		_ = v.sAudioSender.Stop()
-		v.sAudioSender = nil
 	}
 	return nil
 }
@@ -240,6 +252,9 @@ func (v *audioIngester) Ingest(ctx context.Context) error {
 		}
 	}
 
+	// OK, we are ready.
+	v.readyCancel()
+
 	clock := newWallClock()
 	var lastGranule uint64
 
@@ -253,7 +268,7 @@ func (v *audioIngester) Ingest(ctx context.Context) error {
 		}
 
 		// The amount of samples is the difference between the last and current timestamp
-		sampleCount := uint64(pageHeader.GranulePosition - lastGranule)
+		sampleCount := pageHeader.GranulePosition - lastGranule
 		lastGranule = pageHeader.GranulePosition
 		sampleDuration := time.Duration(uint64(time.Millisecond) * 1000 * sampleCount / uint64(codec.ClockRate))
 
@@ -266,7 +281,7 @@ func (v *audioIngester) Ingest(ctx context.Context) error {
 						return 0, err
 					}
 
-					header.SetExtension(uint8(audioLevel.ID), audioLevelPayload)
+					_ = header.SetExtension(uint8(audioLevel.ID), audioLevelPayload)
 				}
 
 				return ri.nextRTPWriter.Write(header, payload, attributes)

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

@@ -36,7 +36,6 @@ import (
 	"strconv"
 	"strings"
 	"sync"
-	"testing"
 	"time"
 
 	"github.com/ossrs/go-oryx-lib/errors"
@@ -207,7 +206,7 @@ func apiRtcRequest(ctx context.Context, apiPath, r, offer string) (string, error
 	logger.Tf(ctx, "Parse response to code=%v, session=%v, sdp=%v bytes",
 		resBody.Code, resBody.Session, len(resBody.SDP))
 
-	return string(resBody.SDP), nil
+	return resBody.SDP, nil
 }
 
 func escapeSDP(sdp string) string {
@@ -219,7 +218,7 @@ func packageAsSTAPA(frames ...*h264reader.NAL) *h264reader.NAL {
 
 	buf := bytes.Buffer{}
 	buf.WriteByte(
-		byte(first.RefIdc<<5)&0x60 | byte(24), // STAP-A
+		first.RefIdc<<5&0x60 | byte(24), // STAP-A
 	)
 
 	for _, frame := range frames {
@@ -325,6 +324,14 @@ func filterTestError(errs ...error) error {
 		if err == nil || errors.Cause(err) == context.Canceled {
 			continue
 		}
+
+		// If url error, server maybe error, do not print the detail log.
+		if r0 := errors.Cause(err); r0 != nil {
+			if r1, ok := r0.(*url.Error); ok {
+				err = r1
+			}
+		}
+
 		filteredErrors = append(filteredErrors, err)
 	}
 
@@ -352,13 +359,13 @@ func srsIsStun(b []byte) bool {
 // @see https://tools.ietf.org/html/rfc2246#section-6.2.1
 // @see srs_is_dtls of https://github.com/ossrs/srs
 func srsIsDTLS(b []byte) bool {
-	return (len(b) >= 13 && (b[0] > 19 && b[0] < 64))
+	return len(b) >= 13 && (b[0] > 19 && b[0] < 64)
 }
 
 // For RTP or RTCP, the V=2 which is in the high 2bits, 0xC0 (1100 0000)
 // @see srs_is_rtp_or_rtcp of https://github.com/ossrs/srs
 func srsIsRTPOrRTCP(b []byte) bool {
-	return (len(b) >= 12 && (b[0]&0xC0) == 0x80)
+	return len(b) >= 12 && (b[0]&0xC0) == 0x80
 }
 
 // For RTCP, PT is [128, 223] (or without marker [0, 95]).
@@ -554,7 +561,7 @@ func (v *DTLSRecord) Unmarshal(b []byte) error {
 		return errors.Errorf("requires 13B only %v", len(b))
 	}
 
-	v.ContentType = DTLSContentType(uint8(b[0]))
+	v.ContentType = DTLSContentType(b[0])
 	v.Version = uint16(b[1])<<8 | uint16(b[2])
 	v.Epoch = uint16(b[3])<<8 | uint16(b[4])
 	v.SequenceNumber = uint64(b[5])<<40 | uint64(b[6])<<32 | uint64(b[7])<<24 | uint64(b[8])<<16 | uint64(b[9])<<8 | uint64(b[10])
@@ -605,11 +612,11 @@ func NewTestWebRTCAPI(options ...TestWebRTCAPIOptionFunc) (*TestWebRTCAPI, error
 
 func (v *TestWebRTCAPI) Close() error {
 	if v.proxy != nil {
-		v.proxy.Close()
+		_ = v.proxy.Close()
 	}
 
 	if v.router != nil {
-		v.router.Stop()
+		_ = v.router.Stop()
 	}
 
 	return nil
@@ -676,14 +683,24 @@ type TestPlayerOptionFunc func(p *TestPlayer) error
 type TestPlayer struct {
 	pc        *webrtc.PeerConnection
 	receivers []*webrtc.RTPReceiver
-	// root api object
+	// We should dispose it.
 	api *TestWebRTCAPI
 	// Optional suffix for stream url.
 	streamSuffix string
 }
 
-func NewTestPlayer(api *TestWebRTCAPI, options ...TestPlayerOptionFunc) (*TestPlayer, error) {
+func CreateApiForPlayer(play *TestPlayer) error {
-	v := &TestPlayer{api: api}
+	api, err := NewTestWebRTCAPI()
+	if err != nil {
+		return err
+	}
+
+	play.api = api
+	return nil
+}
+
+func NewTestPlayer(options ...TestPlayerOptionFunc) (*TestPlayer, error) {
+	v := &TestPlayer{}
 
 	for _, opt := range options {
 		if err := opt(v); err != nil {
@@ -691,19 +708,24 @@ func NewTestPlayer(api *TestWebRTCAPI, options ...TestPlayerOptionFunc) (*TestPl
 		}
 	}
 
-	// The api might be override by options.
-	api = v.api
-
 	return v, nil
 }
 
+func (v *TestPlayer) Setup(vnetClientIP string, options ...TestWebRTCAPIOptionFunc) error {
+	return v.api.Setup(vnetClientIP, options...)
+}
+
 func (v *TestPlayer) Close() error {
 	if v.pc != nil {
-		v.pc.Close()
+		_ = v.pc.Close()
 	}
 
 	for _, receiver := range v.receivers {
-		receiver.Stop()
+		_ = receiver.Stop()
+	}
+
+	if v.api != nil {
+		_ = v.api.Close()
 	}
 
 	return nil
@@ -723,12 +745,16 @@ func (v *TestPlayer) Run(ctx context.Context, cancel context.CancelFunc) error {
 	}
 	v.pc = pc
 
-	pc.AddTransceiverFromKind(webrtc.RTPCodecTypeAudio, webrtc.RTPTransceiverInit{
+	if _, err := pc.AddTransceiverFromKind(webrtc.RTPCodecTypeAudio, webrtc.RTPTransceiverInit{
 		Direction: webrtc.RTPTransceiverDirectionRecvonly,
-	})
+	}); err != nil {
-	pc.AddTransceiverFromKind(webrtc.RTPCodecTypeVideo, webrtc.RTPTransceiverInit{
+		return errors.Wrapf(err, "add track")
+	}
+	if _, err := pc.AddTransceiverFromKind(webrtc.RTPCodecTypeVideo, webrtc.RTPTransceiverInit{
 		Direction: webrtc.RTPTransceiverDirectionRecvonly,
-	})
+	}); err != nil {
+		return errors.Wrapf(err, "add track")
+	}
 
 	offer, err := pc.CreateOffer(nil)
 	if err != nil {
@@ -818,16 +844,28 @@ type TestPublisher struct {
 	aIngester *audioIngester
 	vIngester *videoIngester
 	pc        *webrtc.PeerConnection
-	// root api object
+	// We should dispose it.
 	api *TestWebRTCAPI
 	// Optional suffix for stream url.
 	streamSuffix string
+	// To cancel the publisher, pass by Run.
+	cancel context.CancelFunc
 }
 
-func NewTestPublisher(api *TestWebRTCAPI, options ...TestPublisherOptionFunc) (*TestPublisher, error) {
+func CreateApiForPublisher(pub *TestPublisher) error {
+	api, err := NewTestWebRTCAPI()
+	if err != nil {
+		return err
+	}
+
+	pub.api = api
+	return nil
+}
+
+func NewTestPublisher(options ...TestPublisherOptionFunc) (*TestPublisher, error) {
 	sourceVideo, sourceAudio := *srsPublishVideo, *srsPublishAudio
 
-	v := &TestPublisher{api: api}
+	v := &TestPublisher{}
 
 	for _, opt := range options {
 		if err := opt(v); err != nil {
@@ -835,9 +873,6 @@ func NewTestPublisher(api *TestWebRTCAPI, options ...TestPublisherOptionFunc) (*
 		}
 	}
 
-	// The api might be override by options.
-	api = v.api
-
 	// Create ingesters.
 	if sourceAudio != "" {
 		v.aIngester = NewAudioIngester(sourceAudio)
@@ -847,6 +882,7 @@ func NewTestPublisher(api *TestWebRTCAPI, options ...TestPublisherOptionFunc) (*
 	}
 
 	// Setup the interceptors for packets.
+	api := v.api
 	api.options = append(api.options, func(api *TestWebRTCAPI) {
 		// Filter for RTCP packets.
 		rtcpInterceptor := &RTCPInterceptor{}
@@ -870,17 +906,25 @@ func NewTestPublisher(api *TestWebRTCAPI, options ...TestPublisherOptionFunc) (*
 	return v, nil
 }
 
+func (v *TestPublisher) Setup(vnetClientIP string, options ...TestWebRTCAPIOptionFunc) error {
+	return v.api.Setup(vnetClientIP, options...)
+}
+
 func (v *TestPublisher) Close() error {
 	if v.vIngester != nil {
-		v.vIngester.Close()
+		_ = v.vIngester.Close()
 	}
 
 	if v.aIngester != nil {
-		v.aIngester.Close()
+		_ = v.aIngester.Close()
 	}
 
 	if v.pc != nil {
-		v.pc.Close()
+		_ = v.pc.Close()
+	}
+
+	if v.api != nil {
+		_ = v.api.Close()
 	}
 
 	return nil
@@ -892,6 +936,9 @@ func (v *TestPublisher) SetStreamSuffix(suffix string) *TestPublisher {
 }
 
 func (v *TestPublisher) Run(ctx context.Context, cancel context.CancelFunc) error {
+	// Save the cancel.
+	v.cancel = cancel
+
 	r := fmt.Sprintf("%v://%v%v", srsSchema, *srsServer, *srsStream)
 	if v.streamSuffix != "" {
 		r = fmt.Sprintf("%v-%v", r, v.streamSuffix)
@@ -1012,11 +1059,17 @@ func (v *TestPublisher) Run(ctx context.Context, cancel context.CancelFunc) erro
 		<-ctx.Done()
 
 		if v.aIngester != nil && v.aIngester.sAudioSender != nil {
-			v.aIngester.sAudioSender.Stop()
+			// We MUST wait for the ingester ready(or closed), because it might crash if sender is disposed.
+			<-v.aIngester.ready.Done()
+
+			_ = v.aIngester.Close()
 		}
 
 		if v.vIngester != nil && v.vIngester.sVideoSender != nil {
-			v.vIngester.sVideoSender.Stop()
+			// We MUST wait for the ingester ready(or closed), because it might crash if sender is disposed.
+			<-v.vIngester.ready.Done()
+
+			_ = v.vIngester.Close()
 		}
 	}()
 
@@ -1028,6 +1081,7 @@ func (v *TestPublisher) Run(ctx context.Context, cancel context.CancelFunc) erro
 		if v.aIngester == nil {
 			return
 		}
+		defer v.aIngester.readyCancel()
 
 		select {
 		case <-ctx.Done():
@@ -1072,6 +1126,7 @@ func (v *TestPublisher) Run(ctx context.Context, cancel context.CancelFunc) erro
 		if v.vIngester == nil {
 			return
 		}
+		defer v.vIngester.readyCancel()
 
 		select {
 		case <-ctx.Done():
@@ -1119,47 +1174,3 @@ func (v *TestPublisher) Run(ctx context.Context, cancel context.CancelFunc) erro
 	}
 	return ctx.Err()
 }
-
-func TestRTCServerVersion(t *testing.T) {
-	api := fmt.Sprintf("http://%v:1985/api/v1/versions", *srsServer)
-	req, err := http.NewRequest("POST", api, nil)
-	if err != nil {
-		t.Errorf("Request %v", api)
-		return
-	}
-
-	res, err := http.DefaultClient.Do(req)
-	if err != nil {
-		t.Errorf("Do request %v", api)
-		return
-	}
-
-	b, err := ioutil.ReadAll(res.Body)
-	if err != nil {
-		t.Errorf("Read body of %v", api)
-		return
-	}
-
-	obj := struct {
-		Code   int    `json:"code"`
-		Server string `json:"server"`
-		Data   struct {
-			Major    int    `json:"major"`
-			Minor    int    `json:"minor"`
-			Revision int    `json:"revision"`
-			Version  string `json:"version"`
-		} `json:"data"`
-	}{}
-	if err := json.Unmarshal(b, &obj); err != nil {
-		t.Errorf("Parse %v", string(b))
-		return
-	}
-	if obj.Code != 0 {
-		t.Errorf("Server err code=%v, server=%v", obj.Code, obj.Server)
-		return
-	}
-	if obj.Data.Major == 0 && obj.Data.Minor == 0 {
-		t.Errorf("Invalid version %v", obj.Data)
-		return
-	}
-}

trunk/3rdparty/srs-bench/vnet/udpproxy.go

@@ -1,34 +1,13 @@
-// The MIT License (MIT)
-//
-// Copyright (c) 2021 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 vnet
 
 import (
+	"context"
 	"net"
 	"sync"
 	"time"
-
-	"github.com/pion/transport/vnet"
 )
 
-// A UDP proxy between real server(net.UDPConn) and vnet.UDPConn.
+// UDPProxy is a proxy between real server(net.UDPConn) and vnet.UDPConn.
 //
 // High level design:
 //                           ..............................................
@@ -44,32 +23,9 @@ import (
 //                           :            |        |          ............................:
 //                           :            +--------+                       :
 //                           ...............................................
-//
-// The whole big picture:
-//                           ......................................
-//                           :         Virtual Network (vnet)     :
-//                           :                                    :
-//   +-------+ *         1 +----+         +--------+              :
-//   | :App  |------------>|:Net|--o<-----|:Router |          .............................
-//   +-------+             +----+         |        |          :        UDPProxy           :
-//   +-----------+ *     1 +----+         |        |       +----+     +---------+     +---------+     +--------+
-//   |:STUNServer|-------->|:Net|--o<-----|        |--->o--|:Net|-->o-| vnet.   |-->o-|  net.   |--->-| :Real  |
-//   +-----------+         +----+         |        |       +----+     | UDPConn |     | UDPConn |     | Server |
-//   +-----------+ *     1 +----+         |        |          :       +---------+     +---------+     +--------+
-//   |:TURNServer|-------->|:Net|--o<-----|        |          ............................:
-//   +-----------+         +----+ [1]     |        |              :
-//                           :          1 |        | 1  <<has>>   :
-//                           :      +---<>|        |<>----+ [2]   :
-//                           :      |     +--------+      |       :
-//                         To form  |      *|             v 0..1  :
-//                   a subnet tree  |       o [3]      +-----+    :
-//                           :      |       ^          |:NAT |    :
-//                           :      |       |          +-----+    :
-//                           :      +-------+                     :
-//                           ......................................
 type UDPProxy struct {
 	// The router bind to.
-	router *vnet.Router
+	router *Router
 
 	// Each vnet source, bind to a real socket to server.
 	// key is real server addr, which is net.Addr
@@ -88,19 +44,22 @@ type UDPProxy struct {
 // NewProxy create a proxy, the router for this proxy belongs/bind to. If need to proxy for
 // please create a new proxy for each router. For all addresses we proxy, we will create a
 // vnet.Net in this router and proxy all packets.
-func NewProxy(router *vnet.Router) (*UDPProxy, error) {
+func NewProxy(router *Router) (*UDPProxy, error) {
 	v := &UDPProxy{router: router, timeout: 2 * time.Minute}
 	return v, nil
 }
 
 // Close the proxy, stop all workers.
 func (v *UDPProxy) Close() error {
-	// nolint:godox // TODO: FIXME: Do cleanup.
+	v.workers.Range(func(key, value interface{}) bool {
+		_ = value.(*aUDPProxyWorker).Close()
+		return true
+	})
 	return nil
 }
 
 // Proxy starts a worker for server, ignore if already started.
-func (v *UDPProxy) Proxy(client *vnet.Net, server *net.UDPAddr) error {
+func (v *UDPProxy) Proxy(client *Net, server *net.UDPAddr) error {
 	// Note that even if the worker exists, it's also ok to create a same worker,
 	// because the router will use the last one, and the real server will see a address
 	// change event after we switch to the next worker.
@@ -113,25 +72,44 @@ func (v *UDPProxy) Proxy(client *vnet.Net, server *net.UDPAddr) error {
 	worker := &aUDPProxyWorker{
 		router: v.router, mockRealServerAddr: v.mockRealServerAddr,
 	}
+
+	// Create context for cleanup.
+	var ctx context.Context
+	ctx, worker.ctxDisposeCancel = context.WithCancel(context.Background())
+
 	v.workers.Store(server.String(), worker)
 
-	return worker.Proxy(client, server)
+	return worker.Proxy(ctx, client, server)
 }
 
 // A proxy worker for a specified proxy server.
 type aUDPProxyWorker struct {
-	router             *vnet.Router
+	router             *Router
 	mockRealServerAddr *net.UDPAddr
 
 	// Each vnet source, bind to a real socket to server.
 	// key is vnet client addr, which is net.Addr
 	// value is *net.UDPConn
 	endpoints sync.Map
+
+	// For cleanup.
+	ctxDisposeCancel context.CancelFunc
+	wg               sync.WaitGroup
 }
 
-func (v *aUDPProxyWorker) Proxy(client *vnet.Net, serverAddr *net.UDPAddr) error { // nolint:gocognit
+func (v *aUDPProxyWorker) Close() error {
+	// Notify all goroutines to dispose.
+	v.ctxDisposeCancel()
+
+	// Wait for all goroutines quit.
+	v.wg.Wait()
+
+	return nil
+}
+
+func (v *aUDPProxyWorker) Proxy(ctx context.Context, client *Net, serverAddr *net.UDPAddr) error { // nolint:gocognit
 	// Create vnet for real server by serverAddr.
-	nw := vnet.NewNet(&vnet.NetConfig{
+	nw := NewNet(&NetConfig{
 		StaticIP: serverAddr.IP.String(),
 	})
 	if err := v.router.AddNet(nw); err != nil {
@@ -145,10 +123,71 @@ func (v *aUDPProxyWorker) Proxy(client *vnet.Net, serverAddr *net.UDPAddr) error
 		return err
 	}
 
-	// Start a proxy goroutine.
+	// User stop proxy, we should close the socket.
-	var findEndpointBy func(addr net.Addr) (*net.UDPConn, error)
-	// nolint:godox // TODO: FIXME: Do cleanup.
 	go func() {
+		<-ctx.Done()
+		_ = vnetSocket.Close()
+	}()
+
+	// Got new vnet client, start a new endpoint.
+	findEndpointBy := func(addr net.Addr) (*net.UDPConn, error) {
+		// Exists binding.
+		if value, ok := v.endpoints.Load(addr.String()); ok {
+			// Exists endpoint, reuse it.
+			return value.(*net.UDPConn), nil
+		}
+
+		// The real server we proxy to, for utest to mock it.
+		realAddr := serverAddr
+		if v.mockRealServerAddr != nil {
+			realAddr = v.mockRealServerAddr
+		}
+
+		// Got new vnet client, create new endpoint.
+		realSocket, err := net.DialUDP("udp4", nil, realAddr)
+		if err != nil {
+			return nil, err
+		}
+
+		// User stop proxy, we should close the socket.
+		go func() {
+			<-ctx.Done()
+			_ = realSocket.Close()
+		}()
+
+		// Bind address.
+		v.endpoints.Store(addr.String(), realSocket)
+
+		// Got packet from real serverAddr, we should proxy it to vnet.
+		v.wg.Add(1)
+		go func(vnetClientAddr net.Addr) {
+			defer v.wg.Done()
+
+			buf := make([]byte, 1500)
+			for {
+				n, _, err := realSocket.ReadFrom(buf)
+				if err != nil {
+					return
+				}
+
+				if n <= 0 {
+					continue // Drop packet
+				}
+
+				if _, err := vnetSocket.WriteTo(buf[:n], vnetClientAddr); err != nil {
+					return
+				}
+			}
+		}(addr)
+
+		return realSocket, nil
+	}
+
+	// Start a proxy goroutine.
+	v.wg.Add(1)
+	go func() {
+		defer v.wg.Done()
+
 		buf := make([]byte, 1500)
 
 		for {
@@ -172,51 +211,5 @@ func (v *aUDPProxyWorker) Proxy(client *vnet.Net, serverAddr *net.UDPAddr) error
 		}
 	}()
 
-	// Got new vnet client, start a new endpoint.
-	findEndpointBy = func(addr net.Addr) (*net.UDPConn, error) {
-		// Exists binding.
-		if value, ok := v.endpoints.Load(addr.String()); ok {
-			// Exists endpoint, reuse it.
-			return value.(*net.UDPConn), nil
-		}
-
-		// The real server we proxy to, for utest to mock it.
-		realAddr := serverAddr
-		if v.mockRealServerAddr != nil {
-			realAddr = v.mockRealServerAddr
-		}
-
-		// Got new vnet client, create new endpoint.
-		realSocket, err := net.DialUDP("udp4", nil, realAddr)
-		if err != nil {
-			return nil, err
-		}
-
-		// Bind address.
-		v.endpoints.Store(addr.String(), realSocket)
-
-		// Got packet from real serverAddr, we should proxy it to vnet.
-		// nolint:godox // TODO: FIXME: Do cleanup.
-		go func(vnetClientAddr net.Addr) {
-			buf := make([]byte, 1500)
-			for {
-				n, _, err := realSocket.ReadFrom(buf)
-				if err != nil {
-					return
-				}
-
-				if n <= 0 {
-					continue // Drop packet
-				}
-
-				if _, err := vnetSocket.WriteTo(buf[:n], vnetClientAddr); err != nil {
-					return
-				}
-			}
-		}(addr)
-
-		return realSocket, nil
-	}
-
 	return nil
 }

trunk/3rdparty/srs-bench/vnet/udpproxy_test.go

@@ -1,23 +1,5 @@
-// The MIT License (MIT)
+// +build !wasm
-//
+
-// Copyright (c) 2021 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 vnet
 
 import (
@@ -32,7 +14,6 @@ import (
 	"time"
 
 	"github.com/pion/logging"
-	"github.com/pion/transport/vnet"
 )
 
 type MockUDPEchoServer struct {
@@ -163,7 +144,7 @@ func TestUDPProxyOne2One(t *testing.T) {
 		}
 
 		doVnetProxy := func() error {
-			router, err := vnet.NewRouter(&vnet.RouterConfig{
+			router, err := NewRouter(&RouterConfig{
 				CIDR:          "0.0.0.0/0",
 				LoggerFactory: logging.NewDefaultLoggerFactory(),
 			})
@@ -171,7 +152,7 @@ func TestUDPProxyOne2One(t *testing.T) {
 				return err
 			}
 
-			clientNetwork := vnet.NewNet(&vnet.NetConfig{
+			clientNetwork := NewNet(&NetConfig{
 				StaticIP: "10.0.0.11",
 			})
 			if err = router.AddNet(clientNetwork); err != nil {
@@ -309,7 +290,7 @@ func TestUDPProxyTwo2One(t *testing.T) {
 		}
 
 		doVnetProxy := func() error {
-			router, err := vnet.NewRouter(&vnet.RouterConfig{
+			router, err := NewRouter(&RouterConfig{
 				CIDR:          "0.0.0.0/0",
 				LoggerFactory: logging.NewDefaultLoggerFactory(),
 			})
@@ -317,7 +298,7 @@ func TestUDPProxyTwo2One(t *testing.T) {
 				return err
 			}
 
-			clientNetwork := vnet.NewNet(&vnet.NetConfig{
+			clientNetwork := NewNet(&NetConfig{
 				StaticIP: "10.0.0.11",
 			})
 			if err = router.AddNet(clientNetwork); err != nil {
@@ -487,7 +468,7 @@ func TestUDPProxyProxyTwice(t *testing.T) {
 		}
 
 		doVnetProxy := func() error {
-			router, err := vnet.NewRouter(&vnet.RouterConfig{
+			router, err := NewRouter(&RouterConfig{
 				CIDR:          "0.0.0.0/0",
 				LoggerFactory: logging.NewDefaultLoggerFactory(),
 			})
@@ -495,7 +476,7 @@ func TestUDPProxyProxyTwice(t *testing.T) {
 				return err
 			}
 
-			clientNetwork := vnet.NewNet(&vnet.NetConfig{
+			clientNetwork := NewNet(&NetConfig{
 				StaticIP: "10.0.0.11",
 			})
 			if err = router.AddNet(clientNetwork); err != nil {

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

@@ -0,0 +1,38 @@
+// The MIT License (MIT)
+//
+// Copyright (c) 2021 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 vnet
+
+import (
+	"github.com/pion/transport/vnet"
+)
+
+type Router = vnet.Router
+type Net = vnet.Net
+
+type NetConfig = vnet.NetConfig
+type RouterConfig = vnet.RouterConfig
+
+func NewNet(config *NetConfig) *Net {
+	return vnet.NewNet(config)
+}
+func NewRouter(config *RouterConfig) (*Router, error) {
+	return vnet.NewRouter(config)
+}

trunk/src/app/srs_app_rtc_conn.cpp

@@ -429,19 +429,18 @@ srs_error_t SrsRtcPlayStream::initialize(SrsRequest* req, std::map<uint32_t, Srs
         return srs_error_wrap(err, "rtc fetch source failed");
     }
 
-    if (true) {
+    for (map<uint32_t, SrsRtcTrackDescription*>::iterator it = sub_relations.begin(); it != sub_relations.end(); ++it) {
-        std::map<uint32_t, SrsRtcTrackDescription*>::iterator it = sub_relations.begin();
+        uint32_t ssrc = it->first;
-        while (it != sub_relations.end()) {
+        SrsRtcTrackDescription* desc = it->second;
-            if (it->second->type_ == "audio") {
-                SrsRtcAudioSendTrack* track = new SrsRtcAudioSendTrack(session_, it->second);
-                audio_tracks_.insert(make_pair(it->first, track));
-            }
 
-            if (it->second->type_ == "video") {
+        if (desc->type_ == "audio") {
-                SrsRtcVideoSendTrack* track = new SrsRtcVideoSendTrack(session_, it->second);
+            SrsRtcAudioSendTrack* track = new SrsRtcAudioSendTrack(session_, desc);
-                video_tracks_.insert(make_pair(it->first, track));
+            audio_tracks_.insert(make_pair(ssrc, track));
-            }
+        }
-            ++it;
+
+        if (desc->type_ == "video") {
+            SrsRtcVideoSendTrack* track = new SrsRtcVideoSendTrack(session_, desc);
+            video_tracks_.insert(make_pair(ssrc, track));
         }
     }
 
@@ -605,7 +604,7 @@ srs_error_t SrsRtcPlayStream::send_packet(SrsRtpPacket2*& pkt)
 
     // TODO: FIXME: Maybe refine for performance issue.
     if (!audio_tracks_.count(pkt->header.get_ssrc()) && !video_tracks_.count(pkt->header.get_ssrc())) {
-        srs_warn("ssrc %u not found", pkt->header.get_ssrc());
+        srs_warn("RTC: Drop for ssrc %u not found", pkt->header.get_ssrc());
         return err;
     }