From 8846b9f5f5c68e1ec41812a20b81f16291514c32 Mon Sep 17 00:00:00 2001 From: aler9 <46489434+aler9@users.noreply.github.com> Date: Sun, 25 Apr 2021 18:56:54 +0200 Subject: [PATCH] rename ClientConf into Client --- clientconf.go => client.go | 21 ++++--- clientconn.go | 80 ++++++++++++------------- clientconnpublish.go | 4 +- clientconnpublish_test.go | 20 +++---- clientconnread.go | 14 ++--- clientconnread_test.go | 28 ++++----- clientconnudpl.go | 6 +- examples/client-publish-options/main.go | 6 +- examples/client-read-options/main.go | 6 +- 9 files changed, 92 insertions(+), 93 deletions(-) rename clientconf.go => client.go (86%) diff --git a/clientconf.go b/client.go similarity index 86% rename from clientconf.go rename to client.go index 79be51d5..a376be7f 100644 --- a/clientconf.go +++ b/client.go @@ -9,27 +9,26 @@ import ( "github.com/aler9/gortsplib/pkg/headers" ) -// DefaultClientConf is the default ClientConf. -var DefaultClientConf = ClientConf{} +// DefaultClient is the default Client. +var DefaultClient = &Client{} // Dial connects to a server. func Dial(scheme string, host string) (*ClientConn, error) { - return DefaultClientConf.Dial(scheme, host) + return DefaultClient.Dial(scheme, host) } // DialRead connects to a server and starts reading all tracks. func DialRead(address string) (*ClientConn, error) { - return DefaultClientConf.DialRead(address) + return DefaultClient.DialRead(address) } // DialPublish connects to a server and starts publishing the tracks. func DialPublish(address string, tracks Tracks) (*ClientConn, error) { - return DefaultClientConf.DialPublish(address, tracks) + return DefaultClient.DialPublish(address, tracks) } -// ClientConf allows to initialize a ClientConn. -// All fields are optional. -type ClientConf struct { +// Client is a RTSP client. +type Client struct { // the stream protocol (UDP or TCP). // If nil, it is chosen automatically (first UDP, then, if it fails, TCP). // It defaults to nil. @@ -91,12 +90,12 @@ type ClientConf struct { } // Dial connects to a server. -func (c ClientConf) Dial(scheme string, host string) (*ClientConn, error) { +func (c *Client) Dial(scheme string, host string) (*ClientConn, error) { return newClientConn(c, scheme, host) } // DialRead connects to the address and starts reading all tracks. -func (c ClientConf) DialRead(address string) (*ClientConn, error) { +func (c *Client) DialRead(address string) (*ClientConn, error) { u, err := base.ParseURL(address) if err != nil { return nil, err @@ -137,7 +136,7 @@ func (c ClientConf) DialRead(address string) (*ClientConn, error) { } // DialPublish connects to the address and starts publishing the tracks. -func (c ClientConf) DialPublish(address string, tracks Tracks) (*ClientConn, error) { +func (c *Client) DialPublish(address string, tracks Tracks) (*ClientConn, error) { u, err := base.ParseURL(address) if err != nil { return nil, err diff --git a/clientconn.go b/clientconn.go index fd93eecc..d57313ba 100644 --- a/clientconn.go +++ b/clientconn.go @@ -71,7 +71,7 @@ func (s clientConnState) String() string { // ClientConn is a client-side RTSP connection. type ClientConn struct { - conf ClientConf + c *Client nconn net.Conn isTLS bool br *bufio.Reader @@ -103,42 +103,42 @@ type ClientConn struct { backgroundDone chan struct{} } -func newClientConn(conf ClientConf, scheme string, host string) (*ClientConn, error) { - if conf.TLSConfig == nil { - conf.TLSConfig = &tls.Config{InsecureSkipVerify: true} +func newClientConn(c *Client, scheme string, host string) (*ClientConn, error) { + if c.TLSConfig == nil { + c.TLSConfig = &tls.Config{InsecureSkipVerify: true} } - if conf.ReadTimeout == 0 { - conf.ReadTimeout = 10 * time.Second + if c.ReadTimeout == 0 { + c.ReadTimeout = 10 * time.Second } - if conf.InitialUDPReadTimeout == 0 { - conf.InitialUDPReadTimeout = 3 * time.Second + if c.InitialUDPReadTimeout == 0 { + c.InitialUDPReadTimeout = 3 * time.Second } - if conf.WriteTimeout == 0 { - conf.WriteTimeout = 10 * time.Second + if c.WriteTimeout == 0 { + c.WriteTimeout = 10 * time.Second } - if conf.ReadBufferCount == 0 { - conf.ReadBufferCount = 1 + if c.ReadBufferCount == 0 { + c.ReadBufferCount = 1 } - if conf.ReadBufferSize == 0 { - conf.ReadBufferSize = 2048 + if c.ReadBufferSize == 0 { + c.ReadBufferSize = 2048 } - if conf.DialTimeout == nil { - conf.DialTimeout = net.DialTimeout + if c.DialTimeout == nil { + c.DialTimeout = net.DialTimeout } - if conf.ListenPacket == nil { - conf.ListenPacket = net.ListenPacket + if c.ListenPacket == nil { + c.ListenPacket = net.ListenPacket } - if conf.senderReportPeriod == 0 { - conf.senderReportPeriod = 10 * time.Second + if c.senderReportPeriod == 0 { + c.senderReportPeriod = 10 * time.Second } - if conf.receiverReportPeriod == 0 { - conf.receiverReportPeriod = 10 * time.Second + if c.receiverReportPeriod == 0 { + c.receiverReportPeriod = 10 * time.Second } cc := &ClientConn{ - conf: conf, + c: c, tracks: make(map[int]clientConnTrack), writeError: fmt.Errorf("not running"), } @@ -203,7 +203,7 @@ func (cc *ClientConn) connOpen(scheme string, host string) error { } v := StreamProtocolUDP - if scheme == "rtsps" && cc.conf.StreamProtocol == &v { + if scheme == "rtsps" && cc.c.StreamProtocol == &v { return fmt.Errorf("RTSPS can't be used with UDP") } @@ -211,14 +211,14 @@ func (cc *ClientConn) connOpen(scheme string, host string) error { host += ":554" } - nconn, err := cc.conf.DialTimeout("tcp", host, cc.conf.ReadTimeout) + nconn, err := cc.c.DialTimeout("tcp", host, cc.c.ReadTimeout) if err != nil { return err } conn := func() net.Conn { if scheme == "rtsps" { - return tls.Client(nconn, cc.conf.TLSConfig) + return tls.Client(nconn, cc.c.TLSConfig) } return nconn }() @@ -289,11 +289,11 @@ func (cc *ClientConn) Do(req *base.Request) (*base.Response, error) { // add user agent req.Header["User-Agent"] = base.HeaderValue{"gortsplib"} - if cc.conf.OnRequest != nil { - cc.conf.OnRequest(req) + if cc.c.OnRequest != nil { + cc.c.OnRequest(req) } - cc.nconn.SetWriteDeadline(time.Now().Add(cc.conf.WriteTimeout)) + cc.nconn.SetWriteDeadline(time.Now().Add(cc.c.WriteTimeout)) err := req.Write(cc.bw) if err != nil { return nil, err @@ -304,7 +304,7 @@ func (cc *ClientConn) Do(req *base.Request) (*base.Response, error) { } var res base.Response - cc.nconn.SetReadDeadline(time.Now().Add(cc.conf.ReadTimeout)) + cc.nconn.SetReadDeadline(time.Now().Add(cc.c.ReadTimeout)) if cc.tcpFrameBuffer != nil { // read the response and ignore interleaved frames in between; @@ -322,8 +322,8 @@ func (cc *ClientConn) Do(req *base.Request) (*base.Response, error) { } } - if cc.conf.OnResponse != nil { - cc.conf.OnResponse(&res) + if cc.c.OnResponse != nil { + cc.c.OnResponse(&res) } // get session from response @@ -423,7 +423,7 @@ func (cc *ClientConn) Describe(u *base.URL) (Tracks, *base.Response, error) { if res.StatusCode != base.StatusOK { // redirect - if !cc.conf.RedirectDisable && + if !cc.c.RedirectDisable && res.StatusCode >= base.StatusMovedPermanently && res.StatusCode <= base.StatusUseProxy && len(res.Header["Location"]) == 1 { @@ -530,8 +530,8 @@ func (cc *ClientConn) Setup(mode headers.TransportMode, track *Track, } // protocol set by conf - if cc.conf.StreamProtocol != nil { - return *cc.conf.StreamProtocol + if cc.c.StreamProtocol != nil { + return *cc.c.StreamProtocol } // try UDP @@ -637,7 +637,7 @@ func (cc *ClientConn) Setup(mode headers.TransportMode, track *Track, // switch protocol automatically if res.StatusCode == base.StatusUnsupportedTransport && cc.streamProtocol == nil && - cc.conf.StreamProtocol == nil { + cc.c.StreamProtocol == nil { v := StreamProtocolTCP cc.streamProtocol = &v @@ -668,7 +668,7 @@ func (cc *ClientConn) Setup(mode headers.TransportMode, track *Track, } } - if !cc.conf.AnyPortEnable { + if !cc.c.AnyPortEnable { if thRes.ServerPorts == nil || (thRes.ServerPorts[0] == 0 && thRes.ServerPorts[1] == 0) { rtpListener.close() rtcpListener.close() @@ -732,7 +732,7 @@ func (cc *ClientConn) Setup(mode headers.TransportMode, track *Track, if *cc.streamProtocol == StreamProtocolTCP && cc.tcpFrameBuffer == nil { - cc.tcpFrameBuffer = multibuffer.New(uint64(cc.conf.ReadBufferCount), uint64(cc.conf.ReadBufferSize)) + cc.tcpFrameBuffer = multibuffer.New(uint64(cc.c.ReadBufferCount), uint64(cc.c.ReadBufferSize)) } return res, nil @@ -798,7 +798,7 @@ func (cc *ClientConn) WriteFrame(trackID int, streamType StreamType, payload []b return cc.tracks[trackID].udpRTCPListener.write(payload) } - cc.nconn.SetWriteDeadline(now.Add(cc.conf.WriteTimeout)) + cc.nconn.SetWriteDeadline(now.Add(cc.c.WriteTimeout)) frame := base.InterleavedFrame{ TrackID: trackID, StreamType: streamType, @@ -865,7 +865,7 @@ func (cc *ClientConn) ReadFrames(onFrame func(int, StreamType, []byte)) chan err if *cc.streamProtocol == StreamProtocolUDP && safeState == clientConnStatePlay { if _, ok := err.(liberrors.ErrClientNoUDPPacketsRecently); ok { - if cc.conf.StreamProtocol == nil { + if cc.c.StreamProtocol == nil { prevBaseURL := cc.streamBaseURL oldUseGetParameter := cc.useGetParameter prevTracks := cc.tracks diff --git a/clientconnpublish.go b/clientconnpublish.go index 4ac1828d..efa4e72b 100644 --- a/clientconnpublish.go +++ b/clientconnpublish.go @@ -116,7 +116,7 @@ func (cc *ClientConn) backgroundRecordUDP() error { } }() - reportTicker := time.NewTicker(cc.conf.senderReportPeriod) + reportTicker := time.NewTicker(cc.c.senderReportPeriod) defer reportTicker.Stop() for { @@ -161,7 +161,7 @@ func (cc *ClientConn) backgroundRecordTCP() error { } }() - reportTicker := time.NewTicker(cc.conf.senderReportPeriod) + reportTicker := time.NewTicker(cc.c.senderReportPeriod) defer reportTicker.Stop() for { diff --git a/clientconnpublish_test.go b/clientconnpublish_test.go index 5ec8b38d..c3240034 100644 --- a/clientconnpublish_test.go +++ b/clientconnpublish_test.go @@ -163,7 +163,7 @@ func TestClientPublishSerial(t *testing.T) { require.NoError(t, err) }() - conf := ClientConf{ + c := &Client{ StreamProtocol: func() *StreamProtocol { if proto == "udp" { v := StreamProtocolUDP @@ -177,7 +177,7 @@ func TestClientPublishSerial(t *testing.T) { track, err := NewTrackH264(96, []byte("123456"), []byte("123456")) require.NoError(t, err) - conn, err := conf.DialPublish("rtsp://localhost:8554/teststream", + conn, err := c.DialPublish("rtsp://localhost:8554/teststream", Tracks{track}) require.NoError(t, err) @@ -303,7 +303,7 @@ func TestClientPublishParallel(t *testing.T) { require.NoError(t, err) }() - conf := ClientConf{ + c := &Client{ StreamProtocol: func() *StreamProtocol { if proto == "udp" { v := StreamProtocolUDP @@ -320,7 +320,7 @@ func TestClientPublishParallel(t *testing.T) { writerDone := make(chan struct{}) defer func() { <-writerDone }() - conn, err := conf.DialPublish("rtsp://localhost:8554/teststream", + conn, err := c.DialPublish("rtsp://localhost:8554/teststream", Tracks{track}) require.NoError(t, err) defer conn.Close() @@ -464,7 +464,7 @@ func TestClientPublishPauseSerial(t *testing.T) { require.NoError(t, err) }() - conf := ClientConf{ + c := &Client{ StreamProtocol: func() *StreamProtocol { if proto == "udp" { v := StreamProtocolUDP @@ -478,7 +478,7 @@ func TestClientPublishPauseSerial(t *testing.T) { track, err := NewTrackH264(96, []byte("123456"), []byte("123456")) require.NoError(t, err) - conn, err := conf.DialPublish("rtsp://localhost:8554/teststream", + conn, err := c.DialPublish("rtsp://localhost:8554/teststream", Tracks{track}) require.NoError(t, err) defer conn.Close() @@ -604,7 +604,7 @@ func TestClientPublishPauseParallel(t *testing.T) { require.NoError(t, err) }() - conf := ClientConf{ + c := &Client{ StreamProtocol: func() *StreamProtocol { if proto == "udp" { v := StreamProtocolUDP @@ -618,7 +618,7 @@ func TestClientPublishPauseParallel(t *testing.T) { track, err := NewTrackH264(96, []byte("123456"), []byte("123456")) require.NoError(t, err) - conn, err := conf.DialPublish("rtsp://localhost:8554/teststream", + conn, err := c.DialPublish("rtsp://localhost:8554/teststream", Tracks{track}) require.NoError(t, err) @@ -889,7 +889,7 @@ func TestClientPublishRTCPReport(t *testing.T) { }.Write(bconn.Writer) }() - conf := ClientConf{ + c := &Client{ StreamProtocol: func() *StreamProtocol { v := StreamProtocolTCP return &v @@ -900,7 +900,7 @@ func TestClientPublishRTCPReport(t *testing.T) { track, err := NewTrackH264(96, []byte("123456"), []byte("123456")) require.NoError(t, err) - conn, err := conf.DialPublish("rtsp://localhost:8554/teststream", + conn, err := c.DialPublish("rtsp://localhost:8554/teststream", Tracks{track}) require.NoError(t, err) defer conn.Close() diff --git a/clientconnread.go b/clientconnread.go index fd7fbe20..6000e590 100644 --- a/clientconnread.go +++ b/clientconnread.go @@ -89,14 +89,14 @@ func (cc *ClientConn) backgroundPlayUDP() error { } }() - reportTicker := time.NewTicker(cc.conf.receiverReportPeriod) + reportTicker := time.NewTicker(cc.c.receiverReportPeriod) defer reportTicker.Stop() keepaliveTicker := time.NewTicker(clientConnUDPKeepalivePeriod) defer keepaliveTicker.Stop() checkStreamInitial := true - checkStreamTicker := time.NewTicker(cc.conf.InitialUDPReadTimeout) + checkStreamTicker := time.NewTicker(cc.c.InitialUDPReadTimeout) defer func() { checkStreamTicker.Stop() }() @@ -166,12 +166,12 @@ func (cc *ClientConn) backgroundPlayUDP() error { now := time.Now() for _, cct := range cc.tracks { lft := time.Unix(atomic.LoadInt64(cct.udpRTPListener.lastFrameTime), 0) - if now.Sub(lft) < cc.conf.ReadTimeout { + if now.Sub(lft) < cc.c.ReadTimeout { return false } lft = time.Unix(atomic.LoadInt64(cct.udpRTCPListener.lastFrameTime), 0) - if now.Sub(lft) < cc.conf.ReadTimeout { + if now.Sub(lft) < cc.c.ReadTimeout { return false } } @@ -222,7 +222,7 @@ func (cc *ClientConn) backgroundPlayTCP() error { } }() - reportTicker := time.NewTicker(cc.conf.receiverReportPeriod) + reportTicker := time.NewTicker(cc.c.receiverReportPeriod) defer reportTicker.Stop() checkStreamTicker := time.NewTicker(clientConnCheckStreamPeriod) @@ -239,7 +239,7 @@ func (cc *ClientConn) backgroundPlayTCP() error { now := time.Now() for trackID, cct := range cc.tracks { r := cct.rtcpReceiver.Report(now) - cc.nconn.SetWriteDeadline(time.Now().Add(cc.conf.WriteTimeout)) + cc.nconn.SetWriteDeadline(time.Now().Add(cc.c.WriteTimeout)) frame := base.InterleavedFrame{ TrackID: trackID, StreamType: StreamTypeRTCP, @@ -252,7 +252,7 @@ func (cc *ClientConn) backgroundPlayTCP() error { inTimeout := func() bool { now := time.Now() lft := time.Unix(atomic.LoadInt64(&lastFrameTime), 0) - return now.Sub(lft) >= cc.conf.ReadTimeout + return now.Sub(lft) >= cc.c.ReadTimeout }() if inTimeout { cc.nconn.SetReadDeadline(time.Now()) diff --git a/clientconnread_test.go b/clientconnread_test.go index 532d5384..28dc4832 100644 --- a/clientconnread_test.go +++ b/clientconnread_test.go @@ -357,7 +357,7 @@ func TestClientRead(t *testing.T) { require.NoError(t, err) }() - conf := ClientConf{ + c := &Client{ StreamProtocol: func() *StreamProtocol { if ca.proto == "udp" { v := StreamProtocolUDP @@ -368,7 +368,7 @@ func TestClientRead(t *testing.T) { }(), } - conn, err := conf.DialRead(scheme + "://localhost:8554/teststream") + conn, err := c.DialRead(scheme + "://localhost:8554/teststream") require.NoError(t, err) done := conn.ReadFrames(func(id int, streamType StreamType, payload []byte) { @@ -593,11 +593,11 @@ func TestClientReadAnyPort(t *testing.T) { }) }() - conf := ClientConf{ + c := &Client{ AnyPortEnable: true, } - conn, err := conf.DialRead("rtsp://localhost:8554/teststream") + conn, err := c.DialRead("rtsp://localhost:8554/teststream") require.NoError(t, err) frameRecv := make(chan struct{}) @@ -877,11 +877,11 @@ func TestClientReadAutomaticProtocol(t *testing.T) { conn.Close() }() - conf := ClientConf{ + c := &Client{ ReadTimeout: 1 * time.Second, } - conn, err := conf.DialRead("rtsp://localhost:8554/teststream") + conn, err := c.DialRead("rtsp://localhost:8554/teststream") require.NoError(t, err) frameRecv := make(chan struct{}) @@ -1213,7 +1213,7 @@ func TestClientReadPause(t *testing.T) { require.NoError(t, err) }() - conf := ClientConf{ + c := &Client{ StreamProtocol: func() *StreamProtocol { if proto == "udp" { v := StreamProtocolUDP @@ -1224,7 +1224,7 @@ func TestClientReadPause(t *testing.T) { }(), } - conn, err := conf.DialRead("rtsp://localhost:8554/teststream") + conn, err := c.DialRead("rtsp://localhost:8554/teststream") require.NoError(t, err) firstFrame := int32(0) @@ -1398,7 +1398,7 @@ func TestClientReadRTCPReport(t *testing.T) { require.NoError(t, err) }() - conf := ClientConf{ + c := &Client{ StreamProtocol: func() *StreamProtocol { v := StreamProtocolTCP return &v @@ -1406,7 +1406,7 @@ func TestClientReadRTCPReport(t *testing.T) { receiverReportPeriod: 1 * time.Second, } - conn, err := conf.DialRead("rtsp://localhost:8554/teststream") + conn, err := c.DialRead("rtsp://localhost:8554/teststream") require.NoError(t, err) recv := 0 @@ -1552,7 +1552,7 @@ func TestClientReadErrorTimeout(t *testing.T) { require.NoError(t, err) }() - conf := ClientConf{ + c := &Client{ StreamProtocol: func() *StreamProtocol { switch proto { case "udp": @@ -1569,7 +1569,7 @@ func TestClientReadErrorTimeout(t *testing.T) { ReadTimeout: 1 * time.Second, } - conn, err := conf.DialRead("rtsp://localhost:8554/teststream") + conn, err := c.DialRead("rtsp://localhost:8554/teststream") require.NoError(t, err) defer conn.Close() @@ -1694,14 +1694,14 @@ func TestClientReadIgnoreTCPInvalidTrack(t *testing.T) { require.NoError(t, err) }() - conf := ClientConf{ + c := &Client{ StreamProtocol: func() *StreamProtocol { v := StreamProtocolTCP return &v }(), } - conn, err := conf.DialRead("rtsp://localhost:8554/teststream") + conn, err := c.DialRead("rtsp://localhost:8554/teststream") require.NoError(t, err) defer conn.Close() diff --git a/clientconnudpl.go b/clientconnudpl.go index ff8b6781..5267ed61 100644 --- a/clientconnudpl.go +++ b/clientconnudpl.go @@ -30,7 +30,7 @@ type clientConnUDPListener struct { } func newClientConnUDPListener(cc *ClientConn, port int) (*clientConnUDPListener, error) { - pc, err := cc.conf.ListenPacket("udp", ":"+strconv.FormatInt(int64(port), 10)) + pc, err := cc.c.ListenPacket("udp", ":"+strconv.FormatInt(int64(port), 10)) if err != nil { return nil, err } @@ -43,7 +43,7 @@ func newClientConnUDPListener(cc *ClientConn, port int) (*clientConnUDPListener, return &clientConnUDPListener{ cc: cc, pc: pc, - frameBuffer: multibuffer.New(uint64(cc.conf.ReadBufferCount), uint64(cc.conf.ReadBufferSize)), + frameBuffer: multibuffer.New(uint64(cc.c.ReadBufferCount), uint64(cc.c.ReadBufferSize)), lastFrameTime: func() *int64 { v := int64(0) return &v @@ -114,7 +114,7 @@ func (l *clientConnUDPListener) run() { } func (l *clientConnUDPListener) write(buf []byte) error { - l.pc.SetWriteDeadline(time.Now().Add(l.cc.conf.WriteTimeout)) + l.pc.SetWriteDeadline(time.Now().Add(l.cc.c.WriteTimeout)) _, err := l.pc.WriteTo(buf, &net.UDPAddr{ IP: l.remoteIP, Zone: l.remoteZone, diff --git a/examples/client-publish-options/main.go b/examples/client-publish-options/main.go index 05048834..78750c81 100644 --- a/examples/client-publish-options/main.go +++ b/examples/client-publish-options/main.go @@ -41,8 +41,8 @@ func main() { panic(err) } - // ClientConf allows to set additional client options - conf := gortsplib.ClientConf{ + // Client allows to set additional client options + c := &gortsplib.Client{ // the stream protocol (UDP or TCP). If nil, it is chosen automatically StreamProtocol: nil, // timeout of read operations @@ -52,7 +52,7 @@ func main() { } // connect to the server and start publishing the track - conn, err := conf.DialPublish("rtsp://localhost:8554/mystream", + conn, err := c.DialPublish("rtsp://localhost:8554/mystream", gortsplib.Tracks{track}) if err != nil { panic(err) diff --git a/examples/client-read-options/main.go b/examples/client-read-options/main.go index ec1fd0a9..77a9110c 100644 --- a/examples/client-read-options/main.go +++ b/examples/client-read-options/main.go @@ -12,8 +12,8 @@ import ( // 2. connect to a RTSP server and read all tracks on a path func main() { - // ClientConf allows to set additional client options - conf := gortsplib.ClientConf{ + // Client allows to set additional client options + c := &gortsplib.Client{ // the stream protocol (UDP or TCP). If nil, it is chosen automatically StreamProtocol: nil, // timeout of read operations @@ -23,7 +23,7 @@ func main() { } // connect to the server and start reading all tracks - conn, err := conf.DialRead("rtsp://localhost:8554/mystream") + conn, err := c.DialRead("rtsp://localhost:8554/mystream") if err != nil { panic(err) }