api: add path to RTMP connections, RTSP sessions, WebRTC sessions (#1962) (#2022)

* api: add path to rtmp response

* add 'path' to RTSP and WebRTC sessions too

* add tests

---------

Co-authored-by: aler9 <46489434+aler9@users.noreply.github.com>

apidocs/openapi.yaml

@@ -389,6 +389,8 @@ components:
         state:
           type: string
           enum: [idle, read, publish]
+        path:
+          type: string
         bytesReceived:
           type: integer
           format: int64
@@ -408,6 +410,8 @@ components:
         state:
           type: string
           enum: [idle, read, publish]
+        path:
+          type: string
         bytesReceived:
           type: integer
           format: int64
@@ -496,6 +500,8 @@ components:
         state:
           type: string
           enum: [read, publish]
+        path:
+          type: string
         bytesReceived:
           type: integer
           format: int64

internal/core/api_defs.go

@@ -57,6 +57,7 @@ type apiRTMPConn struct {
 	Created       time.Time `json:"created"`
 	RemoteAddr    string    `json:"remoteAddr"`
 	State         string    `json:"state"`
+	Path          string    `json:"path"`
 	BytesReceived uint64    `json:"bytesReceived"`
 	BytesSent     uint64    `json:"bytesSent"`
 }
@@ -72,6 +73,7 @@ type apiRTSPSession struct {
 	Created       time.Time `json:"created"`
 	RemoteAddr    string    `json:"remoteAddr"`
 	State         string    `json:"state"`
+	Path          string    `json:"path"`
 	BytesReceived uint64    `json:"bytesReceived"`
 	BytesSent     uint64    `json:"bytesSent"`
 }
@@ -90,6 +92,7 @@ type apiWebRTCSession struct {
 	LocalCandidate            string    `json:"localCandidate"`
 	RemoteCandidate           string    `json:"remoteCandidate"`
 	State                     string    `json:"state"`
+	Path                      string    `json:"path"`
 	BytesReceived             uint64    `json:"bytesReceived"`
 	BytesSent                 uint64    `json:"bytesSent"`
 }

internal/core/api_test.go

@@ -685,25 +685,33 @@ func TestAPIProtocolList(t *testing.T) {
 					pa = "rtmpsconns"
 				}
 
+				type item struct {
+					State string `json:"state"`
+					Path  string `json:"path"`
+				}
+
 				var out struct {
 					ItemCount int    `json:"itemCount"`
-					Items     []struct {
+					Items     []item `json:"items"`
-						State string `json:"state"`
-					} `json:"items"`
 				}
 				httpRequest(t, hc, http.MethodGet, "http://localhost:9997/v2/"+pa+"/list", nil, &out)
 
 				if ca != "rtsp conns" && ca != "rtsps conns" {
-					require.Equal(t, "publish", out.Items[0].State)
+					require.Equal(t, item{
+						State: "publish",
+						Path:  "mypath",
+					}, out.Items[0])
 				}
 
 			case "hls":
-				var out struct {
+				type item struct {
-					ItemCount int `json:"itemCount"`
-					Items     []struct {
 					Created     string `json:"created"`
 					LastRequest string `json:"lastRequest"`
-					} `json:"items"`
+				}
+
+				var out struct {
+					ItemCount int    `json:"itemCount"`
+					Items     []item `json:"items"`
 				}
 				httpRequest(t, hc, http.MethodGet, "http://localhost:9997/v2/hlsmuxers/list", nil, &out)
 
@@ -713,13 +721,9 @@ func TestAPIProtocolList(t *testing.T) {
 
 			case "webrtc":
 				type item struct {
-					Created                   time.Time `json:"created"`
-					RemoteAddr                string    `json:"remoteAddr"`
 					PeerConnectionEstablished bool   `json:"peerConnectionEstablished"`
-					LocalCandidate            string    `json:"localCandidate"`
+					State                     string `json:"state"`
-					RemoteCandidate           string    `json:"remoteCandidate"`
+					Path                      string `json:"path"`
-					BytesReceived             uint64    `json:"bytesReceived"`
-					BytesSent                 uint64    `json:"bytesSent"`
 				}
 
 				var out struct {
@@ -728,7 +732,11 @@ func TestAPIProtocolList(t *testing.T) {
 				}
 				httpRequest(t, hc, http.MethodGet, "http://localhost:9997/v2/webrtcsessions/list", nil, &out)
 
-				require.Equal(t, true, out.Items[0].PeerConnectionEstablished)
+				require.Equal(t, item{
+					PeerConnectionEstablished: true,
+					State:                     "read",
+					Path:                      "mypath",
+				}, out.Items[0])
 			}
 		})
 	}

internal/core/rtmp_conn.go

