mediamtx/internal/core/api.go

531 lines
11 KiB
Go
Raw Normal View History

package core
import (
"context"
"encoding/json"
"log"
"net"
"net/http"
"reflect"
"sync"
"github.com/gin-gonic/gin"
"github.com/aler9/mediamtx/internal/conf"
"github.com/aler9/mediamtx/internal/logger"
)
func interfaceIsEmpty(i interface{}) bool {
return reflect.ValueOf(i).Kind() != reflect.Ptr || reflect.ValueOf(i).IsNil()
}
func fillStruct(dest interface{}, source interface{}) {
rvsource := reflect.ValueOf(source).Elem()
rvdest := reflect.ValueOf(dest)
nf := rvsource.NumField()
for i := 0; i < nf; i++ {
fnew := rvsource.Field(i)
if !fnew.IsNil() {
f := rvdest.Elem().FieldByName(rvsource.Type().Field(i).Name)
if f.Kind() == reflect.Ptr {
f.Set(fnew)
} else {
f.Set(fnew.Elem())
}
}
}
}
func generateStructWithOptionalFields(model interface{}) interface{} {
var fields []reflect.StructField
rt := reflect.TypeOf(model)
nf := rt.NumField()
for i := 0; i < nf; i++ {
f := rt.Field(i)
j := f.Tag.Get("json")
if j != "-" && j != "paths" {
fields = append(fields, reflect.StructField{
Name: f.Name,
Type: reflect.PtrTo(f.Type),
Tag: f.Tag,
})
}
}
return reflect.New(reflect.StructOf(fields)).Interface()
}
func loadConfData(ctx *gin.Context) (interface{}, error) {
in := generateStructWithOptionalFields(conf.Conf{})
err := json.NewDecoder(ctx.Request.Body).Decode(in)
if err != nil {
return nil, err
}
return in, err
}
func loadConfPathData(ctx *gin.Context) (interface{}, error) {
in := generateStructWithOptionalFields(conf.PathConf{})
err := json.NewDecoder(ctx.Request.Body).Decode(in)
if err != nil {
return nil, err
}
return in, err
}
2021-08-11 10:45:53 +00:00
type apiPathManager interface {
apiPathsList() pathAPIPathsListRes
2021-08-11 10:45:53 +00:00
}
2022-12-15 23:50:47 +00:00
type apiHLSServer interface {
apiMuxersList() hlsServerAPIMuxersListRes
}
2021-08-11 10:45:53 +00:00
type apiRTSPServer interface {
apiConnsList() rtspServerAPIConnsListRes
apiSessionsList() rtspServerAPISessionsListRes
apiSessionsKick(string) rtspServerAPISessionsKickRes
2021-08-11 10:45:53 +00:00
}
type apiRTMPServer interface {
apiConnsList() rtmpServerAPIConnsListRes
apiConnsKick(id string) rtmpServerAPIConnsKickRes
2021-08-11 10:45:53 +00:00
}
type apiParent interface {
Log(logger.Level, string, ...interface{})
apiConfigSet(conf *conf.Conf)
}
2022-12-15 23:50:47 +00:00
type apiWebRTCServer interface {
apiConnsList() webRTCServerAPIConnsListRes
apiConnsKick(id string) webRTCServerAPIConnsKickRes
}
type api struct {
2022-12-15 23:50:47 +00:00
conf *conf.Conf
pathManager apiPathManager
rtspServer apiRTSPServer
rtspsServer apiRTSPServer
rtmpServer apiRTMPServer
rtmpsServer apiRTMPServer
hlsServer apiHLSServer
webRTCServer apiWebRTCServer
parent apiParent
ln net.Listener
mutex sync.Mutex
s *http.Server
}
func newAPI(
address string,
conf *conf.Conf,
2021-08-11 10:45:53 +00:00
pathManager apiPathManager,
rtspServer apiRTSPServer,
rtspsServer apiRTSPServer,
rtmpServer apiRTMPServer,
2022-08-16 11:53:04 +00:00
rtmpsServer apiRTMPServer,
2021-11-05 16:14:31 +00:00
hlsServer apiHLSServer,
2022-12-15 23:50:47 +00:00
webRTCServer apiWebRTCServer,
parent apiParent,
) (*api, error) {
ln, err := net.Listen("tcp", address)
if err != nil {
return nil, err
}
a := &api{
2022-12-15 23:50:47 +00:00
conf: conf,
pathManager: pathManager,
rtspServer: rtspServer,
rtspsServer: rtspsServer,
rtmpServer: rtmpServer,
rtmpsServer: rtmpsServer,
hlsServer: hlsServer,
webRTCServer: webRTCServer,
parent: parent,
ln: ln,
}
router := gin.New()
router.SetTrustedProxies(nil)
2022-12-15 23:50:47 +00:00
mwLog := httpLoggerMiddleware(a)
router.NoRoute(mwLog)
group := router.Group("/", mwLog)
2021-11-05 16:18:21 +00:00
2021-08-07 16:28:27 +00:00
group.GET("/v1/config/get", a.onConfigGet)
group.POST("/v1/config/set", a.onConfigSet)
group.POST("/v1/config/paths/add/*name", a.onConfigPathsAdd)
group.POST("/v1/config/paths/edit/*name", a.onConfigPathsEdit)
group.POST("/v1/config/paths/remove/*name", a.onConfigPathsDelete)
2022-12-15 23:50:47 +00:00
if !interfaceIsEmpty(a.hlsServer) {
group.GET("/v1/hlsmuxers/list", a.onHLSMuxersList)
}
2021-08-07 16:28:27 +00:00
group.GET("/v1/paths/list", a.onPathsList)
2021-11-05 16:18:21 +00:00
if !interfaceIsEmpty(a.rtspServer) {
group.GET("/v1/rtspconns/list", a.onRTSPConnsList)
2021-11-05 16:18:21 +00:00
group.GET("/v1/rtspsessions/list", a.onRTSPSessionsList)
group.POST("/v1/rtspsessions/kick/:id", a.onRTSPSessionsKick)
}
if !interfaceIsEmpty(a.rtspsServer) {
group.GET("/v1/rtspsconns/list", a.onRTSPSConnsList)
2021-11-05 16:18:21 +00:00
group.GET("/v1/rtspssessions/list", a.onRTSPSSessionsList)
group.POST("/v1/rtspssessions/kick/:id", a.onRTSPSSessionsKick)
}
if !interfaceIsEmpty(a.rtmpServer) {
group.GET("/v1/rtmpconns/list", a.onRTMPConnsList)
group.POST("/v1/rtmpconns/kick/:id", a.onRTMPConnsKick)
}
2022-08-16 11:53:04 +00:00
if !interfaceIsEmpty(a.rtmpsServer) {
group.GET("/v1/rtmpsconns/list", a.onRTMPSConnsList)
group.POST("/v1/rtmpsconns/kick/:id", a.onRTMPSConnsKick)
}
2022-12-15 23:50:47 +00:00
if !interfaceIsEmpty(a.webRTCServer) {
group.GET("/v1/webrtcconns/list", a.onWebRTCConnsList)
group.POST("/v1/webrtcconns/kick/:id", a.onWebRTCConnsKick)
2021-11-05 16:18:21 +00:00
}
a.s = &http.Server{
Handler: router,
ErrorLog: log.New(&nilWriter{}, "", 0),
}
go a.s.Serve(ln)
a.log(logger.Info, "listener opened on "+address)
return a, nil
}
func (a *api) close() {
a.log(logger.Info, "listener is closing")
a.s.Shutdown(context.Background())
2022-08-31 06:53:05 +00:00
a.ln.Close() // in case Shutdown() is called before Serve()
}
func (a *api) log(level logger.Level, format string, args ...interface{}) {
a.parent.Log(level, "[API] "+format, args...)
}
func (a *api) onConfigGet(ctx *gin.Context) {
a.mutex.Lock()
c := a.conf
a.mutex.Unlock()
ctx.JSON(http.StatusOK, c)
}
func (a *api) onConfigSet(ctx *gin.Context) {
in, err := loadConfData(ctx)
if err != nil {
ctx.AbortWithStatus(http.StatusBadRequest)
return
}
a.mutex.Lock()
defer a.mutex.Unlock()
newConf := a.conf.Clone()
fillStruct(newConf, in)
err = newConf.CheckAndFillMissing()
if err != nil {
ctx.AbortWithStatus(http.StatusBadRequest)
return
}
a.conf = newConf
2021-08-07 14:07:08 +00:00
// since reloading the configuration can cause the shutdown of the API,
// call it in a goroutine
go a.parent.apiConfigSet(newConf)
ctx.Status(http.StatusOK)
}
func (a *api) onConfigPathsAdd(ctx *gin.Context) {
name := ctx.Param("name")
if len(name) < 2 || name[0] != '/' {
ctx.AbortWithStatus(http.StatusBadRequest)
return
}
name = name[1:]
in, err := loadConfPathData(ctx)
if err != nil {
ctx.AbortWithStatus(http.StatusBadRequest)
return
}
a.mutex.Lock()
defer a.mutex.Unlock()
newConf := a.conf.Clone()
if _, ok := newConf.Paths[name]; ok {
ctx.AbortWithStatus(http.StatusBadRequest)
return
}
newConfPath := &conf.PathConf{}
fillStruct(newConfPath, in)
newConf.Paths[name] = newConfPath
err = newConf.CheckAndFillMissing()
if err != nil {
ctx.AbortWithStatus(http.StatusBadRequest)
return
}
a.conf = newConf
2021-08-07 14:07:08 +00:00
// since reloading the configuration can cause the shutdown of the API,
// call it in a goroutine
go a.parent.apiConfigSet(newConf)
ctx.Status(http.StatusOK)
}
func (a *api) onConfigPathsEdit(ctx *gin.Context) {
name := ctx.Param("name")
if len(name) < 2 || name[0] != '/' {
ctx.AbortWithStatus(http.StatusBadRequest)
return
}
name = name[1:]
in, err := loadConfPathData(ctx)
if err != nil {
ctx.AbortWithStatus(http.StatusBadRequest)
return
}
a.mutex.Lock()
defer a.mutex.Unlock()
newConf := a.conf.Clone()
newConfPath, ok := newConf.Paths[name]
if !ok {
ctx.AbortWithStatus(http.StatusBadRequest)
return
}
fillStruct(newConfPath, in)
err = newConf.CheckAndFillMissing()
if err != nil {
ctx.AbortWithStatus(http.StatusBadRequest)
return
}
a.conf = newConf
2021-08-07 14:07:08 +00:00
// since reloading the configuration can cause the shutdown of the API,
// call it in a goroutine
go a.parent.apiConfigSet(newConf)
ctx.Status(http.StatusOK)
}
func (a *api) onConfigPathsDelete(ctx *gin.Context) {
name := ctx.Param("name")
if len(name) < 2 || name[0] != '/' {
ctx.AbortWithStatus(http.StatusBadRequest)
return
}
name = name[1:]
a.mutex.Lock()
defer a.mutex.Unlock()
newConf := a.conf.Clone()
if _, ok := newConf.Paths[name]; !ok {
ctx.AbortWithStatus(http.StatusBadRequest)
return
}
delete(newConf.Paths, name)
err := newConf.CheckAndFillMissing()
if err != nil {
ctx.AbortWithStatus(http.StatusBadRequest)
return
}
a.conf = newConf
2021-08-07 14:07:08 +00:00
// since reloading the configuration can cause the shutdown of the API,
// call it in a goroutine
go a.parent.apiConfigSet(newConf)
ctx.Status(http.StatusOK)
}
func (a *api) onPathsList(ctx *gin.Context) {
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()
2022-01-14 22:42:41 +00:00
if res.err != nil {
ctx.AbortWithStatus(http.StatusInternalServerError)
return
}
2022-01-14 22:42:41 +00:00
ctx.JSON(http.StatusOK, res.data)
}
func (a *api) onRTSPSessionsList(ctx *gin.Context) {
res := a.rtspServer.apiSessionsList()
2022-01-14 22:42:41 +00:00
if res.err != nil {
ctx.AbortWithStatus(http.StatusInternalServerError)
return
}
2022-01-14 22:42:41 +00:00
ctx.JSON(http.StatusOK, res.data)
}
func (a *api) onRTSPSessionsKick(ctx *gin.Context) {
id := ctx.Param("id")
res := a.rtspServer.apiSessionsKick(id)
2022-01-14 22:42:41 +00:00
if res.err != nil {
return
}
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) {
res := a.rtspsServer.apiSessionsList()
2022-01-14 22:42:41 +00:00
if res.err != nil {
ctx.AbortWithStatus(http.StatusInternalServerError)
return
}
2022-01-14 22:42:41 +00:00
ctx.JSON(http.StatusOK, res.data)
}
func (a *api) onRTSPSSessionsKick(ctx *gin.Context) {
id := ctx.Param("id")
res := a.rtspsServer.apiSessionsKick(id)
2022-01-14 22:42:41 +00:00
if res.err != nil {
return
}
ctx.Status(http.StatusOK)
}
func (a *api) onRTMPConnsList(ctx *gin.Context) {
res := a.rtmpServer.apiConnsList()
2022-01-14 22:42:41 +00:00
if res.err != nil {
ctx.AbortWithStatus(http.StatusInternalServerError)
return
}
2022-01-14 22:42:41 +00:00
ctx.JSON(http.StatusOK, res.data)
}
func (a *api) onRTMPConnsKick(ctx *gin.Context) {
id := ctx.Param("id")
res := a.rtmpServer.apiConnsKick(id)
2022-01-14 22:42:41 +00:00
if res.err != nil {
return
}
ctx.Status(http.StatusOK)
}
2021-11-05 16:14:31 +00:00
2022-08-16 11:53:04 +00:00
func (a *api) onRTMPSConnsList(ctx *gin.Context) {
res := a.rtmpsServer.apiConnsList()
2022-08-16 11:53:04 +00:00
if res.err != nil {
ctx.AbortWithStatus(http.StatusInternalServerError)
return
}
ctx.JSON(http.StatusOK, res.data)
}
func (a *api) onRTMPSConnsKick(ctx *gin.Context) {
id := ctx.Param("id")
res := a.rtmpsServer.apiConnsKick(id)
2022-08-16 11:53:04 +00:00
if res.err != nil {
return
}
ctx.Status(http.StatusOK)
}
2021-11-05 16:14:31 +00:00
func (a *api) onHLSMuxersList(ctx *gin.Context) {
2022-12-15 23:50:47 +00:00
res := a.hlsServer.apiMuxersList()
if res.err != nil {
ctx.AbortWithStatus(http.StatusInternalServerError)
return
}
ctx.JSON(http.StatusOK, res.data)
}
func (a *api) onWebRTCConnsList(ctx *gin.Context) {
res := a.webRTCServer.apiConnsList()
2022-01-14 22:42:41 +00:00
if res.err != nil {
2021-11-05 16:14:31 +00:00
ctx.AbortWithStatus(http.StatusInternalServerError)
return
}
2022-01-14 22:42:41 +00:00
ctx.JSON(http.StatusOK, res.data)
2021-11-05 16:14:31 +00:00
}
2022-12-15 23:50:47 +00:00
func (a *api) onWebRTCConnsKick(ctx *gin.Context) {
id := ctx.Param("id")
res := a.webRTCServer.apiConnsKick(id)
if res.err != nil {
return
}
ctx.Status(http.StatusOK)
}
// confReload is called by core.
func (a *api) confReload(conf *conf.Conf) {
2021-11-05 16:14:31 +00:00
a.mutex.Lock()
defer a.mutex.Unlock()
a.conf = conf
}