Browse Source

api: add pagination to all /list endpoints (#1808)

pull/1809/head
Alessandro Ros 3 years ago committed by GitHub
parent
commit
9c79197f36
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
  1. 10
      README.md
  2. 165
      apidocs/openapi.yaml
  3. 113
      internal/core/api.go
  4. 132
      internal/core/api_test.go
  5. 6
      internal/core/hls_manager.go
  6. 5
      internal/core/hls_muxer.go
  7. 0
      internal/core/http_server_header.go
  8. 32
      internal/core/metrics.go
  9. 9
      internal/core/path.go
  10. 2
      internal/core/path_manager.go
  11. 11
      internal/core/rtmp_server.go
  12. 22
      internal/core/rtsp_server.go
  13. 11
      internal/core/webrtc_manager.go

10
README.md

@ -10,23 +10,23 @@ Live streams can be published to the server with:
|protocol|variants|video codecs|audio codecs| |protocol|variants|video codecs|audio codecs|
|--------|--------|------------|------------| |--------|--------|------------|------------|
|RTSP clients (FFmpeg, GStreamer)|UDP, TCP, RTSPS|AV1, VP9, VP8, H265, H264, MPEG-4 Video (H263, Xvid), MPEG-2 Video, M-JPEG and any RTP-compatible codec|Opus, MPEG-4 Audio (AAC), MPEG-2 Audio (MP3), G722, G711, LPCM and any RTP-compatible codec| |WebRTC|WHIP|AV1, VP9, VP8, H264|Opus, G722, G711|
|RTSP servers and cameras|UDP, UDP-Multicast, TCP, RTSPS|AV1, VP9, VP8, H265, H264, MPEG-4 Video (H263, Xvid), MPEG-2 Video, M-JPEG and any RTP-compatible codec|Opus, MPEG-4 Audio (AAC), MPEG-2 Audio (MP3), G722, G711, LPCM and any RTP-compatible codec| |RTSP clients|UDP, TCP, RTSPS|AV1, VP9, VP8, H265, H264, MPEG-4 Video (H263, Xvid), MPEG-2 Video, M-JPEG and any RTP-compatible codec|Opus, MPEG-4 Audio (AAC), MPEG-2 Audio (MP3), G722, G711, LPCM and any RTP-compatible codec|
|RTSP servers and cameras|UDP, UDP-Multicast, TCP, RTSPS|AV1, VP9, VP8, H265, H264, MPEG-4 Video (H263, Xvid), MPEG-2 Video, M-JPEG and any RTP-compatible codec|Opus, MPEG-4 Audio (AAC), MPEG-2 Audio (MP3), G722, G711, LPCM and any RTP-compatible codec|
|RTMP clients (OBS Studio)|RTMP, RTMPS, Enhanced RTMP|AV1, H265, H264|MPEG-4 Audio (AAC), MPEG-2 Audio (MP3)| |RTMP clients (OBS Studio)|RTMP, RTMPS, Enhanced RTMP|AV1, H265, H264|MPEG-4 Audio (AAC), MPEG-2 Audio (MP3)|
|RTMP servers and cameras|RTMP, RTMPS, Enhanced RTMP|H264|MPEG-4 Audio (AAC), MPEG-2 Audio (MP3)| |RTMP servers and cameras|RTMP, RTMPS, Enhanced RTMP|H264|MPEG-4 Audio (AAC), MPEG-2 Audio (MP3)|
|HLS servers and cameras|Low-Latency HLS, MP4-based HLS, legacy HLS|H265, H264|Opus, MPEG-4 Audio (AAC)| |HLS servers and cameras|Low-Latency HLS, MP4-based HLS, legacy HLS|H265, H264|Opus, MPEG-4 Audio (AAC)|
|UDP/MPEG-TS streams|Unicast, broadcast, multicast|H265, H264|Opus, MPEG-4 Audio (AAC)| |UDP/MPEG-TS streams|Unicast, broadcast, multicast|H265, H264|Opus, MPEG-4 Audio (AAC)|
|WebRTC|WHIP|AV1, VP9, VP8, H264|Opus, G722, G711|
|Raspberry Pi Cameras||H264|| |Raspberry Pi Cameras||H264||
And can be read from the server with: And can be read from the server with:
|protocol|variants|video codecs|audio codecs| |protocol|variants|video codecs|audio codecs|
|--------|--------|------------|------------| |--------|--------|------------|------------|
|RTSP|UDP, UDP-Multicast, TCP, RTSPS|AV1, VP9, VP8, H265, H264, MPEG-4 Video (H263, Xvid), MPEG-2 Video, M-JPEG and any RTP-compatible codec|Opus, MPEG-4 Audio (AAC), MPEG-2 Audio (MP3), G722, G711, LPCM and any RTP-compatible codec| |WebRTC|WHEP|AV1, VP9, VP8, H264|Opus, G722, G711|
|RTSP|UDP, UDP-Multicast, TCP, RTSPS|AV1, VP9, VP8, H265, H264, MPEG-4 Video (H263, Xvid), MPEG-2 Video, M-JPEG and any RTP-compatible codec|Opus, MPEG-4 Audio (AAC), MPEG-2 Audio (MP3), G722, G711, LPCM and any RTP-compatible codec|
|RTMP|RTMP, RTMPS, Enhanced RTMP|H264|MPEG-4 Audio (AAC), MPEG-2 Audio (MP3)| |RTMP|RTMP, RTMPS, Enhanced RTMP|H264|MPEG-4 Audio (AAC), MPEG-2 Audio (MP3)|
|HLS|Low-Latency HLS, MP4-based HLS, legacy HLS|H265, H264|Opus, MPEG-4 Audio (AAC)| |HLS|Low-Latency HLS, MP4-based HLS, legacy HLS|H265, H264|Opus, MPEG-4 Audio (AAC)|
|WebRTC|WHEP|AV1, VP9, VP8, H264|Opus, G722, G711|
Features: Features:

165
apidocs/openapi.yaml

@ -311,6 +311,8 @@ components:
Path: Path:
type: object type: object
properties: properties:
name:
type: string
confName: confName:
type: string type: string
conf: conf:
@ -355,6 +357,8 @@ components:
RTSPConn: RTSPConn:
type: object type: object
properties: properties:
id:
type: string
created: created:
type: string type: string
remoteAddr: remoteAddr:
@ -369,6 +373,8 @@ components:
RTSPSession: RTSPSession:
type: object type: object
properties: properties:
id:
type: string
created: created:
type: string type: string
remoteAddr: remoteAddr:
@ -386,6 +392,8 @@ components:
RTMPConn: RTMPConn:
type: object type: object
properties: properties:
id:
type: string
created: created:
type: string type: string
remoteAddr: remoteAddr:
@ -403,6 +411,8 @@ components:
HLSMuxer: HLSMuxer:
type: object type: object
properties: properties:
path:
type: string
created: created:
type: string type: string
lastRequest: lastRequest:
@ -414,46 +424,58 @@ components:
HLSMuxersList: HLSMuxersList:
type: object type: object
properties: properties:
pageCount:
type: integer
items: items:
type: object type: array
additionalProperties: items:
$ref: '#/components/schemas/HLSMuxer' $ref: '#/components/schemas/HLSMuxer'
PathsList: PathsList:
type: object type: object
properties: properties:
pageCount:
type: integer
items: items:
type: object type: array
additionalProperties: items:
$ref: '#/components/schemas/Path' $ref: '#/components/schemas/Path'
RTMPConnsList: RTMPConnsList:
type: object type: object
properties: properties:
pageCount:
type: integer
items: items:
type: object type: array
additionalProperties: items:
$ref: '#/components/schemas/RTMPConn' $ref: '#/components/schemas/RTMPConn'
RTSPConnsList: RTSPConnsList:
type: object type: object
properties: properties:
pageCount:
type: integer
items: items:
type: object type: array
additionalProperties: items:
$ref: '#/components/schemas/RTSPConn' $ref: '#/components/schemas/RTSPConn'
RTSPSessionsList: RTSPSessionsList:
type: object type: object
properties: properties:
pageCount:
type: integer
items: items:
type: object type: array
additionalProperties: items:
$ref: '#/components/schemas/RTSPSession' $ref: '#/components/schemas/RTSPSession'
WebRTCConn: WebRTCConn:
type: object type: object
properties: properties:
id:
type: string
created: created:
type: string type: string
remoteAddr: remoteAddr:
@ -477,9 +499,11 @@ components:
WebRTCConnsList: WebRTCConnsList:
type: object type: object
properties: properties:
pageCount:
type: integer
items: items:
type: object type: array
additionalProperties: items:
$ref: '#/components/schemas/WebRTCConn' $ref: '#/components/schemas/WebRTCConn'
paths: paths:
@ -596,6 +620,19 @@ paths:
operationId: hlsMuxersList operationId: hlsMuxersList
summary: returns all HLS muxers. summary: returns all HLS muxers.
description: '' description: ''
parameters:
- name: page
in: query
description: page number.
schema:
type: number
default: 0
- name: itemsPerPage
in: query
description: items per page.
schema:
type: number
default: 100
responses: responses:
'200': '200':
description: the request was successful. description: the request was successful.
@ -613,6 +650,19 @@ paths:
operationId: pathsList operationId: pathsList
summary: returns all paths. summary: returns all paths.
description: '' description: ''
parameters:
- name: page
in: query
description: page number.
schema:
type: number
default: 0
- name: itemsPerPage
in: query
description: items per page.
schema:
type: number
default: 100
responses: responses:
'200': '200':
description: the request was successful. description: the request was successful.
@ -630,6 +680,19 @@ paths:
operationId: rtspConnsList operationId: rtspConnsList
summary: returns all RTSP connections. summary: returns all RTSP connections.
description: '' description: ''
parameters:
- name: page
in: query
description: page number.
schema:
type: number
default: 0
- name: itemsPerPage
in: query
description: items per page.
schema:
type: number
default: 100
responses: responses:
'200': '200':
description: the request was successful. description: the request was successful.
@ -647,6 +710,19 @@ paths:
operationId: rtspSessionsList operationId: rtspSessionsList
summary: returns all RTSP sessions. summary: returns all RTSP sessions.
description: '' description: ''
parameters:
- name: page
in: query
description: page number.
schema:
type: number
default: 0
- name: itemsPerPage
in: query
description: items per page.
schema:
type: number
default: 100
responses: responses:
'200': '200':
description: the request was successful. description: the request was successful.
@ -684,6 +760,19 @@ paths:
operationId: rtspsConnsList operationId: rtspsConnsList
summary: returns all RTSPS connections. summary: returns all RTSPS connections.
description: '' description: ''
parameters:
- name: page
in: query
description: page number.
schema:
type: number
default: 0
- name: itemsPerPage
in: query
description: items per page.
schema:
type: number
default: 100
responses: responses:
'200': '200':
description: the request was successful. description: the request was successful.
@ -701,6 +790,19 @@ paths:
operationId: rtspsSessionsList operationId: rtspsSessionsList
summary: returns all RTSPS sessions. summary: returns all RTSPS sessions.
description: '' description: ''
parameters:
- name: page
in: query
description: page number.
schema:
type: number
default: 0
- name: itemsPerPage
in: query
description: items per page.
schema:
type: number
default: 100
responses: responses:
'200': '200':
description: the request was successful. description: the request was successful.
@ -738,6 +840,19 @@ paths:
operationId: rtmpConnsList operationId: rtmpConnsList
summary: returns all RTMP connections. summary: returns all RTMP connections.
description: '' description: ''
parameters:
- name: page
in: query
description: page number.
schema:
type: number
default: 0
- name: itemsPerPage
in: query
description: items per page.
schema:
type: number
default: 100
responses: responses:
'200': '200':
description: the request was successful. description: the request was successful.
@ -775,6 +890,19 @@ paths:
operationId: rtmpsConnsList operationId: rtmpsConnsList
summary: returns all RTMPS connections. summary: returns all RTMPS connections.
description: '' description: ''
parameters:
- name: page
in: query
description: page number.
schema:
type: number
default: 0
- name: itemsPerPage
in: query
description: items per page.
schema:
type: number
default: 100
responses: responses:
'200': '200':
description: the request was successful. description: the request was successful.
@ -812,6 +940,19 @@ paths:
operationId: webrtcSessionsList operationId: webrtcSessionsList
summary: returns all WebRTC connections. summary: returns all WebRTC connections.
description: '' description: ''
parameters:
- name: page
in: query
description: page number.
schema:
type: number
default: 0
- name: itemsPerPage
in: query
description: items per page.
schema:
type: number
default: 100
responses: responses:
'200': '200':
description: the request was successful. description: the request was successful.

113
internal/core/api.go

@ -7,6 +7,7 @@ import (
"net" "net"
"net/http" "net/http"
"reflect" "reflect"
"strconv"
"sync" "sync"
"time" "time"
@ -79,6 +80,55 @@ func loadConfPathData(ctx *gin.Context) (interface{}, error) {
return in, err return in, err
} }
func paginate2(itemsPtr interface{}, itemsPerPage int, page int) int {
ritems := reflect.ValueOf(itemsPtr).Elem()
itemsLen := ritems.Len()
if itemsLen == 0 {
return 0
}
pageCount := (itemsLen / itemsPerPage) + 1
min := page * itemsPerPage
if min >= itemsLen {
min = itemsLen - 1
}
max := (page + 1) * itemsPerPage
if max >= itemsLen {
max = itemsLen
}
ritems.Set(ritems.Slice(min, max))
return pageCount
}
func paginate(itemsPtr interface{}, itemsPerPageStr string, pageStr string) (int, error) {
itemsPerPage := 100
if itemsPerPageStr != "" {
tmp, err := strconv.ParseUint(itemsPerPageStr, 10, 31)
if err != nil {
return 0, err
}
itemsPerPage = int(tmp)
}
page := 0
if pageStr != "" {
tmp, err := strconv.ParseUint(pageStr, 10, 31)
if err != nil {
return 0, err
}
page = int(tmp)
}
return paginate2(itemsPtr, itemsPerPage, page), nil
}
type apiPathManager interface { type apiPathManager interface {
apiPathsList() pathAPIPathsListRes apiPathsList() pathAPIPathsListRes
} }
@ -389,6 +439,13 @@ func (a *api) onPathsList(ctx *gin.Context) {
return return
} }
pageCount, err := paginate(&res.data.Items, ctx.Query("itemsPerPage"), ctx.Query("page"))
if err != nil {
ctx.AbortWithStatus(http.StatusBadRequest)
return
}
res.data.PageCount = pageCount
ctx.JSON(http.StatusOK, res.data) ctx.JSON(http.StatusOK, res.data)
} }
@ -399,6 +456,13 @@ func (a *api) onRTSPConnsList(ctx *gin.Context) {
return return
} }
pageCount, err := paginate(&res.data.Items, ctx.Query("itemsPerPage"), ctx.Query("page"))
if err != nil {
ctx.AbortWithStatus(http.StatusBadRequest)
return
}
res.data.PageCount = pageCount
ctx.JSON(http.StatusOK, res.data) ctx.JSON(http.StatusOK, res.data)
} }
@ -409,6 +473,13 @@ func (a *api) onRTSPSessionsList(ctx *gin.Context) {
return return
} }
pageCount, err := paginate(&res.data.Items, ctx.Query("itemsPerPage"), ctx.Query("page"))
if err != nil {
ctx.AbortWithStatus(http.StatusBadRequest)
return
}
res.data.PageCount = pageCount
ctx.JSON(http.StatusOK, res.data) ctx.JSON(http.StatusOK, res.data)
} }
@ -434,6 +505,13 @@ func (a *api) onRTSPSConnsList(ctx *gin.Context) {
return return
} }
pageCount, err := paginate(&res.data.Items, ctx.Query("itemsPerPage"), ctx.Query("page"))
if err != nil {
ctx.AbortWithStatus(http.StatusBadRequest)
return
}
res.data.PageCount = pageCount
ctx.JSON(http.StatusOK, res.data) ctx.JSON(http.StatusOK, res.data)
} }
@ -444,6 +522,13 @@ func (a *api) onRTSPSSessionsList(ctx *gin.Context) {
return return
} }
pageCount, err := paginate(&res.data.Items, ctx.Query("itemsPerPage"), ctx.Query("page"))
if err != nil {
ctx.AbortWithStatus(http.StatusBadRequest)
return
}
res.data.PageCount = pageCount
ctx.JSON(http.StatusOK, res.data) ctx.JSON(http.StatusOK, res.data)
} }
@ -469,6 +554,13 @@ func (a *api) onRTMPConnsList(ctx *gin.Context) {
return return
} }
pageCount, err := paginate(&res.data.Items, ctx.Query("itemsPerPage"), ctx.Query("page"))
if err != nil {
ctx.AbortWithStatus(http.StatusBadRequest)
return
}
res.data.PageCount = pageCount
ctx.JSON(http.StatusOK, res.data) ctx.JSON(http.StatusOK, res.data)
} }
@ -494,6 +586,13 @@ func (a *api) onRTMPSConnsList(ctx *gin.Context) {
return return
} }
pageCount, err := paginate(&res.data.Items, ctx.Query("itemsPerPage"), ctx.Query("page"))
if err != nil {
ctx.AbortWithStatus(http.StatusBadRequest)
return
}
res.data.PageCount = pageCount
ctx.JSON(http.StatusOK, res.data) ctx.JSON(http.StatusOK, res.data)
} }
@ -519,6 +618,13 @@ func (a *api) onHLSMuxersList(ctx *gin.Context) {
return return
} }
pageCount, err := paginate(&res.data.Items, ctx.Query("itemsPerPage"), ctx.Query("page"))
if err != nil {
ctx.AbortWithStatus(http.StatusBadRequest)
return
}
res.data.PageCount = pageCount
ctx.JSON(http.StatusOK, res.data) ctx.JSON(http.StatusOK, res.data)
} }
@ -529,6 +635,13 @@ func (a *api) onWebRTCSessionsList(ctx *gin.Context) {
return return
} }
pageCount, err := paginate(&res.data.Items, ctx.Query("itemsPerPage"), ctx.Query("page"))
if err != nil {
ctx.AbortWithStatus(http.StatusBadRequest)
return
}
res.data.PageCount = pageCount
ctx.JSON(http.StatusOK, res.data) ctx.JSON(http.StatusOK, res.data)
} }

