server: shut down session after a TEARDOWN request

pkg/liberrors/server.go

@@ -12,7 +12,7 @@ type ErrServerTCPFramesEnable struct{}
 
 // Error implements the error interface.
 func (e ErrServerTCPFramesEnable) Error() string {
-	return ""
+	return "tcp frame enable"
 }
 
 // ErrServerTCPFramesDisable is an error that can be returned by a server.
@@ -20,7 +20,7 @@ type ErrServerTCPFramesDisable struct{}
 
 // Error implements the error interface.
 func (e ErrServerTCPFramesDisable) Error() string {
-	return ""
+	return "tcp frame disable"
 }
 
 // ErrServerCSeqMissing is an error that can be returned by a server.
@@ -183,3 +183,19 @@ type ErrServerLinkedToOtherSession struct{}
 func (e ErrServerLinkedToOtherSession) Error() string {
 	return "connection is linked to another session"
 }
+
+// ErrServerTeardown is an error that can be returned by a server.
+type ErrServerTeardown struct{}
+
+// Error implements the error interface.
+func (e ErrServerTeardown) Error() string {
+	return "teardown"
+}
+
+// ErrServerSessionLinkedToOtherConn is an error that can be returned by a server.
+type ErrServerSessionLinkedToOtherConn struct{}
+
+// Error implements the error interface.
+func (e ErrServerSessionLinkedToOtherConn) Error() string {
+	return "session is linked to another connection"
+}

server.go

@@ -42,8 +42,9 @@ func newSessionID(sessions map[string]*ServerSession) (string, error) {
 }
 
 type sessionGetReq struct {
-	id  string
+	id     string
-	res chan *ServerSession
+	create bool
+	res    chan *ServerSession
 }
 
 // Server is a RTSP server.
