update gortsplib

go.mod

@@ -5,7 +5,7 @@ go 1.15
 require (
 	github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751 // indirect
 	github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d // indirect
-	github.com/aler9/gortsplib v0.0.0-20210509121021-994cd442e361
+	github.com/aler9/gortsplib v0.0.0-20210510211300-0d6385640fad
 	github.com/asticode/go-astits v0.0.0-00010101000000-000000000000
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/fsnotify/fsnotify v1.4.9

go.sum

@@ -4,8 +4,8 @@ github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d h1:UQZhZ2O0vMHr2c
 github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d/go.mod h1:rBZYJk541a8SKzHPHnH3zbiI+7dagKZ0cgpgrD7Fyho=
 github.com/aler9/go-astits v0.0.0-20210423195926-582b09ed7c04 h1:CXgQLsU4uxWAmsXNOjGLbj0A+0IlRcpZpMgI13fmVwo=
 github.com/aler9/go-astits v0.0.0-20210423195926-582b09ed7c04/go.mod h1:DkOWmBNQpnr9mv24KfZjq4JawCFX1FCqjLVGvO0DygQ=
-github.com/aler9/gortsplib v0.0.0-20210509121021-994cd442e361 h1:GZCzLl3MiMCgQdiC+xU+eUqJ2/x8ZKba55TJhzgTNnY=
-github.com/aler9/gortsplib v0.0.0-20210509121021-994cd442e361/go.mod h1:zVCg+TQX445hh1pC5QgAuuBvvXZMWLY1XYz626dGFqY=
+github.com/aler9/gortsplib v0.0.0-20210510211300-0d6385640fad h1:u6DNZKoG3gm5pMz+K1duYS5pqbrq6iuiDB1kJc4N0d4=
+github.com/aler9/gortsplib v0.0.0-20210510211300-0d6385640fad/go.mod h1:zVCg+TQX445hh1pC5QgAuuBvvXZMWLY1XYz626dGFqY=
 github.com/aler9/rtmp v0.0.0-20210403095203-3be4a5535927 h1:95mXJ5fUCYpBRdSOnLAQAdJHHKxxxJrVCiaqDi965YQ=
 github.com/aler9/rtmp v0.0.0-20210403095203-3be4a5535927/go.mod h1:vzuE21rowz+lT1NGsWbreIvYulgBpCGnQyeTyFblUHc=
 github.com/asticode/go-astikit v0.20.0 h1:+7N+J4E4lWx2QOkRdOf6DafWJMv6O4RRfgClwQokrH8=

internal/rtspsource/source.go

@@ -1,6 +1,7 @@
 package rtspsource
 
 import (
+	"context"
 	"crypto/sha256"
 	"crypto/tls"
 	"encoding/hex"
@@ -123,49 +124,54 @@ func (s *Source) run() {
 func (s *Source) runInner() bool {
 	s.log(logger.Debug, "connecting")
 
+	client := &gortsplib.Client{
+		StreamProtocol: s.proto,
+		TLSConfig: &tls.Config{
+			InsecureSkipVerify: true,
+			VerifyConnection: func(cs tls.ConnectionState) error {
+				h := sha256.New()
+				h.Write(cs.PeerCertificates[0].Raw)
+				hstr := hex.EncodeToString(h.Sum(nil))
+				fingerprintLower := strings.ToLower(s.fingerprint)
+
+				if hstr != fingerprintLower {
+					return fmt.Errorf("server fingerprint do not match: expected %s, got %s",
+						fingerprintLower, hstr)
+				}
+
+				return nil
+			},
+		},
+		ReadTimeout:     s.readTimeout,
+		WriteTimeout:    s.writeTimeout,
+		ReadBufferCount: s.readBufferCount,
+		ReadBufferSize:  s.readBufferSize,
+		OnRequest: func(req *base.Request) {
+			s.log(logger.Debug, "c->s %v", req)
+		},
+		OnResponse: func(res *base.Response) {
+			s.log(logger.Debug, "s->c %v", res)
+		},
+	}
+
+	ctx, ctxCancel := context.WithCancel(context.Background())
+
 	var conn *gortsplib.ClientConn
 	var err error
-	dialDone := make(chan struct{}, 1)
+	dialDone := make(chan struct{})
 	go func() {
 		defer close(dialDone)
-
-		client := &gortsplib.Client{
-			StreamProtocol: s.proto,
-			TLSConfig: &tls.Config{
-				InsecureSkipVerify: true,
-				VerifyConnection: func(cs tls.ConnectionState) error {
-					h := sha256.New()
-					h.Write(cs.PeerCertificates[0].Raw)
-					hstr := hex.EncodeToString(h.Sum(nil))
-					fingerprintLower := strings.ToLower(s.fingerprint)
-
-					if hstr != fingerprintLower {
-						return fmt.Errorf("server fingerprint do not match: expected %s, got %s",
-							fingerprintLower, hstr)
-					}
-
-					return nil
-				},
-			},
-			ReadTimeout:     s.readTimeout,
-			WriteTimeout:    s.writeTimeout,
-			ReadBufferCount: s.readBufferCount,
-			ReadBufferSize:  s.readBufferSize,
-			OnRequest: func(req *base.Request) {
-				s.log(logger.Debug, "c->s %v", req)
-			},
-			OnResponse: func(res *base.Response) {
-				s.log(logger.Debug, "s->c %v", res)
-			},
-		}
-
-		conn, err = client.DialRead(s.ur)
+		conn, err = client.DialReadContext(ctx, s.ur)
 	}()
 
 	select {
 	case <-s.terminate:
+		ctxCancel()
+		<-dialDone
 		return false
+
 	case <-dialDone:
+		ctxCancel()
 	}
 
 	if err != nil {
@@ -190,17 +196,20 @@ func (s *Source) runInner() bool {
 		<-res
 	}()
 
-	done := conn.ReadFrames(func(trackID int, streamType gortsplib.StreamType, payload []byte) {
-		res.SP.OnFrame(trackID, streamType, payload)
-	})
+	readErr := make(chan error)
+	go func() {
+		readErr <- conn.ReadFrames(func(trackID int, streamType gortsplib.StreamType, payload []byte) {
+			res.SP.OnFrame(trackID, streamType, payload)
+		})
+	}()
 
 	select {
 	case <-s.terminate:
 		conn.Close()
-		<-done
+		<-readErr
 		return false
 
-	case err := <-done:
+	case err := <-readErr:
 		s.log(logger.Info, "ERR: %s", err)
 		conn.Close()
 		return true

main_clientrtsp_test.go

@@ -452,15 +452,19 @@ func TestClientRTSPPublisherOverride(t *testing.T) {
 			require.NoError(t, err)
 			defer d1.Close()
 
-			recv := make(chan struct{})
-			d1.ReadFrames(func(trackID int, streamType base.StreamType, payload []byte) {
-				if ca == "enabled" {
-					require.Equal(t, []byte{0x05, 0x06, 0x07, 0x08}, payload)
-				} else {
-					require.Equal(t, []byte{0x01, 0x02, 0x03, 0x04}, payload)
-				}
-				close(recv)
-			})
+			readDone := make(chan struct{})
+			frameRecv := make(chan struct{})
+			go func() {
+				defer close(readDone)
+				d1.ReadFrames(func(trackID int, streamType base.StreamType, payload []byte) {
+					if ca == "enabled" {
+						require.Equal(t, []byte{0x05, 0x06, 0x07, 0x08}, payload)
+					} else {
+						require.Equal(t, []byte{0x01, 0x02, 0x03, 0x04}, payload)
+					}
+					close(frameRecv)
+				})
+			}()
 
 			err = s1.WriteFrame(track.ID, gortsplib.StreamTypeRTP,
 				[]byte{0x01, 0x02, 0x03, 0x04})
@@ -476,7 +480,10 @@ func TestClientRTSPPublisherOverride(t *testing.T) {
 				require.NoError(t, err)
 			}
 
-			<-recv
+			<-frameRecv
+
+			d1.Close()
+			<-readDone
 		})
 	}
 }
@@ -511,21 +518,24 @@ func TestClientRTSPNonCompliantFrameSize(t *testing.T) {
 
 		input := bytes.Repeat([]byte{0x01, 0x02, 0x03, 0x04, 0x05}, 4096/5)
 
-		recvDone := make(chan struct{})
-		recvErr := dest.ReadFrames(func(trackID int, streamType gortsplib.StreamType, payload []byte) {
-			require.Equal(t, gortsplib.StreamTypeRTP, streamType)
-			require.Equal(t, input, payload)
-			close(recvDone)
-		})
+		readDone := make(chan struct{})
+		frameRecv := make(chan struct{})
+		go func() {
+			defer close(readDone)
+			dest.ReadFrames(func(trackID int, streamType gortsplib.StreamType, payload []byte) {
+				require.Equal(t, gortsplib.StreamTypeRTP, streamType)
+				require.Equal(t, input, payload)
+				close(frameRecv)
+			})
+		}()
 
 		err = source.WriteFrame(track.ID, gortsplib.StreamTypeRTP, input)
 		require.NoError(t, err)
 
-		select {
-		case <-recvDone:
-		case err := <-recvErr:
-			t.Error(err)
-		}
+		<-frameRecv
+
+		dest.Close()
+		<-readDone
 	})
 
 	t.Run("proxy", func(t *testing.T) {
@@ -572,21 +582,24 @@ func TestClientRTSPNonCompliantFrameSize(t *testing.T) {
 
 		input := bytes.Repeat([]byte{0x01, 0x02, 0x03, 0x04, 0x05}, 4096/5)
 
-		recvDone := make(chan struct{})
-		recvErr := dest.ReadFrames(func(trackID int, streamType gortsplib.StreamType, payload []byte) {
-			require.Equal(t, gortsplib.StreamTypeRTP, streamType)
-			require.Equal(t, input, payload)
-			close(recvDone)
-		})
+		readDone := make(chan struct{})
+		frameRecv := make(chan struct{})
+		go func() {
+			defer close(readDone)
+			dest.ReadFrames(func(trackID int, streamType gortsplib.StreamType, payload []byte) {
+				require.Equal(t, gortsplib.StreamTypeRTP, streamType)
+				require.Equal(t, input, payload)
+				close(frameRecv)
+			})
+		}()
 
 		err = source.WriteFrame(track.ID, gortsplib.StreamTypeRTP, input)
 		require.NoError(t, err)
 
-		select {
-		case <-recvDone:
-		case err := <-recvErr:
-			t.Error(err)
-		}
+		<-frameRecv
+
+		dest.Close()
+		<-readDone
 	})
 }