132
internal/core/api_test.go

@ -189,6 +189,7 @@ func TestAPIPathsList(t *testing.T) {
} }
type path struct { type path struct {
Name string `json:"name"`
Source pathSource `json:"source"` Source pathSource `json:"source"`
SourceReady bool `json:"sourceReady"` SourceReady bool `json:"sourceReady"`
Tracks []string `json:"tracks"` Tracks []string `json:"tracks"`
@ -196,7 +197,8 @@ func TestAPIPathsList(t *testing.T) {
} }
type pathList struct { type pathList struct {
Items map[string]path `json:"items"` PageCount int `json:"pageCount"`
Items []path `json:"items"`
} }
t.Run("rtsp session", func(t *testing.T) { t.Run("rtsp session", func(t *testing.T) {
@ -243,16 +245,16 @@ func TestAPIPathsList(t *testing.T) {
err = httpRequest(http.MethodGet, "http://localhost:9997/v1/paths/list", nil, &out) err = httpRequest(http.MethodGet, "http://localhost:9997/v1/paths/list", nil, &out)
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, pathList{ require.Equal(t, pathList{
Items: map[string]path{ PageCount: 1,
"mypath": { Items: []path{{
Source: pathSource{ Name: "mypath",
Type: "rtspSession", Source: pathSource{
}, Type: "rtspSession",
SourceReady: true,
Tracks: []string{"H264", "MPEG4-audio-gen"},
BytesReceived: 16,
}, },
}, SourceReady: true,
Tracks: []string{"H264", "MPEG4-audio-gen"},
BytesReceived: 16,
}},
}, out) }, out)
}) })
@ -304,15 +306,15 @@ func TestAPIPathsList(t *testing.T) {
err = httpRequest(http.MethodGet, "http://localhost:9997/v1/paths/list", nil, &out) err = httpRequest(http.MethodGet, "http://localhost:9997/v1/paths/list", nil, &out)
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, pathList{ require.Equal(t, pathList{
Items: map[string]path{ PageCount: 1,
"mypath": { Items: []path{{
Source: pathSource{ Name: "mypath",
Type: "rtspsSession", Source: pathSource{
}, Type: "rtspsSession",
SourceReady: true,
Tracks: []string{"H264", "MPEG4-audio-gen"},
}, },
}, SourceReady: true,
Tracks: []string{"H264", "MPEG4-audio-gen"},
}},
}, out) }, out)
}) })
@ -329,15 +331,15 @@ func TestAPIPathsList(t *testing.T) {
err := httpRequest(http.MethodGet, "http://localhost:9997/v1/paths/list", nil, &out) err := httpRequest(http.MethodGet, "http://localhost:9997/v1/paths/list", nil, &out)
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, pathList{ require.Equal(t, pathList{
Items: map[string]path{ PageCount: 1,
"mypath": { Items: []path{{
Source: pathSource{ Name: "mypath",
Type: "rtspSource", Source: pathSource{
}, Type: "rtspSource",
SourceReady: false,
Tracks: []string{},
}, },
}, SourceReady: false,
Tracks: []string{},
}},
}, out) }, out)
}) })
@ -354,15 +356,15 @@ func TestAPIPathsList(t *testing.T) {
err := httpRequest(http.MethodGet, "http://localhost:9997/v1/paths/list", nil, &out) err := httpRequest(http.MethodGet, "http://localhost:9997/v1/paths/list", nil, &out)
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, pathList{ require.Equal(t, pathList{
Items: map[string]path{ PageCount: 1,
"mypath": { Items: []path{{
Source: pathSource{ Name: "mypath",
Type: "rtmpSource", Source: pathSource{
}, Type: "rtmpSource",
SourceReady: false,
Tracks: []string{},
}, },
}, SourceReady: false,
Tracks: []string{},
}},
}, out) }, out)
}) })
@ -379,15 +381,15 @@ func TestAPIPathsList(t *testing.T) {
err := httpRequest(http.MethodGet, "http://localhost:9997/v1/paths/list", nil, &out) err := httpRequest(http.MethodGet, "http://localhost:9997/v1/paths/list", nil, &out)
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, pathList{ require.Equal(t, pathList{
Items: map[string]path{ PageCount: 1,
"mypath": { Items: []path{{
Source: pathSource{ Name: "mypath",
Type: "hlsSource", Source: pathSource{
}, Type: "hlsSource",
SourceReady: false,
Tracks: []string{},
}, },
}, SourceReady: false,
Tracks: []string{},
}},
}, out) }, out)
}) })
} }
@ -589,25 +591,20 @@ func TestAPIProtocolSpecificList(t *testing.T) {
} }
var out struct { var out struct {
Items map[string]struct { Items []struct {
State string `json:"state"` State string `json:"state"`
} `json:"items"` } `json:"items"`
} }
err = httpRequest(http.MethodGet, "http://localhost:9997/v1/"+pa+"/list", nil, &out) err = httpRequest(http.MethodGet, "http://localhost:9997/v1/"+pa+"/list", nil, &out)
require.NoError(t, err) require.NoError(t, err)
var firstID string
for k := range out.Items {
firstID = k
}
if ca != "rtsp conns" && ca != "rtsps conns" { if ca != "rtsp conns" && ca != "rtsps conns" {
require.Equal(t, "publish", out.Items[firstID].State) require.Equal(t, "publish", out.Items[0].State)
} }
case "hls": case "hls":
var out struct { var out struct {
Items map[string]struct { Items []struct {
Created string `json:"created"` Created string `json:"created"`
LastRequest string `json:"lastRequest"` LastRequest string `json:"lastRequest"`
} `json:"items"` } `json:"items"`
@ -615,14 +612,9 @@ func TestAPIProtocolSpecificList(t *testing.T) {
err = httpRequest(http.MethodGet, "http://localhost:9997/v1/hlsmuxers/list", nil, &out) err = httpRequest(http.MethodGet, "http://localhost:9997/v1/hlsmuxers/list", nil, &out)
require.NoError(t, err) require.NoError(t, err)
var firstID string
for k := range out.Items {
firstID = k
}
s := fmt.Sprintf("^%d-", time.Now().Year()) s := fmt.Sprintf("^%d-", time.Now().Year())
require.Regexp(t, s, out.Items[firstID].Created) require.Regexp(t, s, out.Items[0].Created)
require.Regexp(t, s, out.Items[firstID].LastRequest) require.Regexp(t, s, out.Items[0].LastRequest)
case "webrtc": case "webrtc":
type item struct { type item struct {
@ -636,18 +628,13 @@ func TestAPIProtocolSpecificList(t *testing.T) {
} }
var out struct { var out struct {
Items map[string]item `json:"items"` PageCount int `json:"pageCount"`
Items []item `json:"items"`
} }
err = httpRequest(http.MethodGet, "http://localhost:9997/v1/webrtcsessions/list", nil, &out) err = httpRequest(http.MethodGet, "http://localhost:9997/v1/webrtcsessions/list", nil, &out)
require.NoError(t, err) require.NoError(t, err)
var firstID string require.Equal(t, true, out.Items[0].PeerConnectionEstablished)
for k := range out.Items {
firstID = k
}
itm := out.Items[firstID]
require.Equal(t, true, itm.PeerConnectionEstablished)
} }
}) })
} }
@ -742,21 +729,20 @@ func TestAPIKick(t *testing.T) {
} }
var out1 struct { var out1 struct {
Items map[string]struct{} `json:"items"` Items []struct {
ID string `json:"id"`
} `json:"items"`
} }
err = httpRequest(http.MethodGet, "http://localhost:9997/v1/"+pa+"/list", nil, &out1) err = httpRequest(http.MethodGet, "http://localhost:9997/v1/"+pa+"/list", nil, &out1)
require.NoError(t, err) require.NoError(t, err)
var firstID string err = httpRequest(http.MethodPost, "http://localhost:9997/v1/"+pa+"/kick/"+out1.Items[0].ID, nil, nil)
for k := range out1.Items {
firstID = k
}
err = httpRequest(http.MethodPost, "http://localhost:9997/v1/"+pa+"/kick/"+firstID, nil, nil)
require.NoError(t, err) require.NoError(t, err)
var out2 struct { var out2 struct {
Items map[string]struct{} `json:"items"` Items []struct {
ID string `json:"id"`
} `json:"items"`
} }
err = httpRequest(http.MethodGet, "http://localhost:9997/v1/"+pa+"/list", nil, &out2) err = httpRequest(http.MethodGet, "http://localhost:9997/v1/"+pa+"/list", nil, &out2)
require.NoError(t, err) require.NoError(t, err)