@@ -233,6 +234,11 @@ outer:
 				req.res <- ss
 
 			} else {
+				if !req.create {
+					req.res <- nil
+					continue
+				}
+
 				id, err := newSessionID(s.sessions)
 				if err != nil {
 					req.res <- nil

server_publish_test.go

@@ -482,10 +482,26 @@ func TestServerPublish(t *testing.T) {
 		"tcp",
 	} {
 		t.Run(proto, func(t *testing.T) {
+			connOpened := make(chan struct{})
+			connClosed := make(chan struct{})
+			sessionOpened := make(chan struct{})
+			sessionClosed := make(chan struct{})
 			rtpReceived := uint64(0)
 
 			s := &Server{
 				Handler: &testServerHandler{
+					onConnOpen: func(sc *ServerConn) {
+						close(connOpened)
+					},
+					onConnClose: func(sc *ServerConn, err error) {
+						close(connClosed)
+					},
+					onSessionOpen: func(ss *ServerSession) {
+						close(sessionOpened)
+					},
+					onSessionClose: func(ss *ServerSession) {
+						close(sessionClosed)
+					},
 					onAnnounce: func(ctx *ServerHandlerOnAnnounceCtx) (*base.Response, error) {
 						return &base.Response{
 							StatusCode: base.StatusOK,
@@ -531,6 +547,8 @@ func TestServerPublish(t *testing.T) {
 			defer conn.Close()
 			bconn := bufio.NewReadWriter(bufio.NewReader(conn), bufio.NewWriter(conn))
 
+			<-connOpened
+
 			track, err := NewTrackH264(96, []byte("123456"), []byte("123456"))
 			require.NoError(t, err)
 
@@ -558,6 +576,8 @@ func TestServerPublish(t *testing.T) {
 			require.NoError(t, err)
 			require.Equal(t, base.StatusOK, res.StatusCode)
 
+			<-sessionOpened
+
 			inTH := &headers.Transport{
 				Delivery: func() *base.StreamDelivery {
 					v := base.StreamDeliveryUnicast
@@ -674,6 +694,25 @@ func TestServerPublish(t *testing.T) {
 				require.Equal(t, StreamTypeRTCP, f.StreamType)
 				require.Equal(t, []byte{0x09, 0x0A, 0x0B, 0x0C}, f.Payload)
 			}
+
+			err = base.Request{
+				Method: base.Teardown,
+				URL:    base.MustParseURL("rtsp://localhost:8554/teststream"),
+				Header: base.Header{
+					"CSeq":    base.HeaderValue{"3"},
+					"Session": res.Header["Session"],
+				},
+			}.Write(bconn.Writer)
+			require.NoError(t, err)
+
+			err = res.Read(bconn.Reader)
+			require.NoError(t, err)
+			require.Equal(t, base.StatusOK, res.StatusCode)
+
+			<-sessionClosed
+
+			conn.Close()
+			<-connClosed
 		})
 	}
 }

server_read_test.go

@@ -116,12 +116,13 @@ func TestServerReadSetupPath(t *testing.T) {
 }
 
 func TestServerReadSetupErrorDifferentPaths(t *testing.T) {
-	serverErr := make(chan error)
+	connClosed := make(chan struct{})
 
 	s := &Server{
 		Handler: &testServerHandler{
 			onConnClose: func(sc *ServerConn, err error) {
-				serverErr <- err
+				require.Equal(t, "can't setup tracks with different paths", err.Error())
+				close(connClosed)
 			},
 			onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, error) {
 				return &base.Response{
@@ -185,17 +186,17 @@ func TestServerReadSetupErrorDifferentPaths(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, base.StatusBadRequest, res.StatusCode)
 
-	err = <-serverErr
+	<-connClosed
-	require.Equal(t, "can't setup tracks with different paths", err.Error())
 }
 
 func TestServerReadSetupErrorTrackTwice(t *testing.T) {
-	serverErr := make(chan error)
+	connClosed := make(chan struct{})
 
 	s := &Server{
 		Handler: &testServerHandler{
 			onConnClose: func(sc *ServerConn, err error) {
-				serverErr <- err
+				require.Equal(t, "track 0 has already been setup", err.Error())
+				close(connClosed)
 			},
 			onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, error) {
 				return &base.Response{
@@ -259,8 +260,7 @@ func TestServerReadSetupErrorTrackTwice(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, base.StatusBadRequest, res.StatusCode)
 
-	err = <-serverErr
+	<-connClosed
-	require.Equal(t, "track 0 has already been setup", err.Error())
 }
 
 func TestServerRead(t *testing.T) {
@@ -269,10 +269,26 @@ func TestServerRead(t *testing.T) {
 		"tcp",
 	} {
 		t.Run(proto, func(t *testing.T) {
+			connOpened := make(chan struct{})
+			connClosed := make(chan struct{})
+			sessionOpened := make(chan struct{})
+			sessionClosed := make(chan struct{})
 			framesReceived := make(chan struct{})
 
 			s := &Server{
 				Handler: &testServerHandler{
+					onConnOpen: func(sc *ServerConn) {
+						close(connOpened)
+					},
+					onConnClose: func(sc *ServerConn, err error) {
+						close(connClosed)
+					},
+					onSessionOpen: func(ss *ServerSession) {
+						close(sessionOpened)
+					},
+					onSessionClose: func(ss *ServerSession) {
+						close(sessionClosed)
+					},
 					onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, error) {
 						return &base.Response{
 							StatusCode: base.StatusOK,
@@ -303,9 +319,10 @@ func TestServerRead(t *testing.T) {
 
 			conn, err := net.Dial("tcp", "localhost:8554")
 			require.NoError(t, err)
-			defer conn.Close()
 			bconn := bufio.NewReadWriter(bufio.NewReader(conn), bufio.NewWriter(conn))
 
+			<-connOpened
+
 			inTH := &headers.Transport{
 				Delivery: func() *base.StreamDelivery {
 					v := base.StreamDeliveryUnicast
@@ -344,6 +361,8 @@ func TestServerRead(t *testing.T) {
 			err = th.Read(res.Header["Transport"])
 			require.NoError(t, err)
 
+			<-sessionOpened
+
 			var l1 net.PacketConn
 			var l2 net.PacketConn
 			if proto == "udp" {
@@ -415,6 +434,25 @@ func TestServerRead(t *testing.T) {
 			}
 
 			<-framesReceived
+
+			err = base.Request{
+				Method: base.Teardown,
+				URL:    base.MustParseURL("rtsp://localhost:8554/teststream"),
+				Header: base.Header{
+					"CSeq":    base.HeaderValue{"3"},
+					"Session": res.Header["Session"],
+				},
+			}.Write(bconn.Writer)
+			require.NoError(t, err)
+
+			err = res.Read(bconn.Reader)
+			require.NoError(t, err)
+			require.Equal(t, base.StatusOK, res.StatusCode)
+
+			<-sessionClosed
+
+			conn.Close()
+			<-connClosed
 		})
 	}
 }

server_test.go

@@ -15,7 +15,9 @@ import (
 )
 
 type testServerHandler struct {
+	onConnOpen     func(*ServerConn)
 	onConnClose    func(*ServerConn, error)
+	onSessionOpen  func(*ServerSession)
 	onSessionClose func(*ServerSession)
 	onDescribe     func(*ServerHandlerOnDescribeCtx) (*base.Response, []byte, error)
 	onAnnounce     func(*ServerHandlerOnAnnounceCtx) (*base.Response, error)
@@ -26,12 +28,24 @@ type testServerHandler struct {
 	onFrame        func(*ServerHandlerOnFrameCtx)
 }
 
+func (sh *testServerHandler) OnConnOpen(sc *ServerConn) {
+	if sh.onConnOpen != nil {
+		sh.onConnOpen(sc)
+	}
+}
+
 func (sh *testServerHandler) OnConnClose(sc *ServerConn, err error) {
 	if sh.onConnClose != nil {
 		sh.onConnClose(sc, err)
 	}
 }
 
+func (sh *testServerHandler) OnSessionOpen(ss *ServerSession) {
+	if sh.onSessionOpen != nil {
+		sh.onSessionOpen(ss)
+	}
+}
+
 func (sh *testServerHandler) OnSessionClose(ss *ServerSession) {
 	if sh.onSessionClose != nil {
 		sh.onSessionClose(ss)

serverconn.go

@@ -307,7 +307,7 @@ func (sc *ServerConn) handleRequest(req *base.Request) (*base.Response, error) {
 	case base.Announce:
 		if _, ok := sc.s.Handler.(ServerHandlerOnAnnounce); ok {
 			sres := make(chan *ServerSession)
-			sc.s.sessionGet <- sessionGetReq{id: sxID, res: sres}
+			sc.s.sessionGet <- sessionGetReq{id: sxID, create: true, res: sres}
 			ss := <-sres
 
 			if ss == nil {
@@ -326,7 +326,7 @@ func (sc *ServerConn) handleRequest(req *base.Request) (*base.Response, error) {
 	case base.Setup:
 		if _, ok := sc.s.Handler.(ServerHandlerOnSetup); ok {
 			sres := make(chan *ServerSession)
-			sc.s.sessionGet <- sessionGetReq{id: sxID, res: sres}
+			sc.s.sessionGet <- sessionGetReq{id: sxID, create: true, res: sres}
 			ss := <-sres
 
 			if ss == nil {
@@ -345,7 +345,7 @@ func (sc *ServerConn) handleRequest(req *base.Request) (*base.Response, error) {
 	case base.Play:
 		if _, ok := sc.s.Handler.(ServerHandlerOnPlay); ok {
 			sres := make(chan *ServerSession)
-			sc.s.sessionGet <- sessionGetReq{id: sxID, res: sres}
+			sc.s.sessionGet <- sessionGetReq{id: sxID, create: false, res: sres}
 			ss := <-sres
 
 			if ss == nil {
@@ -371,7 +371,7 @@ func (sc *ServerConn) handleRequest(req *base.Request) (*base.Response, error) {
 	case base.Record:
 		if _, ok := sc.s.Handler.(ServerHandlerOnRecord); ok {
 			sres := make(chan *ServerSession)
-			sc.s.sessionGet <- sessionGetReq{id: sxID, res: sres}
+			sc.s.sessionGet <- sessionGetReq{id: sxID, create: false, res: sres}
 			ss := <-sres
 
 			if ss == nil {
@@ -397,7 +397,7 @@ func (sc *ServerConn) handleRequest(req *base.Request) (*base.Response, error) {
 	case base.Pause:
 		if _, ok := sc.s.Handler.(ServerHandlerOnPause); ok {
 			sres := make(chan *ServerSession)
-			sc.s.sessionGet <- sessionGetReq{id: sxID, res: sres}
+			sc.s.sessionGet <- sessionGetReq{id: sxID, create: false, res: sres}
 			ss := <-sres
 
 			if ss == nil {
@@ -419,24 +419,22 @@ func (sc *ServerConn) handleRequest(req *base.Request) (*base.Response, error) {
 		}
 
 	case base.Teardown:
-		if _, ok := sc.s.Handler.(ServerHandlerOnTeardown); ok {
+		sres := make(chan *ServerSession)
-			sres := make(chan *ServerSession)
+		sc.s.sessionGet <- sessionGetReq{id: sxID, create: false, res: sres}
-			sc.s.sessionGet <- sessionGetReq{id: sxID, res: sres}
+		ss := <-sres
-			ss := <-sres
 
-			if ss == nil {
+		if ss == nil {
-				return &base.Response{
+			return &base.Response{
-					StatusCode: base.StatusBadRequest,
+				StatusCode: base.StatusBadRequest,
-				}, fmt.Errorf("terminated")
+			}, fmt.Errorf("terminated")
-			}
-
-			rres := make(chan requestRes)
-			ss.request <- requestReq{sc: sc, req: req, res: rres}
-			res := <-rres
-
-			return res.res, res.err
 		}
 
+		rres := make(chan requestRes)
+		ss.request <- requestReq{sc: sc, req: req, res: rres}
+		res := <-rres
+
+		return res.res, res.err
+
 	case base.GetParameter:
 		if h, ok := sc.s.Handler.(ServerHandlerOnGetParameter); ok {
 			pathAndQuery, ok := req.URL.RTSPPath()

serverhandler.go

@@ -164,20 +164,6 @@ type ServerHandlerOnSetParameter interface {
 	OnSetParameter(*ServerHandlerOnSetParameterCtx) (*base.Response, error)
 }
 
-// ServerHandlerOnTeardownCtx is the context of a TEARDOWN request.
-type ServerHandlerOnTeardownCtx struct {
-	Session *ServerSession
-	Conn    *ServerConn
-	Req     *base.Request
-	Path    string
-	Query   string
-}
-
-// ServerHandlerOnTeardown can be implemented by a ServerHandler.
-type ServerHandlerOnTeardown interface {
-	OnTeardown(*ServerHandlerOnTeardownCtx) (*base.Response, error)
-}
-
 // ServerHandlerOnFrameCtx is the context of a frame request.
 type ServerHandlerOnFrameCtx struct {
 	Session    *ServerSession

serversession.go

@@ -218,6 +218,12 @@ outer:
 		select {
 		case req := <-ss.request:
 			res, err := ss.handleRequest(req.sc, req.req)
+
+			if _, ok := err.(liberrors.ErrServerTeardown); ok {
+				req.res <- requestRes{res, nil}
+				break outer
+			}
+
 			req.res <- requestRes{res, err}
 
 		case <-checkStreamTicker.C:
@@ -289,6 +295,10 @@ outer:
 }
 
 func (ss *ServerSession) handleRequest(sc *ServerConn, req *base.Request) (*base.Response, error) {
+	if ss.linkedConn != nil && sc != ss.linkedConn {
+		return nil, liberrors.ErrServerSessionLinkedToOtherConn{}
+	}
+
 	switch req.Method {
 	case base.Announce:
 		err := ss.checkState(map[ServerSessionState]struct{}{
@@ -772,22 +782,11 @@ func (ss *ServerSession) handleRequest(sc *ServerConn, req *base.Request) (*base
 		return res, err
 
 	case base.Teardown:
-		pathAndQuery, ok := req.URL.RTSPPath()
+		ss.linkedConn = nil
-		if !ok {
-			return &base.Response{
-				StatusCode: base.StatusBadRequest,
-			}, liberrors.ErrServerNoPath{}
-		}
 
-		path, query := base.PathSplitQuery(pathAndQuery)
+		return &base.Response{
-
+			StatusCode: base.StatusOK,
-		return ss.s.Handler.(ServerHandlerOnTeardown).OnTeardown(&ServerHandlerOnTeardownCtx{
+		}, liberrors.ErrServerTeardown{}
-			Session: ss,
-			Conn:    sc,
-			Req:     req,
-			Path:    path,
-			Query:   query,
-		})
 	}
 
 	return nil, fmt.Errorf("unimplemented")