@@ -214,8 +214,9 @@ type rtmpConn struct {
 	ctxCancel func()
 	uuid      uuid.UUID
 	created   time.Time
+	mutex     sync.Mutex
 	state     rtmpConnState
-	stateMutex sync.Mutex
+	pathName  string
 }
 
 func newRTMPConn(
@@ -279,12 +280,6 @@ func (c *rtmpConn) ip() net.IP {
 	return c.nconn.RemoteAddr().(*net.TCPAddr).IP
 }
 
-func (c *rtmpConn) safeState() rtmpConnState {
-	c.stateMutex.Lock()
-	defer c.stateMutex.Unlock()
-	return c.state
-}
-
 func (c *rtmpConn) run() {
 	defer c.wg.Done()
 
@@ -380,9 +375,10 @@ func (c *rtmpConn) runRead(ctx context.Context, u *url.URL) error {
 
 	defer res.path.readerRemove(pathReaderRemoveReq{author: c})
 
-	c.stateMutex.Lock()
+	c.mutex.Lock()
 	c.state = rtmpConnStateRead
-	c.stateMutex.Unlock()
+	c.pathName = pathName
+	c.mutex.Unlock()
 
 	ringBuffer, _ := ringbuffer.New(uint64(c.readBufferCount))
 	go func() {
@@ -794,9 +790,10 @@ func (c *rtmpConn) runPublish(u *url.URL) error {
 
 	defer res.path.publisherRemove(pathPublisherRemoveReq{author: c})
 
-	c.stateMutex.Lock()
+	c.mutex.Lock()
 	c.state = rtmpConnStatePublish
-	c.stateMutex.Unlock()
+	c.pathName = pathName
+	c.mutex.Unlock()
 
 	videoFormat, audioFormat, err := c.conn.ReadTracks()
 	if err != nil {
@@ -892,12 +889,15 @@ func (c *rtmpConn) apiSourceDescribe() pathAPISourceOrReader {
 }
 
 func (c *rtmpConn) apiItem() *apiRTMPConn {
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
+
 	return &apiRTMPConn{
 		ID:         c.uuid,
 		Created:    c.created,
 		RemoteAddr: c.remoteAddr().String(),
 		State: func() string {
-			switch c.safeState() {
+			switch c.state {
 			case rtmpConnStateRead:
 				return "read"
 
@@ -906,6 +906,7 @@ func (c *rtmpConn) apiItem() *apiRTMPConn {
 			}
 			return "idle"
 		}(),
+		Path:          c.pathName,
 		BytesReceived: c.conn.BytesReceived(),
 		BytesSent:     c.conn.BytesSent(),
 	}

internal/core/rtsp_session.go

@@ -41,9 +41,10 @@ type rtspSession struct {
 	created   time.Time
 	path      *path
 	stream    *stream
-	state      gortsplib.ServerSessionState
-	stateMutex sync.Mutex
 	onReadCmd *externalcmd.Cmd // read
+	mutex     sync.Mutex
+	state     gortsplib.ServerSessionState
+	pathName  string
 }
 
 func newRTSPSession(
@@ -77,12 +78,6 @@ func (s *rtspSession) close() {
 	s.session.Close()
 }
 
-func (s *rtspSession) safeState() gortsplib.ServerSessionState {
-	s.stateMutex.Lock()
-	defer s.stateMutex.Unlock()
-	return s.state
-}
-
 func (s *rtspSession) remoteAddr() net.Addr {
 	return s.author.NetConn().RemoteAddr()
 }
@@ -157,9 +152,10 @@ func (s *rtspSession) onAnnounce(c *rtspConn, ctx *gortsplib.ServerHandlerOnAnno
 
 	s.path = res.path
 
-	s.stateMutex.Lock()
+	s.mutex.Lock()
 	s.state = gortsplib.ServerSessionStatePreRecord
-	s.stateMutex.Unlock()
+	s.pathName = ctx.Path
+	s.mutex.Unlock()
 
 	return &base.Response{
 		StatusCode: base.StatusOK,
@@ -242,9 +238,10 @@ func (s *rtspSession) onSetup(c *rtspConn, ctx *gortsplib.ServerHandlerOnSetupCt
 		s.path = res.path
 		s.stream = res.stream
 
-		s.stateMutex.Lock()
+		s.mutex.Lock()
 		s.state = gortsplib.ServerSessionStatePrePlay
-		s.stateMutex.Unlock()
+		s.pathName = ctx.Path
+		s.mutex.Unlock()
 
 		return &base.Response{
 			StatusCode: base.StatusOK,
@@ -281,9 +278,9 @@ func (s *rtspSession) onPlay(_ *gortsplib.ServerHandlerOnPlayCtx) (*base.Respons
 				})
 		}
 
-		s.stateMutex.Lock()
+		s.mutex.Lock()
 		s.state = gortsplib.ServerSessionStatePlay
-		s.stateMutex.Unlock()
+		s.mutex.Unlock()
 	}
 
 	return &base.Response{
@@ -323,9 +320,9 @@ func (s *rtspSession) onRecord(ctx *gortsplib.ServerHandlerOnRecordCtx) (*base.R
 		}
 	}
 
-	s.stateMutex.Lock()
+	s.mutex.Lock()
 	s.state = gortsplib.ServerSessionStateRecord
-	s.stateMutex.Unlock()
+	s.mutex.Unlock()
 
 	return &base.Response{
 		StatusCode: base.StatusOK,
@@ -341,16 +338,16 @@ func (s *rtspSession) onPause(_ *gortsplib.ServerHandlerOnPauseCtx) (*base.Respo
 			s.onReadCmd.Close()
 		}
 
-		s.stateMutex.Lock()
+		s.mutex.Lock()
 		s.state = gortsplib.ServerSessionStatePrePlay
-		s.stateMutex.Unlock()
+		s.mutex.Unlock()
 
 	case gortsplib.ServerSessionStateRecord:
 		s.path.publisherStop(pathPublisherStopReq{author: s})
 
-		s.stateMutex.Lock()
+		s.mutex.Lock()
 		s.state = gortsplib.ServerSessionStatePreRecord
-		s.stateMutex.Unlock()
+		s.mutex.Unlock()
 	}
 
 	return &base.Response{
@@ -387,12 +384,15 @@ func (s *rtspSession) onDecodeError(ctx *gortsplib.ServerHandlerOnDecodeErrorCtx
 }
 
 func (s *rtspSession) apiItem() *apiRTSPSession {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
+
 	return &apiRTSPSession{
 		ID:         s.uuid,
 		Created:    s.created,
 		RemoteAddr: s.remoteAddr().String(),
 		State: func() string {
-			switch s.safeState() {
+			switch s.state {
 			case gortsplib.ServerSessionStatePrePlay,
 				gortsplib.ServerSessionStatePlay:
 				return "read"
@@ -403,6 +403,7 @@ func (s *rtspSession) apiItem() *apiRTSPSession {
 			}
 			return "idle"
 		}(),
+		Path:          s.pathName,
 		BytesReceived: s.session.BytesReceived(),
 		BytesSent:     s.session.BytesSent(),
 	}

internal/core/webrtc_session.go

@@ -126,9 +126,9 @@ type webRTCSession struct {
 	created    time.Time
 	uuid       uuid.UUID
 	secret     uuid.UUID
-	pcMutex    sync.RWMutex
-	pc         *peerConnection
 	answerSent bool
+	mutex      sync.RWMutex
+	pc         *peerConnection
 
 	chAddRemoteCandidates chan webRTCSessionAddCandidatesReq
 }
@@ -180,12 +180,6 @@ func (s *webRTCSession) close() {
 	s.ctxCancel()
 }
 
-func (s *webRTCSession) safePC() *peerConnection {
-	s.pcMutex.RLock()
-	defer s.pcMutex.RUnlock()
-	return s.pc
-}
-
 func (s *webRTCSession) run() {
 	defer s.wg.Done()
 
@@ -491,9 +485,9 @@ outer:
 		}
 	}
 
-	s.pcMutex.Lock()
+	s.mutex.Lock()
 	s.pc = pc
-	s.pcMutex.Unlock()
+	s.mutex.Unlock()
 
 	return nil
 }
@@ -542,19 +536,21 @@ func (s *webRTCSession) apiReaderDescribe() pathAPISourceOrReader {
 }
 
 func (s *webRTCSession) apiItem() *apiWebRTCSession {
+	s.mutex.RLock()
+	defer s.mutex.RUnlock()
+
 	peerConnectionEstablished := false
 	localCandidate := ""
 	remoteCandidate := ""
 	bytesReceived := uint64(0)
 	bytesSent := uint64(0)
 
-	pc := s.safePC()
+	if s.pc != nil {
-	if pc != nil {
 		peerConnectionEstablished = true
-		localCandidate = pc.localCandidate()
+		localCandidate = s.pc.localCandidate()
-		remoteCandidate = pc.remoteCandidate()
+		remoteCandidate = s.pc.remoteCandidate()
-		bytesReceived = pc.bytesReceived()
+		bytesReceived = s.pc.bytesReceived()
-		bytesSent = pc.bytesSent()
+		bytesSent = s.pc.bytesSent()
 	}
 
 	return &apiWebRTCSession{
@@ -570,6 +566,7 @@ func (s *webRTCSession) apiItem() *apiWebRTCSession {
 			}
 			return "read"
 		}(),
+		Path:          s.req.pathName,
 		BytesReceived: bytesReceived,
 		BytesSent:     bytesSent,
 	}