6
internal/core/hls_manager.go

@ -17,13 +17,15 @@ func (nilWriter) Write(p []byte) (int, error) {
} }
type hlsManagerAPIMuxersListItem struct { type hlsManagerAPIMuxersListItem struct {
Path string `json:"path"`
Created time.Time `json:"created"` Created time.Time `json:"created"`
LastRequest time.Time `json:"lastRequest"` LastRequest time.Time `json:"lastRequest"`
BytesSent uint64 `json:"bytesSent"` BytesSent uint64 `json:"bytesSent"`
} }
type hlsManagerAPIMuxersListData struct { type hlsManagerAPIMuxersListData struct {
Items map[string]hlsManagerAPIMuxersListItem `json:"items"` PageCount int `json:"pageCount"`
Items []hlsManagerAPIMuxersListItem `json:"items"`
} }
type hlsManagerAPIMuxersListRes struct { type hlsManagerAPIMuxersListRes struct {
@ -285,7 +287,7 @@ func (m *hlsManager) apiMuxersList() hlsManagerAPIMuxersListRes {
res := <-req.res res := <-req.res
res.data = &hlsManagerAPIMuxersListData{ res.data = &hlsManagerAPIMuxersListData{
Items: make(map[string]hlsManagerAPIMuxersListItem), Items: []hlsManagerAPIMuxersListItem{},
} }
for _, pa := range res.muxers { for _, pa := range res.muxers {

5
internal/core/hls_muxer.go

@ -202,11 +202,12 @@ func (m *hlsMuxer) run() {
} }
case req := <-m.chAPIHLSMuxersList: case req := <-m.chAPIHLSMuxersList:
req.data.Items[m.pathName] = hlsManagerAPIMuxersListItem{ req.data.Items = append(req.data.Items, hlsManagerAPIMuxersListItem{
Path: m.pathName,
Created: m.created, Created: m.created,
LastRequest: time.Unix(0, atomic.LoadInt64(m.lastRequestTime)), LastRequest: time.Unix(0, atomic.LoadInt64(m.lastRequestTime)),
BytesSent: atomic.LoadUint64(m.bytesSent), BytesSent: atomic.LoadUint64(m.bytesSent),
} })
close(req.res) close(req.res)
case <-innerReady: case <-innerReady:

0
internal/core/http_serverheader.go → internal/core/http_server_header.go

32
internal/core/metrics.go

@ -88,7 +88,7 @@ func (m *metrics) onMetrics(ctx *gin.Context) {
res := m.pathManager.apiPathsList() res := m.pathManager.apiPathsList()
if res.err == nil && len(res.data.Items) != 0 { if res.err == nil && len(res.data.Items) != 0 {
for name, i := range res.data.Items { for _, i := range res.data.Items {
var state string var state string
if i.SourceReady { if i.SourceReady {
state = "ready" state = "ready"
@ -96,7 +96,7 @@ func (m *metrics) onMetrics(ctx *gin.Context) {
state = "notReady" state = "notReady"
} }
tags := "{name=\"" + name + "\",state=\"" + state + "\"}" tags := "{name=\"" + i.Name + "\",state=\"" + state + "\"}"
out += metric("paths", tags, 1) out += metric("paths", tags, 1)
out += metric("paths_bytes_received", tags, int64(i.BytesReceived)) out += metric("paths_bytes_received", tags, int64(i.BytesReceived))
} }
@ -107,8 +107,8 @@ func (m *metrics) onMetrics(ctx *gin.Context) {
if !interfaceIsEmpty(m.hlsManager) { if !interfaceIsEmpty(m.hlsManager) {
res := m.hlsManager.apiMuxersList() res := m.hlsManager.apiMuxersList()
if res.err == nil && len(res.data.Items) != 0 { if res.err == nil && len(res.data.Items) != 0 {
for name, i := range res.data.Items { for _, i := range res.data.Items {
tags := "{name=\"" + name + "\"}" tags := "{name=\"" + i.Path + "\"}"
out += metric("hls_muxers", tags, 1) out += metric("hls_muxers", tags, 1)
out += metric("hls_muxers_bytes_sent", tags, int64(i.BytesSent)) out += metric("hls_muxers_bytes_sent", tags, int64(i.BytesSent))
} }
@ -122,8 +122,8 @@ func (m *metrics) onMetrics(ctx *gin.Context) {
func() { func() {
res := m.rtspServer.apiConnsList() res := m.rtspServer.apiConnsList()
if res.err == nil && len(res.data.Items) != 0 { if res.err == nil && len(res.data.Items) != 0 {
for id, i := range res.data.Items { for _, i := range res.data.Items {
tags := "{id=\"" + id + "\"}" tags := "{id=\"" + i.ID.String() + "\"}"
out += metric("rtsp_conns", tags, 1) out += metric("rtsp_conns", tags, 1)
out += metric("rtsp_conns_bytes_received", tags, int64(i.BytesReceived)) out += metric("rtsp_conns_bytes_received", tags, int64(i.BytesReceived))
out += metric("rtsp_conns_bytes_sent", tags, int64(i.BytesSent)) out += metric("rtsp_conns_bytes_sent", tags, int64(i.BytesSent))
@ -138,8 +138,8 @@ func (m *metrics) onMetrics(ctx *gin.Context) {
func() { func() {
res := m.rtspServer.apiSessionsList() res := m.rtspServer.apiSessionsList()
if res.err == nil && len(res.data.Items) != 0 { if res.err == nil && len(res.data.Items) != 0 {
for id, i := range res.data.Items { for _, i := range res.data.Items {
tags := "{id=\"" + id + "\",state=\"" + i.State + "\"}" tags := "{id=\"" + i.ID.String() + "\",state=\"" + i.State + "\"}"
out += metric("rtsp_sessions", tags, 1) out += metric("rtsp_sessions", tags, 1)
out += metric("rtsp_sessions_bytes_received", tags, int64(i.BytesReceived)) out += metric("rtsp_sessions_bytes_received", tags, int64(i.BytesReceived))
out += metric("rtsp_sessions_bytes_sent", tags, int64(i.BytesSent)) out += metric("rtsp_sessions_bytes_sent", tags, int64(i.BytesSent))
@ -156,8 +156,8 @@ func (m *metrics) onMetrics(ctx *gin.Context) {
func() { func() {
res := m.rtspsServer.apiConnsList() res := m.rtspsServer.apiConnsList()
if res.err == nil && len(res.data.Items) != 0 { if res.err == nil && len(res.data.Items) != 0 {
for id, i := range res.data.Items { for _, i := range res.data.Items {
tags := "{id=\"" + id + "\"}" tags := "{id=\"" + i.ID.String() + "\"}"
out += metric("rtsps_conns", tags, 1) out += metric("rtsps_conns", tags, 1)
out += metric("rtsps_conns_bytes_received", tags, int64(i.BytesReceived)) out += metric("rtsps_conns_bytes_received", tags, int64(i.BytesReceived))
out += metric("rtsps_conns_bytes_sent", tags, int64(i.BytesSent)) out += metric("rtsps_conns_bytes_sent", tags, int64(i.BytesSent))
@ -172,8 +172,8 @@ func (m *metrics) onMetrics(ctx *gin.Context) {
func() { func() {
res := m.rtspsServer.apiSessionsList() res := m.rtspsServer.apiSessionsList()
if res.err == nil && len(res.data.Items) != 0 { if res.err == nil && len(res.data.Items) != 0 {
for id, i := range res.data.Items { for _, i := range res.data.Items {
tags := "{id=\"" + id + "\",state=\"" + i.State + "\"}" tags := "{id=\"" + i.ID.String() + "\",state=\"" + i.State + "\"}"
out += metric("rtsps_sessions", tags, 1) out += metric("rtsps_sessions", tags, 1)
out += metric("rtsps_sessions_bytes_received", tags, int64(i.BytesReceived)) out += metric("rtsps_sessions_bytes_received", tags, int64(i.BytesReceived))
out += metric("rtsps_sessions_bytes_sent", tags, int64(i.BytesSent)) out += metric("rtsps_sessions_bytes_sent", tags, int64(i.BytesSent))
@ -189,8 +189,8 @@ func (m *metrics) onMetrics(ctx *gin.Context) {
if !interfaceIsEmpty(m.rtmpServer) { if !interfaceIsEmpty(m.rtmpServer) {
res := m.rtmpServer.apiConnsList() res := m.rtmpServer.apiConnsList()
if res.err == nil && len(res.data.Items) != 0 { if res.err == nil && len(res.data.Items) != 0 {
for id, i := range res.data.Items { for _, i := range res.data.Items {
tags := "{id=\"" + id + "\",state=\"" + i.State + "\"}" tags := "{id=\"" + i.ID.String() + "\",state=\"" + i.State + "\"}"
out += metric("rtmp_conns", tags, 1) out += metric("rtmp_conns", tags, 1)
out += metric("rtmp_conns_bytes_received", tags, int64(i.BytesReceived)) out += metric("rtmp_conns_bytes_received", tags, int64(i.BytesReceived))
out += metric("rtmp_conns_bytes_sent", tags, int64(i.BytesSent)) out += metric("rtmp_conns_bytes_sent", tags, int64(i.BytesSent))
@ -205,8 +205,8 @@ func (m *metrics) onMetrics(ctx *gin.Context) {
if !interfaceIsEmpty(m.webRTCManager) { if !interfaceIsEmpty(m.webRTCManager) {
res := m.webRTCManager.apiSessionsList() res := m.webRTCManager.apiSessionsList()
if res.err == nil && len(res.data.Items) != 0 { if res.err == nil && len(res.data.Items) != 0 {
for id, i := range res.data.Items { for _, i := range res.data.Items {
tags := "{id=\"" + id + "\"}" tags := "{id=\"" + i.ID.String() + "\"}"
out += metric("webrtc_sessions", tags, 1) out += metric("webrtc_sessions", tags, 1)
out += metric("webrtc_sessions_bytes_received", tags, int64(i.BytesReceived)) out += metric("webrtc_sessions_bytes_received", tags, int64(i.BytesReceived))
out += metric("webrtc_sessions_bytes_sent", tags, int64(i.BytesSent)) out += metric("webrtc_sessions_bytes_sent", tags, int64(i.BytesSent))

9
internal/core/path.go

@ -159,6 +159,7 @@ type pathAPISourceOrReader struct {
} }
type pathAPIPathsListItem struct { type pathAPIPathsListItem struct {
Name string `json:"name"`
ConfName string `json:"confName"` ConfName string `json:"confName"`
Conf *conf.PathConf `json:"conf"` Conf *conf.PathConf `json:"conf"`
Source interface{} `json:"source"` Source interface{} `json:"source"`
@ -169,7 +170,8 @@ type pathAPIPathsListItem struct {
} }
type pathAPIPathsListData struct { type pathAPIPathsListData struct {
Items map[string]pathAPIPathsListItem `json:"items"` PageCount int `json:"pageCount"`
Items []pathAPIPathsListItem `json:"items"`
} }
type pathAPIPathsListRes struct { type pathAPIPathsListRes struct {
@ -897,7 +899,8 @@ func (pa *path) handleReaderAddPost(req pathReaderAddReq) {
} }
func (pa *path) handleAPIPathsList(req pathAPIPathsListSubReq) { func (pa *path) handleAPIPathsList(req pathAPIPathsListSubReq) {
req.data.Items[pa.name] = pathAPIPathsListItem{ req.data.Items = append(req.data.Items, pathAPIPathsListItem{
Name: pa.name,
ConfName: pa.confName, ConfName: pa.confName,
Conf: pa.conf, Conf: pa.conf,
Source: func() interface{} { Source: func() interface{} {
@ -921,7 +924,7 @@ func (pa *path) handleAPIPathsList(req pathAPIPathsListSubReq) {
} }
return ret return ret
}(), }(),
} })
close(req.res) close(req.res)
} }

2
internal/core/path_manager.go

@ -492,7 +492,7 @@ func (pm *pathManager) apiPathsList() pathAPIPathsListRes {
res := <-req.res res := <-req.res
res.data = &pathAPIPathsListData{ res.data = &pathAPIPathsListData{
Items: make(map[string]pathAPIPathsListItem), Items: []pathAPIPathsListItem{},
} }
for _, pa := range res.paths { for _, pa := range res.paths {

11
internal/core/rtmp_server.go

@ -16,6 +16,7 @@ import (
) )
type rtmpServerAPIConnsListItem struct { type rtmpServerAPIConnsListItem struct {
ID uuid.UUID `json:"id"`
Created time.Time `json:"created"` Created time.Time `json:"created"`
RemoteAddr string `json:"remoteAddr"` RemoteAddr string `json:"remoteAddr"`
State string `json:"state"` State string `json:"state"`
@ -24,7 +25,8 @@ type rtmpServerAPIConnsListItem struct {
} }
type rtmpServerAPIConnsListData struct { type rtmpServerAPIConnsListData struct {
Items map[string]rtmpServerAPIConnsListItem `json:"items"` PageCount int `json:"pageCount"`
Items []rtmpServerAPIConnsListItem `json:"items"`
} }
type rtmpServerAPIConnsListRes struct { type rtmpServerAPIConnsListRes struct {
@ -217,11 +219,12 @@ outer:
case req := <-s.chAPISessionsList: case req := <-s.chAPISessionsList:
data := &rtmpServerAPIConnsListData{ data := &rtmpServerAPIConnsListData{
Items: make(map[string]rtmpServerAPIConnsListItem), Items: []rtmpServerAPIConnsListItem{},
} }
for c := range s.conns { for c := range s.conns {
data.Items[c.uuid.String()] = rtmpServerAPIConnsListItem{ data.Items = append(data.Items, rtmpServerAPIConnsListItem{
ID: c.uuid,
Created: c.created, Created: c.created,
RemoteAddr: c.remoteAddr().String(), RemoteAddr: c.remoteAddr().String(),
State: func() string { State: func() string {
@ -236,7 +239,7 @@ outer:
}(), }(),
BytesReceived: c.conn.BytesReceived(), BytesReceived: c.conn.BytesReceived(),
BytesSent: c.conn.BytesSent(), BytesSent: c.conn.BytesSent(),
} })
} }
req.res <- rtmpServerAPIConnsListRes{data: data} req.res <- rtmpServerAPIConnsListRes{data: data}

22
internal/core/rtsp_server.go

@ -20,6 +20,7 @@ import (
) )
type rtspServerAPIConnsListItem struct { type rtspServerAPIConnsListItem struct {
ID uuid.UUID `json:"id"`
Created time.Time `json:"created"` Created time.Time `json:"created"`
RemoteAddr string `json:"remoteAddr"` RemoteAddr string `json:"remoteAddr"`
BytesReceived uint64 `json:"bytesReceived"` BytesReceived uint64 `json:"bytesReceived"`
@ -27,7 +28,8 @@ type rtspServerAPIConnsListItem struct {
} }
type rtspServerAPIConnsListData struct { type rtspServerAPIConnsListData struct {
Items map[string]rtspServerAPIConnsListItem `json:"items"` PageCount int `json:"pageCount"`
Items []rtspServerAPIConnsListItem `json:"items"`
} }
type rtspServerAPIConnsListRes struct { type rtspServerAPIConnsListRes struct {
@ -36,6 +38,7 @@ type rtspServerAPIConnsListRes struct {
} }
type rtspServerAPISessionsListItem struct { type rtspServerAPISessionsListItem struct {
ID uuid.UUID `json:"id"`
Created time.Time `json:"created"` Created time.Time `json:"created"`
RemoteAddr string `json:"remoteAddr"` RemoteAddr string `json:"remoteAddr"`
State string `json:"state"` State string `json:"state"`
@ -44,7 +47,8 @@ type rtspServerAPISessionsListItem struct {
} }
type rtspServerAPISessionsListData struct { type rtspServerAPISessionsListData struct {
Items map[string]rtspServerAPISessionsListItem `json:"items"` PageCount int `json:"pageCount"`
Items []rtspServerAPISessionsListItem `json:"items"`
} }
type rtspServerAPISessionsListRes struct { type rtspServerAPISessionsListRes struct {
@ -381,16 +385,17 @@ func (s *rtspServer) apiConnsList() rtspServerAPIConnsListRes {
defer s.mutex.RUnlock() defer s.mutex.RUnlock()
data := &rtspServerAPIConnsListData{ data := &rtspServerAPIConnsListData{
Items: make(map[string]rtspServerAPIConnsListItem), Items: []rtspServerAPIConnsListItem{},
} }
for _, c := range s.conns { for _, c := range s.conns {
data.Items[c.uuid.String()] = rtspServerAPIConnsListItem{ data.Items = append(data.Items, rtspServerAPIConnsListItem{
ID: c.uuid,
Created: c.created, Created: c.created,
RemoteAddr: c.remoteAddr().String(), RemoteAddr: c.remoteAddr().String(),
BytesReceived: c.conn.BytesReceived(), BytesReceived: c.conn.BytesReceived(),
BytesSent: c.conn.BytesSent(), BytesSent: c.conn.BytesSent(),
} })
} }
return rtspServerAPIConnsListRes{data: data} return rtspServerAPIConnsListRes{data: data}
@ -408,11 +413,12 @@ func (s *rtspServer) apiSessionsList() rtspServerAPISessionsListRes {
defer s.mutex.RUnlock() defer s.mutex.RUnlock()
data := &rtspServerAPISessionsListData{ data := &rtspServerAPISessionsListData{
Items: make(map[string]rtspServerAPISessionsListItem), Items: []rtspServerAPISessionsListItem{},
} }
for _, s := range s.sessions { for _, s := range s.sessions {
data.Items[s.uuid.String()] = rtspServerAPISessionsListItem{ data.Items = append(data.Items, rtspServerAPISessionsListItem{
ID: s.uuid,
Created: s.created, Created: s.created,
RemoteAddr: s.remoteAddr().String(), RemoteAddr: s.remoteAddr().String(),
State: func() string { State: func() string {
@ -429,7 +435,7 @@ func (s *rtspServer) apiSessionsList() rtspServerAPISessionsListRes {
}(), }(),
BytesReceived: s.session.BytesReceived(), BytesReceived: s.session.BytesReceived(),
BytesSent: s.session.BytesSent(), BytesSent: s.session.BytesSent(),
} })
} }
return rtspServerAPISessionsListRes{data: data} return rtspServerAPISessionsListRes{data: data}

11
internal/core/webrtc_manager.go

@ -65,6 +65,7 @@ func linkHeaderToIceServers(link []string) []webrtc.ICEServer {
} }
type webRTCManagerAPISessionsListItem struct { type webRTCManagerAPISessionsListItem struct {
ID uuid.UUID `json:"id"`
Created time.Time `json:"created"` Created time.Time `json:"created"`
RemoteAddr string `json:"remoteAddr"` RemoteAddr string `json:"remoteAddr"`
PeerConnectionEstablished bool `json:"peerConnectionEstablished"` PeerConnectionEstablished bool `json:"peerConnectionEstablished"`
@ -76,7 +77,8 @@ type webRTCManagerAPISessionsListItem struct {
} }
type webRTCManagerAPISessionsListData struct { type webRTCManagerAPISessionsListData struct {
Items map[string]webRTCManagerAPISessionsListItem `json:"items"` PageCount int `json:"pageCount"`
Items []webRTCManagerAPISessionsListItem `json:"items"`
} }
type webRTCManagerAPISessionsListRes struct { type webRTCManagerAPISessionsListRes struct {
@ -308,7 +310,7 @@ outer:
case req := <-m.chAPISessionsList: case req := <-m.chAPISessionsList:
data := &webRTCManagerAPISessionsListData{ data := &webRTCManagerAPISessionsListData{
Items: make(map[string]webRTCManagerAPISessionsListItem), Items: []webRTCManagerAPISessionsListItem{},
} }
for sx := range m.sessions { for sx := range m.sessions {
@ -327,7 +329,8 @@ outer:
bytesSent = pc.bytesSent() bytesSent = pc.bytesSent()
} }
data.Items[sx.uuid.String()] = webRTCManagerAPISessionsListItem{ data.Items = append(data.Items, webRTCManagerAPISessionsListItem{
ID: sx.uuid,
Created: sx.created, Created: sx.created,
RemoteAddr: sx.req.remoteAddr, RemoteAddr: sx.req.remoteAddr,
PeerConnectionEstablished: peerConnectionEstablished, PeerConnectionEstablished: peerConnectionEstablished,
@ -341,7 +344,7 @@ outer:
}(), }(),
BytesReceived: bytesReceived, BytesReceived: bytesReceived,
BytesSent: bytesSent, BytesSent: bytesSent,
} })
} }
req.res <- webRTCManagerAPISessionsListRes{data: data} req.res <- webRTCManagerAPISessionsListRes{data: data}

Loading…
Cancel
Save