main.go

package rtsp

import (
	"net/http"
	"strconv"
	"sync"

	"github.com/bluenviron/gortsplib/v4"
	"go.uber.org/zap"
	. "m7s.live/engine/v4"
	"m7s.live/engine/v4/config"
	"m7s.live/engine/v4/util"
)

type RTSPConfig struct {
	config.Publish
	config.Subscribe
	config.Pull
	config.Push
	ListenAddr       string `default:":554" desc:"rtsp监听地址"`
	UDPAddr          string `default:":8000" desc:"udp rtp监听地址"`
	RTCPAddr         string `default:":8001" desc:"udp rtcp监听地址"`
	WriteBufferCount int    `default:"2048" desc:"rtsp写缓冲区大小"`
	SendOptions      bool   `default:"true" desc:"是否发送options请求"`
	sync.Map
	server *gortsplib.Server
}

func (conf *RTSPConfig) OnEvent(event any) {
	switch v := event.(type) {
	case FirstConfig:
		conf.server = &gortsplib.Server{
			Handler:           conf,
			RTSPAddress:       conf.ListenAddr,
			UDPRTPAddress:     conf.UDPAddr,
			UDPRTCPAddress:    conf.RTCPAddr,
			MulticastIPRange:  "224.1.0.0/16",
			MulticastRTPPort:  8002,
			MulticastRTCPPort: 8003,
			WriteQueueSize:    conf.WriteBufferCount,
		}
		if err := conf.server.Start(); err != nil {
			RTSPPlugin.Error("server start", zap.Error(err))
			RTSPPlugin.Disabled = true
		}
		for streamPath, url := range conf.PullOnStart {
			if err := RTSPPlugin.Pull(streamPath, url, new(RTSPPuller), 0); err != nil {
				RTSPPlugin.Error("pull", zap.String("streamPath", streamPath), zap.String("url", url), zap.Error(err))
			}
		}
	case SEpublish:
		if remoteURL := conf.CheckPush(v.Target.Path); remoteURL != "" {
			if err := RTSPPlugin.Push(v.Target.Path, remoteURL, new(RTSPPusher), false); err != nil {
				RTSPPlugin.Error("push", zap.String("streamPath", v.Target.Path), zap.String("url", remoteURL), zap.Error(err))
			}
		}
	case InvitePublish: //按需拉流
		if remoteURL := conf.CheckPullOnSub(v.Target); remoteURL != "" {
			if err := RTSPPlugin.Pull(v.Target, remoteURL, new(RTSPPuller), 0); err != nil {
				RTSPPlugin.Error("pull", zap.String("streamPath", v.Target), zap.String("url", remoteURL), zap.Error(err))
			}
		}
	}
}

var rtspConfig = &RTSPConfig{}
var RTSPPlugin = InstallPlugin(rtspConfig)

func filterStreams() (ss []*Stream) {
	Streams.Range(func(key string, s *Stream) {
		switch s.Publisher.(type) {
		case *RTSPPublisher, *RTSPPuller:
			ss = append(ss, s)
		}
	})
	return
}

func (*RTSPConfig) API_list(w http.ResponseWriter, r *http.Request) {
	util.ReturnFetchValue(filterStreams, w, r)
}

func (*RTSPConfig) API_Pull(rw http.ResponseWriter, r *http.Request) {
	query := r.URL.Query()
	save, _ := strconv.Atoi(query.Get("save"))
	err := RTSPPlugin.Pull(query.Get("streamPath"), query.Get("target"), new(RTSPPuller), save)
	if err != nil {
		util.ReturnError(util.APIErrorQueryParse, err.Error(), rw, r)
	} else {
		util.ReturnOK(rw, r)
	}
}

func (*RTSPConfig) API_Push(rw http.ResponseWriter, r *http.Request) {
	query := r.URL.Query()
	err := RTSPPlugin.Push(query.Get("streamPath"), query.Get("target"), new(RTSPPusher), query.Has("save"))
	if err != nil {
		util.ReturnError(util.APIErrorQueryParse, err.Error(), rw, r)
	} else {
		util.ReturnOK(rw, r)
	}
}