Browse Source

api, metrics: add endpoints and metrics for RTSP connections (#1233)

new API endpoints:

* /v1/rtspconns/list
* /v1/rtspsconns/list

new metrics:

* rtsp_conns
* rtsps_conns
pull/1234/head
Alessandro Ros 3 years ago committed by GitHub
parent
commit
4ac175d3cc
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
  1. 1
      README.md
  2. 61
      apidocs/openapi.yaml
  3. 55
      internal/core/api.go
  4. 28
      internal/core/api_test.go
  5. 2
      internal/core/core_test.go
  6. 7
      internal/core/hls_server.go
  7. 121
      internal/core/metrics.go
  8. 2
      internal/core/metrics_test.go
  9. 7
      internal/core/path_manager.go
  10. 15
      internal/core/rtmp_server.go
  11. 9
      internal/core/rtsp_conn.go
  12. 84
      internal/core/rtsp_server.go

1
README.md

@ -443,6 +443,7 @@ Obtaining:
``` ```
paths{name="<path_name>",state="ready"} 1 paths{name="<path_name>",state="ready"} 1
rtsp_conns 1
rtsp_sessions{state="idle"} 0 rtsp_sessions{state="idle"} 0
rtsp_sessions{state="read"} 0 rtsp_sessions{state="read"} 0
rtsp_sessions{state="publish"} 1 rtsp_sessions{state="publish"} 1

61
apidocs/openapi.yaml

@ -375,20 +375,19 @@ components:
type: string type: string
enum: [hlsMuxer] enum: [hlsMuxer]
RTSPSession: RTSPConn:
type: object type: object
properties: properties:
created: created:
type: string type: string
remoteAddr: remoteAddr:
type: string type: string
state:
type: string
enum: [idle, read, publish]
RTSPSSession: RTSPSession:
type: object type: object
properties: properties:
created:
type: string
remoteAddr: remoteAddr:
type: string type: string
state: state:
@ -433,21 +432,29 @@ components:
additionalProperties: additionalProperties:
$ref: '#/components/schemas/Path' $ref: '#/components/schemas/Path'
RTSPSessionsList: RTSPConnsList:
type: object type: object
properties: properties:
items: items:
type: object type: object
additionalProperties: additionalProperties:
$ref: '#/components/schemas/RTSPSession' $ref: '#/components/schemas/RTSPConn'
RTSPSConnsList:
type: object
properties:
items:
type: object
additionalProperties:
$ref: '#/components/schemas/RTSPConn'
RTSPSSessionsList: RTSPSessionsList:
type: object type: object
properties: properties:
items: items:
type: object type: object
additionalProperties: additionalProperties:
$ref: '#/components/schemas/RTSPSSession' $ref: '#/components/schemas/RTSPSession'
RTMPConnsList: RTMPConnsList:
type: object type: object
@ -599,6 +606,23 @@ paths:
'500': '500':
description: internal server error. description: internal server error.
/v1/rtspconns/list:
get:
operationId: rtspConnsList
summary: returns all active RTSP connections.
description: ''
responses:
'200':
description: the request was successful.
content:
application/json:
schema:
$ref: '#/components/schemas/RTSPConnsList'
'400':
description: invalid request.
'500':
description: internal server error.
/v1/rtspsessions/list: /v1/rtspsessions/list:
get: get:
operationId: rtspSessionsList operationId: rtspSessionsList
@ -616,6 +640,23 @@ paths:
'500': '500':
description: internal server error. description: internal server error.
/v1/rtspsconns/list:
get:
operationId: rtspsConnsList
summary: returns all active RTSPS connections.
description: ''
responses:
'200':
description: the request was successful.
content:
application/json:
schema:
$ref: '#/components/schemas/RTSPConnsList'
'400':
description: invalid request.
'500':
description: internal server error.
/v1/rtspsessions/kick/{id}: /v1/rtspsessions/kick/{id}:
post: post:
operationId: rtspSessionsKick operationId: rtspSessionsKick
@ -647,7 +688,7 @@ paths:
content: content:
application/json: application/json:
schema: schema:
$ref: '#/components/schemas/RTSPSSessionsList' $ref: '#/components/schemas/RTSPSessionsList'
'400': '400':
description: invalid request. description: invalid request.
'500': '500':

55
internal/core/api.go

@ -83,21 +83,22 @@ func loadConfPathData(ctx *gin.Context) (interface{}, error) {
} }
type apiPathManager interface { type apiPathManager interface {
apiPathsList(req pathAPIPathsListReq) pathAPIPathsListRes apiPathsList() pathAPIPathsListRes
} }
type apiRTSPServer interface { type apiRTSPServer interface {
apiSessionsList(req rtspServerAPISessionsListReq) rtspServerAPISessionsListRes apiConnsList() rtspServerAPIConnsListRes
apiSessionsKick(req rtspServerAPISessionsKickReq) rtspServerAPISessionsKickRes apiSessionsList() rtspServerAPISessionsListRes
apiSessionsKick(string) rtspServerAPISessionsKickRes
} }
type apiRTMPServer interface { type apiRTMPServer interface {
apiConnsList(req rtmpServerAPIConnsListReq) rtmpServerAPIConnsListRes apiConnsList() rtmpServerAPIConnsListRes
apiConnsKick(req rtmpServerAPIConnsKickReq) rtmpServerAPIConnsKickRes apiConnsKick(id string) rtmpServerAPIConnsKickRes
} }
type apiHLSServer interface { type apiHLSServer interface {
apiHLSMuxersList(req hlsServerAPIMuxersListReq) hlsServerAPIMuxersListRes apiHLSMuxersList() hlsServerAPIMuxersListRes
} }
type apiParent interface { type apiParent interface {
@ -162,11 +163,13 @@ func newAPI(
group.GET("/v1/paths/list", a.onPathsList) group.GET("/v1/paths/list", a.onPathsList)
if !interfaceIsEmpty(a.rtspServer) { if !interfaceIsEmpty(a.rtspServer) {
group.GET("/v1/rtspconns/list", a.onRTSPConnsList)
group.GET("/v1/rtspsessions/list", a.onRTSPSessionsList) group.GET("/v1/rtspsessions/list", a.onRTSPSessionsList)
group.POST("/v1/rtspsessions/kick/:id", a.onRTSPSessionsKick) group.POST("/v1/rtspsessions/kick/:id", a.onRTSPSessionsKick)
} }
if !interfaceIsEmpty(a.rtspsServer) { if !interfaceIsEmpty(a.rtspsServer) {
group.GET("/v1/rtspsconns/list", a.onRTSPSConnsList)
group.GET("/v1/rtspssessions/list", a.onRTSPSSessionsList) group.GET("/v1/rtspssessions/list", a.onRTSPSSessionsList)
group.POST("/v1/rtspssessions/kick/:id", a.onRTSPSSessionsKick) group.POST("/v1/rtspssessions/kick/:id", a.onRTSPSSessionsKick)
} }
@ -382,7 +385,17 @@ func (a *api) onConfigPathsDelete(ctx *gin.Context) {
} }
func (a *api) onPathsList(ctx *gin.Context) { func (a *api) onPathsList(ctx *gin.Context) {
res := a.pathManager.apiPathsList(pathAPIPathsListReq{}) res := a.pathManager.apiPathsList()
if res.err != nil {
ctx.AbortWithStatus(http.StatusInternalServerError)
return
}
ctx.JSON(http.StatusOK, res.data)
}
func (a *api) onRTSPConnsList(ctx *gin.Context) {
res := a.rtspServer.apiConnsList()
if res.err != nil { if res.err != nil {
ctx.AbortWithStatus(http.StatusInternalServerError) ctx.AbortWithStatus(http.StatusInternalServerError)
return return
@ -392,7 +405,7 @@ func (a *api) onPathsList(ctx *gin.Context) {
} }
func (a *api) onRTSPSessionsList(ctx *gin.Context) { func (a *api) onRTSPSessionsList(ctx *gin.Context) {
res := a.rtspServer.apiSessionsList(rtspServerAPISessionsListReq{}) res := a.rtspServer.apiSessionsList()
if res.err != nil { if res.err != nil {
ctx.AbortWithStatus(http.StatusInternalServerError) ctx.AbortWithStatus(http.StatusInternalServerError)
return return
@ -404,7 +417,7 @@ func (a *api) onRTSPSessionsList(ctx *gin.Context) {
func (a *api) onRTSPSessionsKick(ctx *gin.Context) { func (a *api) onRTSPSessionsKick(ctx *gin.Context) {
id := ctx.Param("id") id := ctx.Param("id")
res := a.rtspServer.apiSessionsKick(rtspServerAPISessionsKickReq{id: id}) res := a.rtspServer.apiSessionsKick(id)
if res.err != nil { if res.err != nil {
ctx.AbortWithStatus(http.StatusNotFound) ctx.AbortWithStatus(http.StatusNotFound)
return return
@ -413,8 +426,18 @@ func (a *api) onRTSPSessionsKick(ctx *gin.Context) {
ctx.Status(http.StatusOK) ctx.Status(http.StatusOK)
} }
func (a *api) onRTSPSConnsList(ctx *gin.Context) {
res := a.rtspsServer.apiConnsList()
if res.err != nil {
ctx.AbortWithStatus(http.StatusInternalServerError)
return
}
ctx.JSON(http.StatusOK, res.data)
}
func (a *api) onRTSPSSessionsList(ctx *gin.Context) { func (a *api) onRTSPSSessionsList(ctx *gin.Context) {
res := a.rtspsServer.apiSessionsList(rtspServerAPISessionsListReq{}) res := a.rtspsServer.apiSessionsList()
if res.err != nil { if res.err != nil {
ctx.AbortWithStatus(http.StatusInternalServerError) ctx.AbortWithStatus(http.StatusInternalServerError)
return return
@ -426,7 +449,7 @@ func (a *api) onRTSPSSessionsList(ctx *gin.Context) {
func (a *api) onRTSPSSessionsKick(ctx *gin.Context) { func (a *api) onRTSPSSessionsKick(ctx *gin.Context) {
id := ctx.Param("id") id := ctx.Param("id")
res := a.rtspsServer.apiSessionsKick(rtspServerAPISessionsKickReq{id: id}) res := a.rtspsServer.apiSessionsKick(id)
if res.err != nil { if res.err != nil {
ctx.AbortWithStatus(http.StatusNotFound) ctx.AbortWithStatus(http.StatusNotFound)
return return
@ -436,7 +459,7 @@ func (a *api) onRTSPSSessionsKick(ctx *gin.Context) {
} }
func (a *api) onRTMPConnsList(ctx *gin.Context) { func (a *api) onRTMPConnsList(ctx *gin.Context) {
res := a.rtmpServer.apiConnsList(rtmpServerAPIConnsListReq{}) res := a.rtmpServer.apiConnsList()
if res.err != nil { if res.err != nil {
ctx.AbortWithStatus(http.StatusInternalServerError) ctx.AbortWithStatus(http.StatusInternalServerError)
return return
@ -448,7 +471,7 @@ func (a *api) onRTMPConnsList(ctx *gin.Context) {
func (a *api) onRTMPConnsKick(ctx *gin.Context) { func (a *api) onRTMPConnsKick(ctx *gin.Context) {
id := ctx.Param("id") id := ctx.Param("id")
res := a.rtmpServer.apiConnsKick(rtmpServerAPIConnsKickReq{id: id}) res := a.rtmpServer.apiConnsKick(id)
if res.err != nil { if res.err != nil {
ctx.AbortWithStatus(http.StatusNotFound) ctx.AbortWithStatus(http.StatusNotFound)
return return
@ -458,7 +481,7 @@ func (a *api) onRTMPConnsKick(ctx *gin.Context) {
} }
func (a *api) onRTMPSConnsList(ctx *gin.Context) { func (a *api) onRTMPSConnsList(ctx *gin.Context) {
res := a.rtmpsServer.apiConnsList(rtmpServerAPIConnsListReq{}) res := a.rtmpsServer.apiConnsList()
if res.err != nil { if res.err != nil {
ctx.AbortWithStatus(http.StatusInternalServerError) ctx.AbortWithStatus(http.StatusInternalServerError)
return return
@ -470,7 +493,7 @@ func (a *api) onRTMPSConnsList(ctx *gin.Context) {
func (a *api) onRTMPSConnsKick(ctx *gin.Context) { func (a *api) onRTMPSConnsKick(ctx *gin.Context) {
id := ctx.Param("id") id := ctx.Param("id")
res := a.rtmpsServer.apiConnsKick(rtmpServerAPIConnsKickReq{id: id}) res := a.rtmpsServer.apiConnsKick(id)
if res.err != nil { if res.err != nil {
ctx.AbortWithStatus(http.StatusNotFound) ctx.AbortWithStatus(http.StatusNotFound)
return return
@ -480,7 +503,7 @@ func (a *api) onRTMPSConnsKick(ctx *gin.Context) {
} }
func (a *api) onHLSMuxersList(ctx *gin.Context) { func (a *api) onHLSMuxersList(ctx *gin.Context) {
res := a.hlsServer.apiHLSMuxersList(hlsServerAPIMuxersListReq{}) res := a.hlsServer.apiHLSMuxersList()
if res.err != nil { if res.err != nil {
ctx.AbortWithStatus(http.StatusInternalServerError) ctx.AbortWithStatus(http.StatusInternalServerError)
return return

28
internal/core/api_test.go

@ -370,8 +370,10 @@ func TestAPIProtocolSpecificList(t *testing.T) {
defer os.Remove(serverKeyFpath) defer os.Remove(serverKeyFpath)
for _, ca := range []string{ for _, ca := range []string{
"rtsp", "rtsp conns",
"rtsps", "rtsp sessions",
"rtsps conns",
"rtsps sessions",
"rtmp", "rtmp",
"rtmps", "rtmps",
"hls", "hls",
@ -380,7 +382,7 @@ func TestAPIProtocolSpecificList(t *testing.T) {
conf := "api: yes\n" conf := "api: yes\n"
switch ca { switch ca {
case "rtsps": case "rtsps conns", "rtsps sessions":
conf += "protocols: [tcp]\n" + conf += "protocols: [tcp]\n" +
"encryption: strict\n" + "encryption: strict\n" +
"serverCert: " + serverCertFpath + "\n" + "serverCert: " + serverCertFpath + "\n" +
@ -406,7 +408,7 @@ func TestAPIProtocolSpecificList(t *testing.T) {
} }
switch ca { switch ca {
case "rtsp": case "rtsp conns", "rtsp sessions":
source := gortsplib.Client{} source := gortsplib.Client{}
err := source.StartPublishing("rtsp://localhost:8554/mypath", err := source.StartPublishing("rtsp://localhost:8554/mypath",
@ -414,7 +416,7 @@ func TestAPIProtocolSpecificList(t *testing.T) {
require.NoError(t, err) require.NoError(t, err)
defer source.Close() defer source.Close()
case "rtsps": case "rtsps conns", "rtsps sessions":
source := gortsplib.Client{ source := gortsplib.Client{
TLSConfig: &tls.Config{InsecureSkipVerify: true}, TLSConfig: &tls.Config{InsecureSkipVerify: true},
} }
@ -478,13 +480,19 @@ func TestAPIProtocolSpecificList(t *testing.T) {
} }
switch ca { switch ca {
case "rtsp", "rtsps", "rtmp", "rtmps": case "rtsp conns", "rtsp sessions", "rtsps conns", "rtsps sessions", "rtmp", "rtmps":
var pa string var pa string
switch ca { switch ca {
case "rtsp": case "rtsp conns":
pa = "rtspconns"
case "rtsp sessions":
pa = "rtspsessions" pa = "rtspsessions"
case "rtsps": case "rtsps conns":
pa = "rtspsconns"
case "rtsps sessions":
pa = "rtspssessions" pa = "rtspssessions"
case "rtmp": case "rtmp":
@ -507,7 +515,9 @@ func TestAPIProtocolSpecificList(t *testing.T) {
firstID = k firstID = k
} }
require.Equal(t, "publish", out.Items[firstID].State) if ca != "rtsp conns" && ca != "rtsps conns" {
require.Equal(t, "publish", out.Items[firstID].State)
}
case "hls": case "hls":
var out struct { var out struct {

2
internal/core/core_test.go

@ -163,7 +163,7 @@ func TestCorePathAutoDeletion(t *testing.T) {
} }
}() }()
res := p.pathManager.apiPathsList(pathAPIPathsListReq{}) res := p.pathManager.apiPathsList()
require.NoError(t, res.err) require.NoError(t, res.err)
require.Equal(t, 0, len(res.data.Items)) require.Equal(t, 0, len(res.data.Items))

7
internal/core/hls_server.go

@ -392,8 +392,11 @@ func (s *hlsServer) pathSourceNotReady(pa *path) {
} }
// apiHLSMuxersList is called by api. // apiHLSMuxersList is called by api.
func (s *hlsServer) apiHLSMuxersList(req hlsServerAPIMuxersListReq) hlsServerAPIMuxersListRes { func (s *hlsServer) apiHLSMuxersList() hlsServerAPIMuxersListRes {
req.res = make(chan hlsServerAPIMuxersListRes) req := hlsServerAPIMuxersListReq{
res: make(chan hlsServerAPIMuxersListRes),
}
select { select {
case s.chAPIMuxerList <- req: case s.chAPIMuxerList <- req:
res := <-req.res res := <-req.res

121
internal/core/metrics.go

@ -18,19 +18,20 @@ func metric(key string, value int64) string {
} }
type metricsPathManager interface { type metricsPathManager interface {
apiPathsList(req pathAPIPathsListReq) pathAPIPathsListRes apiPathsList() pathAPIPathsListRes
} }
type metricsRTSPServer interface { type metricsRTSPServer interface {
apiSessionsList(req rtspServerAPISessionsListReq) rtspServerAPISessionsListRes apiConnsList() rtspServerAPIConnsListRes
apiSessionsList() rtspServerAPISessionsListRes
} }
type metricsRTMPServer interface { type metricsRTMPServer interface {
apiConnsList(req rtmpServerAPIConnsListReq) rtmpServerAPIConnsListRes apiConnsList() rtmpServerAPIConnsListRes
} }
type metricsHLSServer interface { type metricsHLSServer interface {
apiHLSMuxersList(req hlsServerAPIMuxersListReq) hlsServerAPIMuxersListRes apiHLSMuxersList() hlsServerAPIMuxersListRes
} }
type metricsParent interface { type metricsParent interface {
@ -90,7 +91,7 @@ func (m *metrics) log(level logger.Level, format string, args ...interface{}) {
func (m *metrics) onMetrics(ctx *gin.Context) { func (m *metrics) onMetrics(ctx *gin.Context) {
out := "" out := ""
res := m.pathManager.apiPathsList(pathAPIPathsListReq{}) res := m.pathManager.apiPathsList()
if res.err == nil { if res.err == nil {
for name, p := range res.data.Items { for name, p := range res.data.Items {
if p.SourceReady { if p.SourceReady {
@ -102,61 +103,79 @@ func (m *metrics) onMetrics(ctx *gin.Context) {
} }
if !interfaceIsEmpty(m.rtspServer) { if !interfaceIsEmpty(m.rtspServer) {
res := m.rtspServer.apiSessionsList(rtspServerAPISessionsListReq{}) func() {
if res.err == nil { res := m.rtspServer.apiConnsList()
idleCount := int64(0) if res.err == nil {
readCount := int64(0) out += metric("rtsp_conns", int64(len(res.data.Items)))
publishCount := int64(0)
for _, i := range res.data.Items {
switch i.State {
case "idle":
idleCount++
case "read":
readCount++
case "publish":
publishCount++
}
} }
}()
func() {
res := m.rtspServer.apiSessionsList()
if res.err == nil {
idleCount := int64(0)
readCount := int64(0)
publishCount := int64(0)
for _, i := range res.data.Items {
switch i.State {
case "idle":
idleCount++
case "read":
readCount++
case "publish":
publishCount++
}
}
out += metric("rtsp_sessions{state=\"idle\"}", out += metric("rtsp_sessions{state=\"idle\"}",
idleCount) idleCount)
out += metric("rtsp_sessions{state=\"read\"}", out += metric("rtsp_sessions{state=\"read\"}",
readCount) readCount)
out += metric("rtsp_sessions{state=\"publish\"}", out += metric("rtsp_sessions{state=\"publish\"}",
publishCount) publishCount)
} }
}()
} }
if !interfaceIsEmpty(m.rtspsServer) { if !interfaceIsEmpty(m.rtspsServer) {
res := m.rtspsServer.apiSessionsList(rtspServerAPISessionsListReq{}) func() {
if res.err == nil { res := m.rtspsServer.apiConnsList()
idleCount := int64(0) if res.err == nil {
readCount := int64(0) out += metric("rtsps_conns", int64(len(res.data.Items)))
publishCount := int64(0)
for _, i := range res.data.Items {
switch i.State {
case "idle":
idleCount++
case "read":
readCount++
case "publish":
publishCount++
}
} }
}()
func() {
res := m.rtspsServer.apiSessionsList()
if res.err == nil {
idleCount := int64(0)
readCount := int64(0)
publishCount := int64(0)
for _, i := range res.data.Items {
switch i.State {
case "idle":
idleCount++
case "read":
readCount++
case "publish":
publishCount++
}
}
out += metric("rtsps_sessions{state=\"idle\"}", out += metric("rtsps_sessions{state=\"idle\"}",
idleCount) idleCount)
out += metric("rtsps_sessions{state=\"read\"}", out += metric("rtsps_sessions{state=\"read\"}",
readCount) readCount)
out += metric("rtsps_sessions{state=\"publish\"}", out += metric("rtsps_sessions{state=\"publish\"}",
publishCount) publishCount)
} }
}()
} }
if !interfaceIsEmpty(m.rtmpServer) { if !interfaceIsEmpty(m.rtmpServer) {
res := m.rtmpServer.apiConnsList(rtmpServerAPIConnsListReq{}) res := m.rtmpServer.apiConnsList()
if res.err == nil { if res.err == nil {
idleCount := int64(0) idleCount := int64(0)
readCount := int64(0) readCount := int64(0)
@ -183,7 +202,7 @@ func (m *metrics) onMetrics(ctx *gin.Context) {
} }
if !interfaceIsEmpty(m.hlsServer) { if !interfaceIsEmpty(m.hlsServer) {
res := m.hlsServer.apiHLSMuxersList(hlsServerAPIMuxersListReq{}) res := m.hlsServer.apiHLSMuxersList()
if res.err == nil { if res.err == nil {
for name := range res.data.Items { for name := range res.data.Items {
out += metric("hls_muxers{name=\""+name+"\"}", 1) out += metric("hls_muxers{name=\""+name+"\"}", 1)

2
internal/core/metrics_test.go

@ -102,9 +102,11 @@ func TestMetrics(t *testing.T) {
"rtmp_conns{state=\"idle\"}": "0", "rtmp_conns{state=\"idle\"}": "0",
"rtmp_conns{state=\"publish\"}": "1", "rtmp_conns{state=\"publish\"}": "1",
"rtmp_conns{state=\"read\"}": "0", "rtmp_conns{state=\"read\"}": "0",
"rtsp_conns": "1",
"rtsp_sessions{state=\"idle\"}": "0", "rtsp_sessions{state=\"idle\"}": "0",
"rtsp_sessions{state=\"publish\"}": "1", "rtsp_sessions{state=\"publish\"}": "1",
"rtsp_sessions{state=\"read\"}": "0", "rtsp_sessions{state=\"read\"}": "0",
"rtsps_conns": "0",
"rtsps_sessions{state=\"idle\"}": "0", "rtsps_sessions{state=\"idle\"}": "0",
"rtsps_sessions{state=\"publish\"}": "0", "rtsps_sessions{state=\"publish\"}": "0",
"rtsps_sessions{state=\"read\"}": "0", "rtsps_sessions{state=\"read\"}": "0",

7
internal/core/path_manager.go

@ -407,8 +407,11 @@ func (pm *pathManager) hlsServerSet(s pathManagerHLSServer) {
} }
// apiPathsList is called by api. // apiPathsList is called by api.
func (pm *pathManager) apiPathsList(req pathAPIPathsListReq) pathAPIPathsListRes { func (pm *pathManager) apiPathsList() pathAPIPathsListRes {
req.res = make(chan pathAPIPathsListRes) req := pathAPIPathsListReq{
res: make(chan pathAPIPathsListRes),
}
select { select {
case pm.chAPIPathsList <- req: case pm.chAPIPathsList <- req:
res := <-req.res res := <-req.res

15
internal/core/rtmp_server.go

@ -314,8 +314,11 @@ func (s *rtmpServer) connClose(c *rtmpConn) {
} }
// apiConnsList is called by api. // apiConnsList is called by api.
func (s *rtmpServer) apiConnsList(req rtmpServerAPIConnsListReq) rtmpServerAPIConnsListRes { func (s *rtmpServer) apiConnsList() rtmpServerAPIConnsListRes {
req.res = make(chan rtmpServerAPIConnsListRes) req := rtmpServerAPIConnsListReq{
res: make(chan rtmpServerAPIConnsListRes),
}
select { select {
case s.chAPIConnsList <- req: case s.chAPIConnsList <- req:
return <-req.res return <-req.res
@ -326,8 +329,12 @@ func (s *rtmpServer) apiConnsList(req rtmpServerAPIConnsListReq) rtmpServerAPICo
} }
// apiConnsKick is called by api. // apiConnsKick is called by api.
func (s *rtmpServer) apiConnsKick(req rtmpServerAPIConnsKickReq) rtmpServerAPIConnsKickRes { func (s *rtmpServer) apiConnsKick(id string) rtmpServerAPIConnsKickRes {
req.res = make(chan rtmpServerAPIConnsKickRes) req := rtmpServerAPIConnsKickReq{
id: id,
res: make(chan rtmpServerAPIConnsKickRes),
}
select { select {
case s.chAPIConnsKick <- req: case s.chAPIConnsKick <- req:
return <-req.res return <-req.res

9
internal/core/rtsp_conn.go

@ -25,6 +25,7 @@ type rtspConnParent interface {
} }
type rtspConn struct { type rtspConn struct {
id string
externalAuthenticationURL string externalAuthenticationURL string
rtspAddress string rtspAddress string
authMethods []headers.AuthMethod authMethods []headers.AuthMethod
@ -36,6 +37,7 @@ type rtspConn struct {
conn *gortsplib.ServerConn conn *gortsplib.ServerConn
parent rtspConnParent parent rtspConnParent
created time.Time
onConnectCmd *externalcmd.Cmd onConnectCmd *externalcmd.Cmd
authUser string authUser string
authPass string authPass string
@ -44,6 +46,7 @@ type rtspConn struct {
} }
func newRTSPConn( func newRTSPConn(
id string,
externalAuthenticationURL string, externalAuthenticationURL string,
rtspAddress string, rtspAddress string,
authMethods []headers.AuthMethod, authMethods []headers.AuthMethod,
@ -56,6 +59,7 @@ func newRTSPConn(
parent rtspConnParent, parent rtspConnParent,
) *rtspConn { ) *rtspConn {
c := &rtspConn{ c := &rtspConn{
id: id,
externalAuthenticationURL: externalAuthenticationURL, externalAuthenticationURL: externalAuthenticationURL,
rtspAddress: rtspAddress, rtspAddress: rtspAddress,
authMethods: authMethods, authMethods: authMethods,
@ -66,6 +70,7 @@ func newRTSPConn(
pathManager: pathManager, pathManager: pathManager,
conn: conn, conn: conn,
parent: parent, parent: parent,
created: time.Now(),
} }
c.log(logger.Info, "opened") c.log(logger.Info, "opened")
@ -98,6 +103,10 @@ func (c *rtspConn) Conn() *gortsplib.ServerConn {
return c.conn return c.conn
} }
func (c *rtspConn) remoteAddr() net.Addr {
return c.conn.NetConn().RemoteAddr()
}
func (c *rtspConn) ip() net.IP { func (c *rtspConn) ip() net.IP {
return c.conn.NetConn().RemoteAddr().(*net.TCPAddr).IP return c.conn.NetConn().RemoteAddr().(*net.TCPAddr).IP
} }

84
internal/core/rtsp_server.go

@ -20,6 +20,20 @@ import (
"github.com/aler9/rtsp-simple-server/internal/logger" "github.com/aler9/rtsp-simple-server/internal/logger"
) )
type rtspServerAPIConnsListItem struct {
Created time.Time `json:"created"`
RemoteAddr string `json:"remoteAddr"`
}
type rtspServerAPIConnsListData struct {
Items map[string]rtspServerAPIConnsListItem `json:"items"`
}
type rtspServerAPIConnsListRes struct {
data *rtspServerAPIConnsListData
err error
}
type rtspServerAPISessionsListItem struct { type rtspServerAPISessionsListItem struct {
Created time.Time `json:"created"` Created time.Time `json:"created"`
RemoteAddr string `json:"remoteAddr"` RemoteAddr string `json:"remoteAddr"`
@ -35,16 +49,10 @@ type rtspServerAPISessionsListRes struct {
err error err error
} }
type rtspServerAPISessionsListReq struct{}
type rtspServerAPISessionsKickRes struct { type rtspServerAPISessionsKickRes struct {
err error err error
} }
type rtspServerAPISessionsKickReq struct {
id string
}
type rtspServerParent interface { type rtspServerParent interface {
Log(logger.Level, string, ...interface{}) Log(logger.Level, string, ...interface{})
} }
@ -259,9 +267,40 @@ func (s *rtspServer) newSessionID() (string, error) {
} }
} }
func (s *rtspServer) newConnID() (string, error) {
for {
b := make([]byte, 4)
_, err := rand.Read(b)
if err != nil {
return "", err
}
u := uint32(b[3])<<24 | uint32(b[2])<<16 | uint32(b[1])<<8 | uint32(b[0])
u %= 899999999
u += 100000000
id := strconv.FormatUint(uint64(u), 10)
alreadyPresent := func() bool {
for _, c := range s.conns {
if c.id == id {
return true
}
}
return false
}()
if !alreadyPresent {
return id, nil
}
}
}
// OnConnOpen implements gortsplib.ServerHandlerOnConnOpen. // OnConnOpen implements gortsplib.ServerHandlerOnConnOpen.
func (s *rtspServer) OnConnOpen(ctx *gortsplib.ServerHandlerOnConnOpenCtx) { func (s *rtspServer) OnConnOpen(ctx *gortsplib.ServerHandlerOnConnOpenCtx) {
s.mutex.Lock()
id, _ := s.newConnID()
c := newRTSPConn( c := newRTSPConn(
id,
s.externalAuthenticationURL, s.externalAuthenticationURL,
s.rtspAddress, s.rtspAddress,
s.authMethods, s.authMethods,
@ -272,9 +311,9 @@ func (s *rtspServer) OnConnOpen(ctx *gortsplib.ServerHandlerOnConnOpenCtx) {
s.pathManager, s.pathManager,
ctx.Conn, ctx.Conn,
s) s)
s.mutex.Lock()
s.conns[ctx.Conn] = c s.conns[ctx.Conn] = c
s.mutex.Unlock() s.mutex.Unlock()
ctx.Conn.SetUserData(c) ctx.Conn.SetUserData(c)
} }
@ -380,8 +419,33 @@ func (s *rtspServer) OnDecodeError(ctx *gortsplib.ServerHandlerOnDecodeErrorCtx)
se.onDecodeError(ctx) se.onDecodeError(ctx)
} }
// apiConnsList is called by api and metrics.
func (s *rtspServer) apiConnsList() rtspServerAPIConnsListRes {
select {
case <-s.ctx.Done():
return rtspServerAPIConnsListRes{err: fmt.Errorf("terminated")}
default:
}
s.mutex.RLock()
defer s.mutex.RUnlock()
data := &rtspServerAPIConnsListData{
Items: make(map[string]rtspServerAPIConnsListItem),
}
for _, c := range s.conns {
data.Items[c.id] = rtspServerAPIConnsListItem{
Created: c.created,
RemoteAddr: c.remoteAddr().String(),
}
}
return rtspServerAPIConnsListRes{data: data}
}
// apiSessionsList is called by api and metrics. // apiSessionsList is called by api and metrics.
func (s *rtspServer) apiSessionsList(req rtspServerAPISessionsListReq) rtspServerAPISessionsListRes { func (s *rtspServer) apiSessionsList() rtspServerAPISessionsListRes {
select { select {
case <-s.ctx.Done(): case <-s.ctx.Done():
return rtspServerAPISessionsListRes{err: fmt.Errorf("terminated")} return rtspServerAPISessionsListRes{err: fmt.Errorf("terminated")}
@ -418,7 +482,7 @@ func (s *rtspServer) apiSessionsList(req rtspServerAPISessionsListReq) rtspServe
} }
// apiSessionsKick is called by api. // apiSessionsKick is called by api.
func (s *rtspServer) apiSessionsKick(req rtspServerAPISessionsKickReq) rtspServerAPISessionsKickRes { func (s *rtspServer) apiSessionsKick(id string) rtspServerAPISessionsKickRes {
select { select {
case <-s.ctx.Done(): case <-s.ctx.Done():
return rtspServerAPISessionsKickRes{err: fmt.Errorf("terminated")} return rtspServerAPISessionsKickRes{err: fmt.Errorf("terminated")}
@ -429,7 +493,7 @@ func (s *rtspServer) apiSessionsKick(req rtspServerAPISessionsKickReq) rtspServe
defer s.mutex.RUnlock() defer s.mutex.RUnlock()
for key, se := range s.sessions { for key, se := range s.sessions {
if se.id == req.id { if se.id == id {
se.close() se.close()
delete(s.sessions, key) delete(s.sessions, key)
se.onClose(liberrors.ErrServerTerminated{}) se.onClose(liberrors.ErrServerTerminated{})

Loading…
Cancel
Save