From 6d02ec947860eaa400d820e0cbacd7de00439797 Mon Sep 17 00:00:00 2001 From: aler9 <46489434+aler9@users.noreply.github.com> Date: Mon, 10 May 2021 23:23:56 +0200 Subject: [PATCH] update gortsplib --- go.mod | 2 +- go.sum | 4 +- internal/rtspsource/source.go | 85 +++++++++++++++++++---------------- main_clientrtsp_test.go | 77 ++++++++++++++++++------------- 4 files changed, 95 insertions(+), 73 deletions(-) diff --git a/go.mod b/go.mod index a6d7a014..c2651bd9 100644 --- a/go.mod +++ b/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 diff --git a/go.sum b/go.sum index 28ffd34a..1cbe5d37 100644 --- a/go.sum +++ b/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= diff --git a/internal/rtspsource/source.go b/internal/rtspsource/source.go index e90157a2..0dc206ba 100644 --- a/internal/rtspsource/source.go +++ b/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 diff --git a/main_clientrtsp_test.go b/main_clientrtsp_test.go index 334500fb..8b0d2ec4 100644 --- a/main_clientrtsp_test.go +++ b/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 }) }