apps/gb-cms
1
0
Fork 0
mirror of https://github.com/lkmio/gb-cms.git synced 2025-12-24 11:51:52 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

feat: 支持目录和报警级联转发通知

Browse Source
This commit is contained in:
ydajiang
2025-10-17 10:47:25 +08:00
parent 4fad7b3d5d
commit 250544b05a
32 changed files with 2523 additions and 1862 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1766
api.go
View File

File diff suppressed because it is too large Load Diff

345
api/api.go Normal file
View File

@@ -0,0 +1,345 @@
package api
import (
"gb-cms/common"
"gb-cms/dao"
"github.com/gorilla/mux"
"github.com/gorilla/websocket"
"net/http"
"os"
"strings"
"time"
)
type ApiServer struct {
router *mux.Router
upgrader *websocket.Upgrader
}
type InviteParams struct {
DeviceID string `json:"serial"`
ChannelID string `json:"code"`
StartTime string `json:"starttime"`
EndTime string `json:"endtime"`
Setup string `json:"setup"`
Speed string `json:"speed"`
Token string `json:"token"`
Download bool `json:"download"`
streamId common.StreamID
}
type StreamParams struct {
Stream common.StreamID `json:"stream"` // Source
Protocol int `json:"protocol"` // 推拉流协议
RemoteAddr string `json:"remote_addr"` // peer地址
}
type PlayDoneParams struct {
StreamParams
Sink string `json:"sink"`
}
type QueryRecordParams struct {
DeviceID string `json:"serial"`
ChannelID string `json:"code"`
Timeout int `json:"timeout"`
StartTime string `json:"starttime"`
EndTime string `json:"endtime"`
//Type_ string `json:"type"`
Command string `json:"command"` // 云台控制命令 left/up/right/down/zoomin/zoomout
}
type DeviceChannelID struct {
DeviceID string `json:"device_id"`
ChannelID string `json:"channel_id"`
}
type SeekParams struct {
StreamId common.StreamID `json:"stream_id"`
Seconds int `json:"seconds"`
}
type BroadcastParams struct {
DeviceID string `json:"device_id"`
ChannelID string `json:"channel_id"`
StreamId common.StreamID `json:"stream_id"`
Setup *common.SetupType `json:"setup"`
}
type RecordParams struct {
StreamParams
Path string `json:"path"`
}
type StreamIDParams struct {
StreamID common.StreamID `json:"streamid"`
Command string `json:"command"`
Scale float64 `json:"scale"`
}
type PageQuery struct {
PageNumber *int `json:"page_number"` // 页数
PageSize *int `json:"page_size"` // 每页大小
TotalPages int `json:"total_pages"` // 总页数
TotalCount int `json:"total_count"` // 总记录数
Data interface{} `json:"data"`
}
type SetMediaTransportReq struct {
DeviceID string `json:"serial"`
MediaTransport string `json:"media_transport"`
MediaTransportMode string `json:"media_transport_mode"`
}
// QueryDeviceChannel 查询设备和通道的参数
type QueryDeviceChannel struct {
DeviceID string `json:"serial"`
GroupID string `json:"dir_serial"`
PCode string `json:"pcode"`
Start int `json:"start"`
Limit int `json:"limit"`
Keyword string `json:"q"`
Online string `json:"online"`
Enable string `json:"enable"`
ChannelType string `json:"channel_type"` // dir-查询子目录
Order string `json:"order"` // asc/desc
Sort string `json:"sort"` // Channel-根据数据库ID排序/iD-根据通道ID排序
SMS string `json:"sms"`
Filter string `json:"filter"`
Priority int `json:"priority"` // 报警参数
Method int `json:"method"`
StartTime string `json:"starttime"`
EndTime string `json:"endtime"`
}
type DeleteDevice struct {
DeviceID string `json:"serial"`
IP string `json:"ip"`
Forbid bool `json:"forbid"`
UA string `json:"ua"`
}
type SetEnable struct {
ID int `json:"id"`
Enable bool `json:"enable"`
ShareAllChannel bool `json:"shareallchannel"`
}
type QueryCascadeChannelList struct {
QueryDeviceChannel
ID string `json:"id"`
Related bool `json:"related"` // 只看已选择
Reverse bool `json:"reverse"`
}
type ChannelListResult struct {
ChannelCount int `json:"ChannelCount"`
ChannelList []*LiveGBSChannel `json:"ChannelList"`
}
type CascadeChannel struct {
CascadeID string
*LiveGBSChannel
}
type CustomChannel struct {
DeviceID string `json:"serial"`
ChannelID string `json:"code"`
CustomID string `json:"id"`
}
type DeviceInfo struct {
DeviceID string `json:"serial"`
CustomName string `json:"custom_name"`
MediaTransport string `json:"media_transport"`
MediaTransportMode string `json:"media_transport_mode"`
StreamMode string `json:"stream_mode"`
SMSID string `json:"sms_id"`
SMSGroupID string `json:"sms_group_id"`
RecvStreamIP string `json:"recv_stream_ip"`
ContactIP string `json:"contact_ip"`
Charset string `json:"charset"`
CatalogInterval int `json:"catalog_interval"`
SubscribeInterval int `json:"subscribe_interval"`
CatalogSubscribe bool `json:"catalog_subscribe"`
AlarmSubscribe bool `json:"alarm_subscribe"`
PositionSubscribe bool `json:"position_subscribe"`
PTZSubscribe bool `json:"ptz_subscribe"`
RecordCenter bool `json:"record_center"`
RecordIndistinct bool `json:"record_indistinct"`
CivilCodeFirst bool `json:"civil_code_first"`
KeepOriginalTree bool `json:"keep_original_tree"`
Password string `json:"password"`
DropChannelType string `json:"drop_channel_type"`
Longitude float64 `json:"longitude"`
Latitude float64 `json:"latitude"`
}
type Empty struct {
}
var apiServer *ApiServer
func init() {
apiServer = &ApiServer{
upgrader: &websocket.Upgrader{
CheckOrigin: func(r *http.Request) bool {
return true
},
},
router: mux.NewRouter(),
}
}
// 验证和刷新token
func withVerify(f func(w http.ResponseWriter, req *http.Request)) func(http.ResponseWriter, *http.Request) {
return func(w http.ResponseWriter, req *http.Request) {
cookie, err := req.Cookie("token")
if err != nil {
w.WriteHeader(http.StatusUnauthorized)
return
}
ok := TokenManager.Refresh(cookie.Value, time.Now())
if !ok {
w.WriteHeader(http.StatusUnauthorized)
return
} else if AdminMD5 == PwdMD5 && req.URL.Path != "/api/v1/modifypassword" && req.URL.Path != "/api/v1/userinfo" {
// 如果没有修改默认密码, 只允许放行这2个接口
return
}
f(w, req)
}
}
func withVerify2(onSuccess func(w http.ResponseWriter, req *http.Request), onFailure func(w http.ResponseWriter, req *http.Request)) func(http.ResponseWriter, *http.Request) {
return func(w http.ResponseWriter, req *http.Request) {
cookie, err := req.Cookie("token")
if err == nil && TokenManager.Refresh(cookie.Value, time.Now()) {
onSuccess(w, req)
} else {
onFailure(w, req)
}
}
}
func StartApiServer(addr string) {
apiServer.router.HandleFunc("/api/v1/hook/on_play", common.WithJsonParams(apiServer.OnPlay, &PlayDoneParams{}))
apiServer.router.HandleFunc("/api/v1/hook/on_play_done", common.WithJsonParams(apiServer.OnPlayDone, &PlayDoneParams{}))
apiServer.router.HandleFunc("/api/v1/hook/on_publish", common.WithJsonParams(apiServer.OnPublish, &StreamParams{}))
apiServer.router.HandleFunc("/api/v1/hook/on_publish_done", common.WithJsonParams(apiServer.OnPublishDone, &StreamParams{}))
apiServer.router.HandleFunc("/api/v1/hook/on_idle_timeout", common.WithJsonParams(apiServer.OnIdleTimeout, &StreamParams{}))
apiServer.router.HandleFunc("/api/v1/hook/on_receive_timeout", common.WithJsonParams(apiServer.OnReceiveTimeout, &StreamParams{}))
apiServer.router.HandleFunc("/api/v1/hook/on_record", common.WithJsonParams(apiServer.OnRecord, &RecordParams{}))
apiServer.router.HandleFunc("/api/v1/hook/on_started", apiServer.OnStarted)
apiServer.router.HandleFunc("/api/v1/stream/start", withVerify(common.WithFormDataParams(apiServer.OnStreamStart, InviteParams{}))) // 实时预览
apiServer.router.HandleFunc("/api/v1/stream/stop", withVerify(common.WithFormDataParams(apiServer.OnCloseLiveStream, InviteParams{}))) // 关闭实时预览
apiServer.router.HandleFunc("/api/v1/playback/start", withVerify(common.WithFormDataParams(apiServer.OnPlaybackStart, InviteParams{}))) // 回放/下载
apiServer.router.HandleFunc("/api/v1/playback/stop", withVerify(common.WithFormDataParams(apiServer.OnCloseStream, StreamIDParams{}))) // 关闭回放/下载
apiServer.router.HandleFunc("/api/v1/playback/control", withVerify(common.WithFormDataParams(apiServer.OnPlaybackControl, StreamIDParams{}))) // 回放控制
apiServer.router.HandleFunc("/api/v1/device/list", withVerify(common.WithQueryStringParams(apiServer.OnDeviceList, QueryDeviceChannel{}))) // 查询设备列表
apiServer.router.HandleFunc("/api/v1/device/channeltree", withVerify(common.WithQueryStringParams(apiServer.OnDeviceTree, QueryDeviceChannel{}))) // 设备树
apiServer.router.HandleFunc("/api/v1/device/channellist", withVerify(common.WithQueryStringParams(apiServer.OnChannelList, QueryDeviceChannel{}))) // 查询通道列表
apiServer.router.HandleFunc("/api/v1/device/fetchcatalog", withVerify(common.WithQueryStringParams(apiServer.OnCatalogQuery, QueryDeviceChannel{}))) // 更新通道
apiServer.router.HandleFunc("/api/v1/device/remove", withVerify(common.WithFormDataParams(apiServer.OnDeviceRemove, DeleteDevice{}))) // 删除设备
apiServer.router.HandleFunc("/api/v1/device/setmediatransport", withVerify(common.WithFormDataParams(apiServer.OnDeviceMediaTransportSet, SetMediaTransportReq{}))) // 设置设备媒体传输模式
apiServer.router.HandleFunc("/api/v1/playback/recordlist", withVerify(common.WithQueryStringParams(apiServer.OnRecordList, QueryRecordParams{}))) // 查询录像列表
apiServer.router.HandleFunc("/api/v1/stream/info", withVerify(apiServer.OnStreamInfo))
apiServer.router.HandleFunc("/api/v1/playback/streaminfo", withVerify(apiServer.OnStreamInfo))
apiServer.router.HandleFunc("/api/v1/device/session/list", withVerify(common.WithQueryStringParams(apiServer.OnSessionList, QueryDeviceChannel{}))) // 推流列表
apiServer.router.HandleFunc("/api/v1/device/session/stop", withVerify(common.WithFormDataParams(apiServer.OnSessionStop, StreamIDParams{}))) // 关闭流
apiServer.router.HandleFunc("/api/v1/device/setchannelid", withVerify(common.WithFormDataParams(apiServer.OnCustomChannelSet, CustomChannel{}))) // 自定义通道ID
apiServer.router.HandleFunc("/api/v1/playback/seek", common.WithJsonResponse(apiServer.OnSeekPlayback, &SeekParams{})) // 回放seek
apiServer.router.HandleFunc("/api/v1/control/ptz", withVerify(common.WithFormDataParams(apiServer.OnPTZControl, QueryRecordParams{}))) // 云台控制
apiServer.router.HandleFunc("/api/v1/cascade/list", withVerify(common.WithQueryStringParams(apiServer.OnPlatformList, QueryDeviceChannel{}))) // 级联设备列表
apiServer.router.HandleFunc("/api/v1/cascade/save", withVerify(common.WithFormDataParams(apiServer.OnPlatformAdd, LiveGBSCascade{}))) // 添加级联设备
apiServer.router.HandleFunc("/api/v1/cascade/setenable", withVerify(common.WithFormDataParams(apiServer.OnEnableSet, SetEnable{}))) // 添加级联设备
apiServer.router.HandleFunc("/api/v1/cascade/remove", withVerify(common.WithFormDataParams(apiServer.OnPlatformRemove, SetEnable{}))) // 删除级联设备
apiServer.router.HandleFunc("/api/v1/cascade/channellist", withVerify(common.WithQueryStringParams(apiServer.OnPlatformChannelList, QueryCascadeChannelList{}))) // 级联设备通道列表
apiServer.router.HandleFunc("/api/v1/cascade/savechannels", withVerify(apiServer.OnPlatformChannelBind)) // 级联绑定通道
apiServer.router.HandleFunc("/api/v1/cascade/removechannels", withVerify(apiServer.OnPlatformChannelUnbind)) // 级联解绑通道
apiServer.router.HandleFunc("/api/v1/cascade/setshareallchannel", withVerify(common.WithFormDataParams(apiServer.OnShareAllChannel, SetEnable{}))) // 开启或取消级联所有通道
apiServer.router.HandleFunc("/api/v1/cascade/pushcatalog", withVerify(common.WithFormDataParams(apiServer.OnCatalogPush, SetEnable{}))) // 推送目录
apiServer.router.HandleFunc("/api/v1/device/setinfo", withVerify(common.WithFormDataParams(apiServer.OnDeviceInfoSet, DeviceInfo{}))) // 编辑设备信息
apiServer.router.HandleFunc("/api/v1/alarm/list", withVerify(common.WithQueryStringParams(apiServer.OnAlarmList, QueryDeviceChannel{}))) // 报警查询
apiServer.router.HandleFunc("/api/v1/alarm/remove", withVerify(common.WithFormDataParams(apiServer.OnAlarmRemove, SetEnable{}))) // 删除报警
apiServer.router.HandleFunc("/api/v1/alarm/clear", withVerify(common.WithFormDataParams(apiServer.OnAlarmClear, Empty{}))) // 清空报警
// 暂未开发
apiServer.router.HandleFunc("/api/v1/sms/list", withVerify(func(w http.ResponseWriter, req *http.Request) {})) // 流媒体服务器列表
apiServer.router.HandleFunc("/api/v1/cloudrecord/querychannels", withVerify(func(w http.ResponseWriter, req *http.Request) {})) // 云端录像
apiServer.router.HandleFunc("/api/v1/user/list", withVerify(func(w http.ResponseWriter, req *http.Request) {})) // 用户管理
apiServer.router.HandleFunc("/api/v1/log/list", withVerify(func(w http.ResponseWriter, req *http.Request) {})) // 操作日志
apiServer.router.HandleFunc("/api/v1/getbaseconfig", withVerify(func(w http.ResponseWriter, req *http.Request) {}))
apiServer.router.HandleFunc("/api/v1/gm/cert/list", withVerify(func(w http.ResponseWriter, req *http.Request) {}))
apiServer.router.HandleFunc("/api/v1/getrequestkey", withVerify(func(w http.ResponseWriter, req *http.Request) {}))
apiServer.router.HandleFunc("/api/v1/record/start", withVerify(apiServer.OnRecordStart)) // 开启录制
apiServer.router.HandleFunc("/api/v1/record/stop", withVerify(apiServer.OnRecordStop)) // 关闭录制
apiServer.router.HandleFunc("/api/v1/broadcast/invite", common.WithJsonResponse(apiServer.OnBroadcast, &BroadcastParams{Setup: &common.DefaultSetupType})) // 发起语音广播
apiServer.router.HandleFunc("/api/v1/broadcast/hangup", common.WithJsonResponse(apiServer.OnHangup, &BroadcastParams{})) // 挂断广播会话
apiServer.router.HandleFunc("/api/v1/control/ws-talk/{device}/{channel}", withVerify(apiServer.OnTalk)) // 一对一语音对讲
apiServer.router.HandleFunc("/api/v1/jt/device/add", common.WithJsonResponse(apiServer.OnVirtualDeviceAdd, &dao.JTDeviceModel{}))
apiServer.router.HandleFunc("/api/v1/jt/device/edit", common.WithJsonResponse(apiServer.OnVirtualDeviceEdit, &dao.JTDeviceModel{}))
apiServer.router.HandleFunc("/api/v1/jt/device/remove", common.WithJsonResponse(apiServer.OnVirtualDeviceRemove, &dao.JTDeviceModel{}))
apiServer.router.HandleFunc("/api/v1/jt/device/list", common.WithJsonResponse(apiServer.OnVirtualDeviceList, &PageQuery{}))
apiServer.router.HandleFunc("/api/v1/jt/channel/add", common.WithJsonResponse(apiServer.OnVirtualChannelAdd, &dao.ChannelModel{}))
apiServer.router.HandleFunc("/api/v1/jt/channel/edit", common.WithJsonResponse(apiServer.OnVirtualChannelEdit, &dao.ChannelModel{}))
apiServer.router.HandleFunc("/api/v1/jt/channel/remove", common.WithJsonResponse(apiServer.OnVirtualChannelRemove, &dao.ChannelModel{}))
apiServer.router.HandleFunc("/logout", func(writer http.ResponseWriter, req *http.Request) {
cookie, err := req.Cookie("token")
if err == nil {
TokenManager.Remove(cookie.Value)
writer.Header().Set("Location", "/login.html")
writer.WriteHeader(http.StatusFound)
return
}
})
registerLiveGBSApi()
// 前端路由
htmlRoot := "./html/"
fileServer := http.FileServer(http.Dir(htmlRoot))
apiServer.router.PathPrefix("/").HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
// 处理无扩展名的路径，自动添加.html扩展名
path := request.URL.Path
if !strings.Contains(path, ".") {
// 检查是否存在对应的.html文件
htmlPath := htmlRoot + path + ".html"
if _, err := os.Stat(htmlPath); err == nil {
// 如果存在对应的.html文件，则直接返回该文件
http.ServeFile(writer, request, htmlPath)
return
}
}
fileServer.ServeHTTP(writer, request)
})
srv := &http.Server{
Handler: apiServer.router,
Addr: addr,
// Good practice: enforce timeouts for servers you create!
WriteTimeout: 30 * time.Second,
ReadTimeout: 30 * time.Second,
}
err := srv.ListenAndServe()
if err != nil {
panic(err)
}
}

74
api/api_alarm.go Normal file
View File

@@ -0,0 +1,74 @@
package api
import (
"gb-cms/common"
"gb-cms/dao"
"net/http"
)
func (api *ApiServer) OnAlarmList(q *QueryDeviceChannel, _ http.ResponseWriter, _ *http.Request) (interface{}, error) {
if q.Limit < 1 {
q.Limit = 10
}
conditions := make(map[string]interface{}, 0)
// 报警查询参数
if q.Keyword != "" {
conditions["q"] = q.Keyword
}
if q.StartTime != "" {
conditions["starttime"] = q.StartTime
}
if q.EndTime != "" {
conditions["endtime"] = q.EndTime
}
if q.Priority > 0 {
conditions["alarm_priority"] = q.Priority
}
if q.Method > 0 {
conditions["alarm_method"] = q.Method
}
alarms, count, err := dao.Alarm.QueryAlarmList((q.Start/q.Limit)+1, q.Limit, conditions)
if err != nil {
return nil, err
}
v := struct {
AlarmCount int
AlarmList []*dao.AlarmModel
AlarmPublishToRedis bool
AlarmReserveDays int
}{
AlarmCount: count,
AlarmList: alarms,
AlarmPublishToRedis: true,
AlarmReserveDays: common.Config.AlarmReserveDays,
}
return &v, nil
}
func (api *ApiServer) OnAlarmRemove(params *SetEnable, _ http.ResponseWriter, _ *http.Request) (interface{}, error) {
// 删除报警
if err := dao.Alarm.DeleteAlarm(params.ID); err != nil {
return nil, err
}
return "OK", nil
}
func (api *ApiServer) OnAlarmClear(_ *Empty, _ http.ResponseWriter, req *http.Request) (interface{}, error) {
// 清空报警
if err := dao.Alarm.ClearAlarm(); err != nil {
return nil, err
}
return "OK", nil
}

342
api/api_cascade.go Normal file
View File

@@ -0,0 +1,342 @@
package api
import (
"errors"
"fmt"
"gb-cms/common"
"gb-cms/dao"
"gb-cms/log"
"gb-cms/stack"
"net"
"net/http"
"strconv"
)
func (api *ApiServer) OnPlatformAdd(v *LiveGBSCascade, _ http.ResponseWriter, _ *http.Request) (interface{}, error) {
log.Sugar.Debugf("添加级联设备 %v", *v)
if v.Username == "" {
v.Username = common.Config.SipID
log.Sugar.Infof("级联设备使用本级域: %s", common.Config.SipID)
}
var err error
if len(v.Username) != 20 {
err = fmt.Errorf("用户名长度必须20位")
return nil, err
} else if len(v.Serial) != 20 {
err = fmt.Errorf("上级ID长度必须20位")
return nil, err
}
if err != nil {
log.Sugar.Errorf("添加级联设备失败 err: %s", err.Error())
return nil, err
}
v.Status = "OFF"
model := dao.PlatformModel{
SIPUAOptions: common.SIPUAOptions{
Name: v.Name,
Username: v.Username,
Password: v.Password,
ServerID: v.Serial,
ServerAddr: net.JoinHostPort(v.Host, strconv.Itoa(v.Port)),
Transport: v.CommandTransport,
RegisterExpires: v.RegisterInterval,
KeepaliveInterval: v.KeepaliveInterval,
Status: common.OFF,
},
Enable: v.Enable,
}
platform, err := stack.NewPlatform(&model.SIPUAOptions, common.SipStack)
if err != nil {
return nil, err
}
// 编辑国标设备
if v.ID != "" {
// 停止旧的
oldPlatform := stack.PlatformManager.Remove(model.ServerAddr)
if oldPlatform != nil {
oldPlatform.Stop()
}
// 更新数据库
id, _ := strconv.ParseInt(v.ID, 10, 64)
model.ID = uint(id)
err = dao.Platform.UpdatePlatform(&model)
} else {
err = dao.Platform.SavePlatform(&model)
}
if err == nil && v.Enable {
if !stack.PlatformManager.Add(model.ServerAddr, platform) {
err = fmt.Errorf("地址冲突. key: %s", model.ServerAddr)
if err != nil {
_ = dao.Platform.DeletePlatformByAddr(model.ServerAddr)
}
} else {
platform.Start()
}
}
if err != nil {
log.Sugar.Errorf("添加级联设备失败 err: %s", err.Error())
return nil, err
}
return "OK", nil
}
func (api *ApiServer) OnPlatformRemove(v *SetEnable, _ http.ResponseWriter, _ *http.Request) (interface{}, error) {
log.Sugar.Debugf("删除级联设备 %v", *v)
platform, _ := dao.Platform.QueryPlatformByID(v.ID)
if platform == nil {
return nil, fmt.Errorf("级联设备不存在")
}
_ = dao.Platform.DeletePlatformByID(v.ID)
client := stack.PlatformManager.Remove(platform.ServerAddr)
if client != nil {
client.Stop()
}
return "OK", nil
}
func (api *ApiServer) OnPlatformList(q *QueryDeviceChannel, _ http.ResponseWriter, _ *http.Request) (interface{}, error) {
// 分页参数
if q.Limit < 1 {
q.Limit = 10
}
response := struct {
CascadeCount int `json:"CascadeCount"`
CascadeList []*LiveGBSCascade `json:"CascadeList"`
}{}
platforms, total, err := dao.Platform.QueryPlatforms((q.Start/q.Limit)+1, q.Limit, q.Keyword, q.Enable, q.Online)
if err == nil {
response.CascadeCount = total
for _, platform := range platforms {
host, p, _ := net.SplitHostPort(platform.ServerAddr)
port, _ := strconv.Atoi(p)
response.CascadeList = append(response.CascadeList, &LiveGBSCascade{
ID: strconv.Itoa(int(platform.ID)),
Enable: platform.Enable,
Name: platform.Name,
Serial: platform.ServerID,
Realm: platform.ServerID[:10],
Host: host,
Port: port,
LocalSerial: platform.Username,
Username: platform.Username,
Password: platform.Password,
Online: platform.Status == common.ON,
Status: platform.Status,
RegisterInterval: platform.RegisterExpires,
KeepaliveInterval: platform.KeepaliveInterval,
CommandTransport: platform.Transport,
Charset: "GB2312",
CatalogGroupSize: 1,
LoadLimit: 0,
CivilCodeLimit: 8,
DigestAlgorithm: "",
GM: false,
Cert: "***",
CreatedAt: platform.CreatedAt.Format("2006-01-02 15:04:05"),
UpdatedAt: platform.UpdatedAt.Format("2006-01-02 15:04:05"),
})
}
}
return response, nil
}
func (api *ApiServer) OnPlatformChannelBind(w http.ResponseWriter, r *http.Request) {
idStr := r.FormValue("id")
channels := r.Form["channels[]"]
var err error
id, _ := strconv.Atoi(idStr)
_, err = dao.Platform.QueryPlatformByID(id)
if err == nil {
err = dao.Platform.BindChannels(id, channels)
}
if err != nil {
w.WriteHeader(http.StatusBadRequest)
_ = common.HttpResponseJson(w, err.Error())
} else {
_ = common.HttpResponseJson(w, "OK")
}
}
func (api *ApiServer) OnPlatformChannelUnbind(w http.ResponseWriter, r *http.Request) {
idStr := r.FormValue("id")
channels := r.Form["channels[]"]
var err error
id, _ := strconv.Atoi(idStr)
_, err = dao.Platform.QueryPlatformByID(id)
if err == nil {
err = dao.Platform.UnbindChannels(id, channels)
}
if err != nil {
w.WriteHeader(http.StatusBadRequest)
_ = common.HttpResponseJson(w, err.Error())
} else {
_ = common.HttpResponseJson(w, "OK")
}
}
func (api *ApiServer) OnEnableSet(params *SetEnable, _ http.ResponseWriter, _ *http.Request) (interface{}, error) {
model, err := dao.Platform.QueryPlatformByID(params.ID)
if err != nil {
return nil, err
}
err = dao.Platform.UpdateEnable(params.ID, params.Enable)
if err != nil {
return nil, err
}
if params.Enable {
if stack.PlatformManager.Find(model.ServerAddr) != nil {
return nil, errors.New("device already started")
}
platform, err := stack.NewPlatform(&model.SIPUAOptions, common.SipStack)
if err != nil {
_ = dao.Platform.UpdateEnable(params.ID, false)
return nil, err
}
stack.PlatformManager.Add(platform.ServerAddr, platform)
platform.Start()
} else if client := stack.PlatformManager.Remove(model.ServerAddr); client != nil {
client.Stop()
}
return "OK", nil
}
func (api *ApiServer) OnPlatformChannelList(q *QueryCascadeChannelList, w http.ResponseWriter, req *http.Request) (interface{}, error) {
response := struct {
ChannelCount int `json:"ChannelCount"`
ChannelList []*CascadeChannel `json:"ChannelList"`
ChannelRelateCount *int `json:"ChannelRelateCount,omitempty"`
ShareAllChannel *bool `json:"ShareAllChannel,omitempty"`
}{}
id, err := strconv.Atoi(q.ID)
if err != nil {
return nil, err
}
// livegbs前端, 如果开启级联所有通道, 是不允许再只看已选择或取消绑定通道
platform, err := dao.Platform.QueryPlatformByID(id)
if err != nil {
return nil, err
}
// 只看已选择
if q.Related == true {
list, total, err := dao.Platform.QueryPlatformChannelList(id)
if err != nil {
return nil, err
}
response.ChannelCount = total
ChannelList := ChannelModels2LiveGBSChannels(q.Start+1, list, "")
for _, channel := range ChannelList {
response.ChannelList = append(response.ChannelList, &CascadeChannel{
CascadeID: q.ID,
LiveGBSChannel: channel,
})
}
} else {
list, err := api.OnChannelList(&q.QueryDeviceChannel, w, req)
if err != nil {
return nil, err
}
result := list.(*ChannelListResult)
response.ChannelCount = result.ChannelCount
for _, channel := range result.ChannelList {
var cascadeId string
if exist, _ := dao.Platform.QueryPlatformChannelExist(id, channel.DeviceID, channel.ID); exist {
cascadeId = q.ID
}
// 判断该通道是否选中
response.ChannelList = append(response.ChannelList, &CascadeChannel{
cascadeId, channel,
})
}
response.ChannelRelateCount = new(int)
response.ShareAllChannel = new(bool)
// 级联设备通道总数
if count, err := dao.Platform.QueryPlatformChannelCount(id); err != nil {
return nil, err
} else {
response.ChannelRelateCount = &count
}
*response.ShareAllChannel = platform.ShareAll
}
return &response, nil
}
func (api *ApiServer) OnShareAllChannel(q *SetEnable, _ http.ResponseWriter, _ *http.Request) (interface{}, error) {
var err error
if q.ShareAllChannel {
// 删除所有已经绑定的通道, 设置级联所有通道为true
if err = dao.Platform.DeletePlatformChannels(q.ID); err == nil {
err = dao.Platform.SetShareAllChannel(q.ID, true)
}
} else {
// 设置级联所有通道为false
err = dao.Platform.SetShareAllChannel(q.ID, false)
}
if err != nil {
return nil, err
}
return "OK", nil
}
func (api *ApiServer) OnCustomChannelSet(q *CustomChannel, _ http.ResponseWriter, _ *http.Request) (interface{}, error) {
if len(q.CustomID) != 20 {
return nil, fmt.Errorf("20位国标ID")
}
if err := dao.Channel.UpdateCustomID(q.DeviceID, q.ChannelID, q.CustomID); err != nil {
return nil, err
}
return "OK", nil
}
func (api *ApiServer) OnCatalogPush(params *SetEnable, _ http.ResponseWriter, _ *http.Request) (interface{}, error) {
// 使用notify发送目录列表
model, err := dao.Platform.QueryPlatformByID(params.ID)
if err != nil {
return nil, err
} else if client := stack.PlatformManager.Find(model.ServerAddr); client != nil {
client.PushCatalog()
return "OK", nil
} else {
return nil, errors.New("device not found")
}
}

434
api/api_device.go Normal file
View File

@@ -0,0 +1,434 @@
package api
import (
"context"
"fmt"
"gb-cms/common"
"gb-cms/dao"
"gb-cms/log"
"gb-cms/stack"
"math"
"net/http"
"strconv"
"strings"
"time"
)
func (api *ApiServer) OnDeviceList(q *QueryDeviceChannel, _ http.ResponseWriter, r *http.Request) (interface{}, error) {
// 分页参数
if q.Limit < 1 {
q.Limit = 10
}
var status string
if "" == q.Online {
} else if "true" == q.Online {
status = "ON"
} else if "false" == q.Online {
status = "OFF"
}
if "desc" != q.Order {
q.Order = "asc"
}
devices, total, err := dao.Device.QueryDevices((q.Start/q.Limit)+1, q.Limit, status, q.Keyword, q.Order)
if err != nil {
log.Sugar.Errorf("查询设备列表失败 err: %s", err.Error())
return nil, err
}
response := struct {
DeviceCount int
DeviceList_ []LiveGBSDevice `json:"DeviceList"`
}{
DeviceCount: total,
}
// livgbs设备离线后的最后心跳时间, 涉及到是否显示非法设备的批量删除按钮
offlineTime := time.Date(1, time.January, 1, 0, 0, 0, 0, time.UTC).Format("2006-01-02 15:04:05")
for _, device := range devices {
// 更新正在查询通道的进度
var catalogProgress string
data := stack.UniqueTaskManager.Find(stack.GenerateCatalogTaskID(device.GetID()))
if data != nil {
catalogSize := data.(*stack.CatalogProgress)
if catalogSize.TotalSize > 0 {
catalogProgress = fmt.Sprintf("%d/%d", catalogSize.RecvSize, catalogSize.TotalSize)
}
}
var lastKeealiveTime string
if device.Online() {
lastKeealiveTime = device.LastHeartbeat.Format("2006-01-02 15:04:05")
} else {
lastKeealiveTime = offlineTime
}
if device.CatalogInterval < 1 {
device.CatalogInterval = dao.DefaultCatalogInterval
}
response.DeviceList_ = append(response.DeviceList_, LiveGBSDevice{
AlarmSubscribe: device.AlarmSubscribe, // 报警订阅
CatalogInterval: device.CatalogInterval, // 目录刷新时间
CatalogProgress: catalogProgress,
CatalogSubscribe: device.CatalogSubscribe, // 目录订阅
ChannelCount: device.ChannelsTotal,
ChannelOverLoad: false,
Charset: "GB2312",
CivilCodeFirst: false,
CommandTransport: device.Transport,
ContactIP: "",
CreatedAt: device.CreatedAt.Format("2006-01-02 15:04:05"),
CustomName: "",
DropChannelType: "",
GBVer: "",
ID: device.GetID(),
KeepOriginalTree: false,
LastKeepaliveAt: lastKeealiveTime,
LastRegisterAt: device.RegisterTime.Format("2006-01-02 15:04:05"),
Latitude: 0,
Longitude: 0,
//Manufacturer: device.Manufacturer,
Manufacturer: device.UserAgent,
MediaTransport: device.Setup.Transport(),
MediaTransportMode: device.Setup.String(),
Name: device.Name,
Online: device.Online(),
PTZSubscribe: false, // PTZ订阅2022
Password: "",
PositionSubscribe: device.PositionSubscribe, // 位置订阅
RecordCenter: false,
RecordIndistinct: false,
RecvStreamIP: "",
RemoteIP: device.RemoteIP,
RemotePort: device.RemotePort,
RemoteRegion: "",
SMSGroupID: "",
SMSID: "",
StreamMode: "",
SubscribeInterval: 0,
Type: "GB",
UpdatedAt: device.UpdatedAt.Format("2006-01-02 15:04:05"),
})
}
return &response, nil
}
func (api *ApiServer) OnChannelList(q *QueryDeviceChannel, _ http.ResponseWriter, r *http.Request) (interface{}, error) {
// 分页参数
if q.Limit < 1 {
q.Limit = 10
}
var deviceName string
if q.DeviceID != "" {
device, err := dao.Device.QueryDevice(q.DeviceID)
if err != nil {
log.Sugar.Errorf("查询设备失败 err: %s", err.Error())
return nil, err
}
deviceName = device.Name
}
var status string
if "" == q.Online {
} else if "true" == q.Online {
status = "ON"
} else if "false" == q.Online {
status = "OFF"
}
if "desc" != q.Order {
q.Order = "asc"
}
channels, total, err := dao.Channel.QueryChannels(q.DeviceID, q.GroupID, (q.Start/q.Limit)+1, q.Limit, status, q.Keyword, q.Order, q.Sort, q.ChannelType == "dir")
if err != nil {
log.Sugar.Errorf("查询通道列表失败 err: %s", err.Error())
return nil, err
}
response := ChannelListResult{
ChannelCount: total,
}
index := q.Start + 1
response.ChannelList = ChannelModels2LiveGBSChannels(index, channels, deviceName)
return &response, nil
}
func (api *ApiServer) OnRecordList(v *QueryRecordParams, _ http.ResponseWriter, _ *http.Request) (interface{}, error) {
log.Sugar.Debugf("查询录像列表 %v", *v)
model, _ := dao.Device.QueryDevice(v.DeviceID)
if model == nil || !model.Online() {
log.Sugar.Errorf("查询录像列表失败, 设备离线 device: %s", v.DeviceID)
return nil, fmt.Errorf("设备离线")
}
device := &stack.Device{DeviceModel: model}
sn := stack.GetSN()
err := device.QueryRecord(v.ChannelID, v.StartTime, v.EndTime, sn, "all")
if err != nil {
log.Sugar.Errorf("发送查询录像请求失败 err: %s", err.Error())
return nil, err
}
// 设置查询超时时长
timeout := int(math.Max(math.Min(5, float64(v.Timeout)), 60))
withTimeout, cancelFunc := context.WithTimeout(context.Background(), time.Duration(timeout)*time.Second)
var recordList []stack.RecordInfo
stack.SNManager.AddEvent(sn, func(data interface{}) {
response := data.(*stack.QueryRecordInfoResponse)
if len(response.DeviceList.Devices) > 0 {
recordList = append(recordList, response.DeviceList.Devices...)
}
// 所有记录响应完毕
if len(recordList) >= response.SumNum {
cancelFunc()
}
})
select {
case _ = <-withTimeout.Done():
break
}
response := struct {
DeviceID string
Name string
RecordList []struct {
DeviceID string
EndTime string
FileSize uint64
Name string
Secrecy string
StartTime string
Type string
}
SumNum int `json:"sumNum"`
}{
DeviceID: v.DeviceID,
Name: model.Name,
SumNum: len(recordList),
}
for _, record := range recordList {
log.Sugar.Infof("查询录像列表 %v", record)
response.RecordList = append(response.RecordList, struct {
DeviceID string
EndTime string
FileSize uint64
Name string
Secrecy string
StartTime string
Type string
}{
DeviceID: record.DeviceID,
EndTime: record.EndTime,
FileSize: record.FileSize,
Name: record.Name,
Secrecy: record.Secrecy,
StartTime: record.StartTime,
Type: record.Type,
})
}
return &response, nil
}
func (api *ApiServer) OnDeviceMediaTransportSet(req *SetMediaTransportReq, _ http.ResponseWriter, _ *http.Request) (interface{}, error) {
var setupType common.SetupType
if "udp" == strings.ToLower(req.MediaTransport) {
setupType = common.SetupTypeUDP
} else if "passive" == strings.ToLower(req.MediaTransportMode) {
setupType = common.SetupTypePassive
} else if "active" == strings.ToLower(req.MediaTransportMode) {
setupType = common.SetupTypeActive
} else {
return nil, fmt.Errorf("media_transport_mode error")
}
err := dao.Device.UpdateMediaTransport(req.DeviceID, setupType)
if err != nil {
return nil, err
}
return "OK", nil
}
func (api *ApiServer) OnCatalogQuery(params *QueryDeviceChannel, _ http.ResponseWriter, _ *http.Request) (interface{}, error) {
deviceModel, err := dao.Device.QueryDevice(params.DeviceID)
if err != nil {
return nil, err
}
if deviceModel == nil {
return nil, fmt.Errorf("not found device")
}
list, err := (&stack.Device{DeviceModel: deviceModel}).QueryCatalog(15)
if err != nil {
return nil, err
}
response := struct {
ChannelCount int `json:"ChannelCount"`
ChannelList []*dao.ChannelModel `json:"ChannelList"`
}{
ChannelCount: len(list),
ChannelList: list,
}
return &response, nil
}
func (api *ApiServer) OnDeviceTree(q *QueryDeviceChannel, _ http.ResponseWriter, _ *http.Request) (interface{}, error) {
var response []*LiveGBSDeviceTree
// 查询所有设备
if q.DeviceID == "" && q.PCode == "" {
devices, err := dao.Device.LoadDevices()
if err != nil {
return nil, err
}
for _, model := range devices {
count, _ := dao.Channel.QueryChanelCount(model.DeviceID, true)
deviceCount, _ := dao.Channel.QueryChanelCount(model.DeviceID, false)
onlineCount, _ := dao.Channel.QueryOnlineChanelCount(model.DeviceID, false)
response = append(response, &LiveGBSDeviceTree{Code: "", Custom: false, CustomID: "", CustomName: "", ID: model.DeviceID, Latitude: 0, Longitude: 0, Manufacturer: model.Manufacturer, Name: model.Name, OnlineSubCount: onlineCount, Parental: true, PtzType: 0, Serial: model.DeviceID, Status: model.Status.String(), SubCount: count, SubCountDevice: deviceCount})
}
} else {
// 查询设备下的某个目录的所有通道
if q.PCode == "" {
q.PCode = q.DeviceID
}
channels, _, _ := dao.Channel.QueryChannels(q.DeviceID, q.PCode, -1, 0, "", "", "asc", "", false)
for _, channel := range channels {
id := channel.RootID + ":" + channel.DeviceID
latitude, _ := strconv.ParseFloat(channel.Latitude, 10)
longitude, _ := strconv.ParseFloat(channel.Longitude, 10)
var deviceCount int
var onlineCount int
if channel.SubCount > 0 {
deviceCount, _ = dao.Channel.QuerySubChannelCount(channel.RootID, channel.DeviceID, false)
onlineCount, _ = dao.Channel.QueryOnlineSubChannelCount(channel.RootID, channel.DeviceID, false)
}
response = append(response, &LiveGBSDeviceTree{Code: channel.DeviceID, Custom: false, CustomID: "", CustomName: "", ID: id, Latitude: latitude, Longitude: longitude, Manufacturer: channel.Manufacturer, Name: channel.Name, OnlineSubCount: onlineCount, Parental: false, PtzType: 0, Serial: channel.RootID, Status: channel.Status.String(), SubCount: channel.SubCount, SubCountDevice: deviceCount})
}
}
return &response, nil
}
func (api *ApiServer) OnDeviceRemove(q *DeleteDevice, _ http.ResponseWriter, _ *http.Request) (interface{}, error) {
var err error
if q.IP != "" {
// 删除IP下的所有设备
err = dao.Device.DeleteDevicesByIP(q.IP)
} else if q.UA != "" {
// 删除UA下的所有设备
err = dao.Device.DeleteDevicesByUA(q.UA)
} else {
// 删除单个设备
err = dao.Device.DeleteDevice(q.DeviceID)
}
if err != nil {
return nil, err
} else if q.Forbid {
if q.IP != "" {
// 拉黑IP
err = dao.Blacklist.SaveIP(q.IP)
} else if q.UA != "" {
// 拉黑UA
err = dao.Blacklist.SaveUA(q.UA)
}
}
return "OK", nil
}
func (api *ApiServer) OnDeviceInfoSet(params *DeviceInfo, w http.ResponseWriter, req *http.Request) (interface{}, error) {
model, err := dao.Device.QueryDevice(params.DeviceID)
if err != nil {
return nil, err
}
device := stack.Device{DeviceModel: model}
// 更新的字段和值
conditions := make(map[string]interface{}, 0)
// 刷新目录间隔
if params.CatalogInterval != model.CatalogInterval && params.CatalogInterval != dao.DefaultCatalogInterval && params.CatalogInterval >= 60 {
conditions["catalog_interval"] = params.CatalogInterval
}
if model.CatalogSubscribe != params.CatalogSubscribe {
conditions["catalog_subscribe"] = params.CatalogSubscribe
// 开启目录订阅
if params.CatalogSubscribe {
_ = device.SubscribeCatalog()
} else {
// 取消目录订阅
device.UnsubscribeCatalog()
}
}
if model.PositionSubscribe != params.PositionSubscribe {
conditions["position_subscribe"] = params.PositionSubscribe
// 开启位置订阅
if params.PositionSubscribe {
_ = device.SubscribePosition()
} else {
// 取消位置订阅
device.UnsubscribePosition()
}
}
if model.AlarmSubscribe != params.AlarmSubscribe {
conditions["alarm_subscribe"] = params.AlarmSubscribe
// 开启报警订阅
if params.AlarmSubscribe {
_ = device.SubscribeAlarm()
} else {
// 取消报警订阅
device.UnsubscribeAlarm()
}
}
// 更新设备信息
if len(conditions) > 0 {
if err = dao.Device.UpdateDevice(params.DeviceID, conditions); err != nil {
return nil, err
}
}
return "OK", nil
}
func (api *ApiServer) OnPTZControl(v *QueryRecordParams, _ http.ResponseWriter, _ *http.Request) (interface{}, error) {
log.Sugar.Debugf("PTZ控制 %v", *v)
model, _ := dao.Device.QueryDevice(v.DeviceID)
if model == nil || !model.Online() {
log.Sugar.Errorf("PTZ控制失败, 设备离线 device: %s", v.DeviceID)
return nil, fmt.Errorf("设备离线")
}
device := &stack.Device{DeviceModel: model}
device.ControlPTZ(v.Command, v.ChannelID)
return "OK", nil
}

229
api/api_hook.go Normal file
View File

@@ -0,0 +1,229 @@
package api
import (
"gb-cms/common"
"gb-cms/dao"
"gb-cms/hook"
"gb-cms/log"
"gb-cms/stack"
"github.com/lkmio/avformat/utils"
"net/http"
"strings"
)
func (api *ApiServer) OnPlay(params *PlayDoneParams, w http.ResponseWriter, r *http.Request) {
log.Sugar.Infof("播放事件. protocol: %s stream: %s", params.Protocol, params.Stream)
// [注意]: windows上使用cmd/power shell推拉流如果要携带多个参数, 请用双引号将与号引起来("&")
// session_id是为了同一个录像文件, 允许同时点播多个.当然如果实时流支持多路预览, 也是可以的.
//ffplay -i rtmp://127.0.0.1/34020000001320000001/34020000001310000001
//ffplay -i http://127.0.0.1:8080/34020000001320000001/34020000001310000001.flv?setup=passive
//ffplay -i http://127.0.0.1:8080/34020000001320000001/34020000001310000001.m3u8?setup=passive
//ffplay -i rtsp://test:123456@127.0.0.1/34020000001320000001/34020000001310000001?setup=passive
// 回放示例
//ffplay -i rtmp://127.0.0.1/34020000001320000001/34020000001310000001.session_id_0?setup=passive"&"stream_type=playback"&"start_time=2024-06-18T15:20:56"&"end_time=2024-06-18T15:25:56
//ffplay -i rtmp://127.0.0.1/34020000001320000001/34020000001310000001.session_id_0?setup=passive&stream_type=playback&start_time=2024-06-18T15:20:56&end_time=2024-06-18T15:25:56
// 拉流地址携带的参数
query := r.URL.Query()
jtSource := query.Get("forward_type") == "gateway_1078"
// 跳过非国标拉流
sourceStream := strings.Split(string(params.Stream), "/")
if !jtSource && (len(sourceStream) != 2 || len(sourceStream[0]) != 20 || len(sourceStream[1]) < 20) {
log.Sugar.Infof("跳过非国标拉流 stream: %s", params.Stream)
return
}
deviceId := sourceStream[0]
channelId := sourceStream[1]
if len(channelId) > 20 {
channelId = channelId[:20]
}
var code int
// 通知1078信令服务器
if jtSource {
if len(sourceStream) != 2 {
code = http.StatusBadRequest
log.Sugar.Errorf("1078信令服务器转发请求参数错误")
return
}
simNumber := sourceStream[0]
channelNumber := sourceStream[1]
response, err := hook.PostOnInviteEvent(simNumber, channelNumber)
if err != nil {
code = http.StatusInternalServerError
log.Sugar.Errorf("通知1078信令服务器失败 err: %s sim number: %s channel number: %s", err.Error(), simNumber, channelNumber)
} else if code = response.StatusCode; code != http.StatusOK {
log.Sugar.Errorf("通知1078信令服务器失败. 响应状态码: %d sim number: %s channel number: %s", response.StatusCode, simNumber, channelNumber)
}
} else {
// livegbs前端即使退出的播放，还是会拉流. 如果在hook中发起invite, 会造成不必要的请求.
// 流不存在, 返回404
if params.Protocol < stack.TransStreamGBCascaded {
// 播放授权
streamToken := query.Get("stream_token")
if TokenManager.Find(streamToken) == nil {
w.WriteHeader(http.StatusUnauthorized)
log.Sugar.Errorf("播放鉴权失败, token不存在 token: %s", streamToken)
} else if stream, _ := dao.Stream.QueryStream(params.Stream); stream == nil {
w.WriteHeader(http.StatusNotFound)
} else {
_ = dao.Sink.CreateSink(&dao.SinkModel{
SinkID: params.Sink,
StreamID: params.Stream,
Protocol: params.Protocol,
RemoteAddr: params.RemoteAddr,
})
}
return
} else if stack.TransStreamGBTalk == params.Protocol {
// 对讲/广播
w.WriteHeader(http.StatusOK)
return
}
// 级联, 在此处请求流
inviteParams := &InviteParams{
DeviceID: deviceId,
ChannelID: channelId,
StartTime: query.Get("start_time"),
EndTime: query.Get("end_time"),
Setup: strings.ToLower(query.Get("setup")),
Speed: query.Get("speed"),
streamId: params.Stream,
}
var stream *dao.StreamModel
var err error
streamType := strings.ToLower(query.Get("stream_type"))
if "playback" == streamType {
code, stream, err = api.DoInvite(common.InviteTypePlay, inviteParams, false)
} else if "download" == streamType {
code, stream, err = api.DoInvite(common.InviteTypeDownload, inviteParams, false)
} else {
code, stream, err = api.DoInvite(common.InviteTypePlay, inviteParams, false)
}
if err != nil {
log.Sugar.Errorf("请求流失败 err: %s", err.Error())
utils.Assert(http.StatusOK != code)
} else if http.StatusOK == code {
_ = stream.ID
_ = dao.Sink.CreateSink(&dao.SinkModel{
SinkID: params.Sink,
StreamID: params.Stream,
Protocol: params.Protocol,
RemoteAddr: params.RemoteAddr,
})
}
}
w.WriteHeader(code)
}
func (api *ApiServer) OnPlayDone(params *PlayDoneParams, _ http.ResponseWriter, _ *http.Request) {
log.Sugar.Debugf("播放结束事件. protocol: %s stream: %s", params.Protocol, params.Stream)
sink, _ := dao.Sink.DeleteSink(params.Sink)
if sink == nil {
return
}
// 级联断开连接, 向上级发送Bye请求
if params.Protocol == stack.TransStreamGBCascaded {
if platform := stack.PlatformManager.Find(sink.ServerAddr); platform != nil {
callID, _ := sink.Dialog.CallID()
platform.(*stack.Platform).CloseStream(callID.Value(), true, false)
}
} else {
(&stack.Sink{sink}).Close(true, false)
}
}
func (api *ApiServer) OnPublish(params *StreamParams, w http.ResponseWriter, _ *http.Request) {
log.Sugar.Debugf("推流事件. protocol: %s stream: %s", params.Protocol, params.Stream)
if stack.SourceTypeRtmp == params.Protocol {
return
}
stream := stack.EarlyDialogs.Find(string(params.Stream))
if stream != nil {
stream.Put(200)
} else {
log.Sugar.Infof("推流事件. 未找到stream. stream: %s", params.Stream)
}
// 创建stream
if params.Protocol == stack.SourceTypeGBTalk || params.Protocol == stack.SourceType1078 {
s := &dao.StreamModel{
StreamID: params.Stream,
Protocol: params.Protocol,
}
if params.Protocol != stack.SourceTypeGBTalk {
s.DeviceID = params.Stream.DeviceID()
s.ChannelID = params.Stream.ChannelID()
}
_, ok := dao.Stream.SaveStream(s)
if !ok {
log.Sugar.Errorf("处理推流事件失败, stream已存在. id: %s", params.Stream)
w.WriteHeader(http.StatusBadRequest)
return
}
}
}
func (api *ApiServer) OnPublishDone(params *StreamParams, _ http.ResponseWriter, _ *http.Request) {
log.Sugar.Debugf("推流结束事件. protocol: %s stream: %s", params.Protocol, params.Stream)
//stack.CloseStream(params.Stream, false)
//// 对讲websocket断开连接
//if stack.SourceTypeGBTalk == params.Protocol {
//
//}
}
func (api *ApiServer) OnIdleTimeout(params *StreamParams, w http.ResponseWriter, _ *http.Request) {
log.Sugar.Debugf("推流空闲超时事件. protocol: %s stream: %s", params.Protocol, params.Stream)
// 非rtmp空闲超时, 返回非200应答, 删除会话
if stack.SourceTypeRtmp != params.Protocol {
w.WriteHeader(http.StatusForbidden)
stack.CloseStream(params.Stream, false)
}
}
func (api *ApiServer) OnReceiveTimeout(params *StreamParams, w http.ResponseWriter, _ *http.Request) {
log.Sugar.Debugf("收流超时事件. protocol: %s stream: %s", params.Protocol, params.Stream)
// 非rtmp推流超时, 返回非200应答, 删除会话
if stack.SourceTypeRtmp != params.Protocol {
w.WriteHeader(http.StatusForbidden)
stack.CloseStream(params.Stream, false)
}
}
func (api *ApiServer) OnRecord(params *RecordParams, _ http.ResponseWriter, _ *http.Request) {
log.Sugar.Infof("录制事件. protocol: %s stream: %s path:%s ", params.Protocol, params.Stream, params.Path)
}
func (api *ApiServer) OnStarted(_ http.ResponseWriter, _ *http.Request) {
log.Sugar.Infof("lkm启动")
streams, _ := dao.Stream.DeleteStreams()
for _, stream := range streams {
(&stack.Stream{StreamModel: stream}).Close(true, false)
}
sinks, _ := dao.Sink.DeleteSinks()
for _, sink := range sinks {
(&stack.Sink{SinkModel: sink}).Close(true, false)
}
}

2
api_jt.go → api/api_jt.go
View File

@@ -1,4 +1,4 @@
package main
package api
import (
"fmt"

3
api_livegbs.go → api/api_livegbs.go
View File

@@ -1,4 +1,4 @@
package main
package api
import (
"fmt"
@@ -95,7 +95,6 @@ func FormatUptime(d time.Duration) string {
}
func registerLiveGBSApi() {
serverInfoBase := ServerInfoBase{
CopyrightText: fmt.Sprintf("Copyright © %d \u003ca href=\"//github.com/lkmio\" target=\"_blank\"\u003egithub.com/lkmio\u003c/a\u003e Released under MIT License", time.Now().Year()),
DemoUser: "",

334
api/api_stream.go Normal file
View File

@@ -0,0 +1,334 @@
package api
import (
"encoding/json"
"errors"
"fmt"
"gb-cms/common"
"gb-cms/dao"
"gb-cms/log"
"gb-cms/stack"
"github.com/ghettovoice/gosip/sip"
"net/http"
"net/url"
"strconv"
"strings"
"time"
)
func (api *ApiServer) OnStreamStart(v *InviteParams, w http.ResponseWriter, r *http.Request) (interface{}, error) {
return api.DoStartStream(v, w, r, "stream")
}
func (api *ApiServer) OnPlaybackStart(v *InviteParams, w http.ResponseWriter, r *http.Request) (interface{}, error) {
if v.Download {
return api.DoStartStream(v, w, r, "download")
} else {
return api.DoStartStream(v, w, r, "playback")
}
}
func (api *ApiServer) DoStartStream(v *InviteParams, w http.ResponseWriter, r *http.Request, action string) (interface{}, error) {
var code int
var stream *dao.StreamModel
var err error
if "playback" == action {
code, stream, err = apiServer.DoInvite(common.InviteTypePlayback, v, true)
} else if "download" == action {
code, stream, err = apiServer.DoInvite(common.InviteTypeDownload, v, true)
} else if "stream" == action {
code, stream, err = apiServer.DoInvite(common.InviteTypePlay, v, true)
} else {
return nil, fmt.Errorf("action not found")
}
if http.StatusOK != code {
log.Sugar.Errorf("请求流失败 err: %s", err.Error())
return nil, err
}
// 录像下载, 转发到streaminfo接口
if "download" == action {
if r.URL.RawQuery == "" {
r.URL.RawQuery = "streamid=" + string(v.streamId)
} else if r.URL.RawQuery != "" {
r.URL.RawQuery += "&streamid=" + string(v.streamId)
}
common.HttpForwardTo("/api/v1/stream/info", w, r)
return nil, nil
}
var urls map[string]string
urls = make(map[string]string, 10)
for _, streamUrl := range stream.Urls {
var streamName string
if strings.HasPrefix(streamUrl, "ws") {
streamName = "WS_FLV"
} else if strings.HasSuffix(streamUrl, ".flv") {
streamName = "FLV"
} else if strings.HasSuffix(streamUrl, ".m3u8") {
streamName = "HLS"
} else if strings.HasSuffix(streamUrl, ".rtc") {
streamName = "WEBRTC"
} else if strings.HasPrefix(streamUrl, "rtmp") {
streamName = "RTMP"
} else if strings.HasPrefix(streamUrl, "rtsp") {
streamName = "RTSP"
}
// 加上登录的token, 播放授权
streamUrl += "?stream_token=" + v.Token
// 兼容livegbs前端播放webrtc
if streamName == "WEBRTC" {
if strings.HasPrefix(streamUrl, "http") {
streamUrl = strings.Replace(streamUrl, "http", "webrtc", 1)
} else if strings.HasPrefix(streamUrl, "https") {
streamUrl = strings.Replace(streamUrl, "https", "webrtcs", 1)
}
streamUrl += "&wf=livegbs"
}
urls[streamName] = streamUrl
}
response := LiveGBSStream{
AudioEnable: false,
CDN: "",
CascadeSize: 0,
ChannelID: v.ChannelID,
ChannelName: "未读取通道名",
ChannelPTZType: 0,
CloudRecord: false,
DecodeSize: 0,
DeviceID: v.DeviceID,
Duration: 1,
FLV: urls["FLV"],
HLS: urls["HLS"],
InBitRate: 0,
InBytes: 0,
NumOutputs: 0,
Ondemand: true,
OutBytes: 0,
RTMP: urls["RTMP"],
RecordStartAt: "",
RelaySize: 0,
SMSID: "",
SnapURL: "",
SourceAudioCodecName: "",
SourceAudioSampleRate: 0,
SourceVideoCodecName: "",
SourceVideoFrameRate: 0,
SourceVideoHeight: 0,
SourceVideoWidth: 0,
StartAt: "",
StreamID: string(stream.StreamID),
Transport: "TCP",
VideoFrameCount: 0,
WEBRTC: urls["WEBRTC"],
WS_FLV: urls["WS_FLV"],
}
return response, err
}
// DoInvite 发起Invite请求
// @params sync 是否异步等待流媒体的publish事件(确认收到流), 目前请求流分两种方式，流媒体hook和http接口, hook方式同步等待确认收到流再应答, http接口直接应答成功。
func (api *ApiServer) DoInvite(inviteType common.InviteType, params *InviteParams, sync bool) (int, *dao.StreamModel, error) {
device, _ := dao.Device.QueryDevice(params.DeviceID)
if device == nil || !device.Online() {
return http.StatusNotFound, nil, fmt.Errorf("设备离线 id: %s", params.DeviceID)
}
// 解析回放或下载的时间范围参数
var startTimeSeconds string
var endTimeSeconds string
if common.InviteTypePlay != inviteType {
startTime, err := time.ParseInLocation("2006-01-02T15:04:05", params.StartTime, time.Local)
if err != nil {
return http.StatusBadRequest, nil, err
}
endTime, err := time.ParseInLocation("2006-01-02T15:04:05", params.EndTime, time.Local)
if err != nil {
return http.StatusBadRequest, nil, err
}
startTimeSeconds = strconv.FormatInt(startTime.Unix(), 10)
endTimeSeconds = strconv.FormatInt(endTime.Unix(), 10)
}
if params.streamId == "" {
params.streamId = common.GenerateStreamID(inviteType, device.GetID(), params.ChannelID, params.StartTime, params.EndTime)
}
if params.Setup == "" {
params.Setup = device.Setup.String()
}
// 解析回放或下载速度参数
speed, _ := strconv.Atoi(params.Speed)
if speed < 1 {
speed = 4
}
d := &stack.Device{DeviceModel: device}
stream, err := d.StartStream(inviteType, params.streamId, params.ChannelID, startTimeSeconds, endTimeSeconds, params.Setup, speed, sync)
if err != nil {
return http.StatusInternalServerError, nil, err
}
return http.StatusOK, stream, nil
}
func (api *ApiServer) OnCloseStream(v *StreamIDParams, _ http.ResponseWriter, _ *http.Request) (interface{}, error) {
stack.CloseStream(v.StreamID, true)
return "OK", nil
}
func (api *ApiServer) OnCloseLiveStream(v *InviteParams, _ http.ResponseWriter, _ *http.Request) (interface{}, error) {
id := common.GenerateStreamID(common.InviteTypePlay, v.DeviceID, v.ChannelID, "", "")
stack.CloseStream(id, true)
return "OK", nil
}
func (api *ApiServer) OnStreamInfo(w http.ResponseWriter, r *http.Request) {
common.HttpForwardTo("/api/v1/stream/info", w, r)
}
func (api *ApiServer) OnSessionList(q *QueryDeviceChannel, _ http.ResponseWriter, r *http.Request) (interface{}, error) {
// 分页参数
if q.Limit < 1 {
q.Limit = 10
}
//filter := q.Filter // playing-正在播放/stream-不包含回放和下载/record-正在回放的流/hevc-h265流/cascade-级联
var streams []*dao.StreamModel
var err error
if "cascade" == q.Filter {
protocols := []int{stack.TransStreamGBCascaded}
var ids []string
ids, _, err = dao.Sink.QueryStreamIds(protocols, (q.Start/q.Limit)+1, q.Limit)
if len(ids) > 0 {
streams, err = dao.Stream.QueryStreamsByIds(ids)
}
} else if "stream" == q.Filter {
streams, _, err = dao.Stream.QueryStreams(q.Keyword, (q.Start/q.Limit)+1, q.Limit, "play")
} else if "record" == q.Filter {
streams, _, err = dao.Stream.QueryStreams(q.Keyword, (q.Start/q.Limit)+1, q.Limit, "playback")
} else if "playing" == q.Filter {
protocols := []int{stack.TransStreamRtmp, stack.TransStreamFlv, stack.TransStreamRtsp, stack.TransStreamHls, stack.TransStreamRtc}
var ids []string
ids, _, err = dao.Sink.QueryStreamIds(protocols, (q.Start/q.Limit)+1, q.Limit)
if len(ids) > 0 {
streams, err = dao.Stream.QueryStreamsByIds(ids)
}
} else {
streams, _, err = dao.Stream.QueryStreams(q.Keyword, (q.Start/q.Limit)+1, q.Limit, "")
}
if err != nil {
return nil, err
}
response := struct {
SessionCount int
SessionList []*StreamInfo
}{}
bytes := make([]byte, 4096)
for _, stream := range streams {
values := url.Values{}
values.Set("streamid", string(stream.StreamID))
resp, err := stack.MSQueryStreamInfo(r.Header, values.Encode())
if err != nil {
return nil, err
}
var n int
n, err = resp.Body.Read(bytes)
_ = resp.Body.Close()
if n < 1 {
break
}
info := &StreamInfo{}
err = json.Unmarshal(bytes[:n], info)
if err != nil {
return nil, err
}
info.ChannelName = stream.Name
response.SessionList = append(response.SessionList, info)
}
response.SessionCount = len(response.SessionList)
return &response, nil
}
func (api *ApiServer) OnSessionStop(params *StreamIDParams, _ http.ResponseWriter, _ *http.Request) (interface{}, error) {
err := stack.MSCloseSource(string(params.StreamID))
if err != nil {
return nil, err
}
return "OK", nil
}
func (api *ApiServer) OnRecordStart(writer http.ResponseWriter, request *http.Request) {
common.HttpForwardTo("/api/v1/record/start", writer, request)
}
func (api *ApiServer) OnRecordStop(writer http.ResponseWriter, request *http.Request) {
common.HttpForwardTo("/api/v1/record/stop", writer, request)
}
func (api *ApiServer) OnPlaybackControl(params *StreamIDParams, _ http.ResponseWriter, _ *http.Request) (interface{}, error) {
if "scale" != params.Command || params.Scale <= 0 || params.Scale > 4 {
return nil, errors.New("scale error")
}
stream, err := dao.Stream.QueryStream(params.StreamID)
if err != nil {
return nil, err
} else if stream.Dialog == nil {
return nil, errors.New("stream not found")
}
// 查找設備
device, err := dao.Device.QueryDevice(stream.DeviceID)
if err != nil {
return nil, err
}
s := &stack.Device{DeviceModel: device}
s.ScalePlayback(stream.Dialog, params.Scale)
err = stack.MSSpeedSet(string(params.StreamID), params.Scale)
if err != nil {
return nil, err
}
return "OK", nil
}
func (api *ApiServer) OnSeekPlayback(v *SeekParams, _ http.ResponseWriter, _ *http.Request) (interface{}, error) {
log.Sugar.Debugf("快进回放 %v", *v)
model, _ := dao.Stream.QueryStream(v.StreamId)
if model == nil || model.Dialog == nil {
log.Sugar.Errorf("快进回放失败 stream不存在 %s", v.StreamId)
return nil, fmt.Errorf("stream不存在")
}
stream := &stack.Stream{StreamModel: model}
seekRequest := stream.CreateRequestFromDialog(sip.INFO)
seq, _ := seekRequest.CSeq()
body := fmt.Sprintf(stack.SeekBodyFormat, seq.SeqNo, v.Seconds)
seekRequest.SetBody(body, true)
seekRequest.RemoveHeader(stack.RtspMessageType.Name())
seekRequest.AppendHeader(&stack.RtspMessageType)
common.SipStack.SendRequest(seekRequest)
return nil, nil
}

85
api/api_talk.go Normal file
View File

@@ -0,0 +1,85 @@
package api
import (
"context"
"fmt"
"gb-cms/common"
"gb-cms/dao"
"gb-cms/log"
"gb-cms/stack"
"github.com/gorilla/mux"
"net/http"
"time"
)
func (api *ApiServer) OnHangup(v *BroadcastParams, _ http.ResponseWriter, _ *http.Request) (interface{}, error) {
log.Sugar.Debugf("广播挂断 %v", *v)
id := common.GenerateStreamID(common.InviteTypeBroadcast, v.DeviceID, v.ChannelID, "", "")
if sink, _ := dao.Sink.DeleteSinkBySinkStreamID(id); sink != nil {
(&stack.Sink{SinkModel: sink}).Close(true, true)
}
return nil, nil
}
func (api *ApiServer) OnBroadcast(v *BroadcastParams, _ http.ResponseWriter, r *http.Request) (interface{}, error) {
log.Sugar.Debugf("广播邀请 %v", *v)
model, _ := dao.Device.QueryDevice(v.DeviceID)
if model == nil || !model.Online() {
return nil, fmt.Errorf("设备离线")
}
// 主讲人id
//stream, _ := dao.Stream.QueryStream(v.StreamId)
//if stream == nil {
// return nil, fmt.Errorf("找不到主讲人")
//}
device := &stack.Device{DeviceModel: model}
_, err := device.StartBroadcast(v.StreamId, v.DeviceID, v.ChannelID, r.Context())
return nil, err
}
func (api *ApiServer) OnTalk(w http.ResponseWriter, r *http.Request) {
vars := mux.Vars(r)
deviceId := vars["device"]
channelId := vars["channel"]
_, online := stack.OnlineDeviceManager.Find(deviceId)
if !online {
w.WriteHeader(http.StatusBadRequest)
_ = common.HttpResponseJson(w, "设备离线")
return
}
model, err := dao.Device.QueryDevice(deviceId)
if err != nil {
w.WriteHeader(http.StatusBadRequest)
_ = common.HttpResponseJson(w, "设备不存在")
return
}
// 目前只实现livegbs的一对一的对讲, stream id就是通道的广播id
streamid := common.GenerateStreamID(common.InviteTypeBroadcast, deviceId, channelId, "", "")
device := &stack.Device{DeviceModel: model}
ctx, _ := context.WithTimeout(context.Background(), time.Second*10)
sinkModel, err := device.StartBroadcast(streamid, deviceId, channelId, ctx)
if err != nil {
w.WriteHeader(http.StatusBadRequest)
_ = common.HttpResponseJson(w, "广播失败")
return
}
err = common.WSForwardTo(r.URL.Path, w, r)
if err != nil {
log.Sugar.Errorf("广播失败 err: %s", err.Error())
}
log.Sugar.Infof("广播结束 device: %s/%s", deviceId, channelId)
// 对讲结束, 关闭sink
sink := &stack.Sink{SinkModel: sinkModel}
sink.Close(true, true)
}

6
livegbs_bean.go → api/livegbs_bean.go
View File

@@ -1,4 +1,4 @@
package main
package api
import (
"gb-cms/common"
@@ -214,8 +214,8 @@ type LiveGBSCascade struct {
DigestAlgorithm string
GM bool
Cert string
CreateAt string
UpdateAt string
CreatedAt string
UpdatedAt string
}
func ChannelModels2LiveGBSChannels(index int, channels []*dao.ChannelModel, deviceName string) []*LiveGBSChannel {

2
stats.go → api/stats.go
View File

@@ -1,4 +1,4 @@
package main
package api
// 每秒钟统计系统资源占用, 包括: cpu/流量/磁盘/内存
import (

8
temp_pwd.go → api/temp_pwd.go
View File

@@ -1,4 +1,4 @@
package main
package api
import (
"crypto/md5"
@@ -8,6 +8,12 @@ import (
"time"
)
var (
AdminMD5 string // 明文密码"admin"的MD5值
PwdMD5 string
StartUpTime time.Time
)
func GenerateTempPwd() string {
// 根据字母数字符号生成12位随机密码
// 字母数字符号

2
token_manager.go → api/token_manager.go
View File

@@ -1,4 +1,4 @@
package main
package api
import (
"math/rand"

11
common/sip.go
View File

@@ -98,3 +98,14 @@ func SetToTag(response sip.Message) {
to := toHeader[0].(*sip.ToHeader)
to.Params = sip.NewParams().Add("tag", sip.String{Str: util.RandString(10)})
}
func SetHeader(msg sip.Message, header sip.Header) {
msg.RemoveHeader(header.Name())
msg.AppendHeader(header)
}
func SetHeaderIfNotExist(msg sip.Message, header sip.Header) {
if len(msg.GetHeaders(header.Name())) < 1 {
msg.AppendHeader(header)
}
}

20
dao/channel.go
View File

@@ -9,9 +9,9 @@ import (
// GBModel 解决`Model`变量名与gorm.Model冲突
type GBModel struct {
ID uint `gorm:"primarykey"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"-"`
ID uint `gorm:"primarykey" xml:"-"`
CreatedAt time.Time `json:"created_at" xml:"-"`
UpdatedAt time.Time `json:"-" xml:"-"`
}
type ChannelModel struct {
@@ -49,11 +49,11 @@ type ChannelModel struct {
Status common.OnlineStatus `json:"status" xml:"Status,omitempty"`
Longitude string `json:"longitude" xml:"Longitude,omitempty"`
Latitude string `json:"latitude" xml:"Latitude,omitempty"`
Setup common.SetupType `json:"setup,omitempty"`
Setup common.SetupType `json:"setup,omitempty" xml:"-"`
ChannelNumber int `json:"channel_number" xml:"-"` // 对应1078的通道号
SubCount int `json:"-" xml:"-"` // 子节点数量
IsDir bool `json:"-" xml:"-"` // 是否是目录
CustomID *string `gorm:"unique"` // 自定义通道ID
CustomID *string `gorm:"unique" xml:"-"` // 自定义通道ID
Event string `json:"-" xml:"Event,omitempty" gorm:"-"` // <!-- 状态改变事件ON:上线,OFF:离线,VLOST:视频丢失,DEFECT:故障,ADD:增加,DEL:删除,UPDATE:更新(必选)-->
}
@@ -329,3 +329,13 @@ func (d *daoChannel) QueryChannelName(rootId string, channelId string) (string,
return channel.Name, nil
}
func (d *daoChannel) QueryCustomID(rootId string, channelId string) (string, error) {
var channel ChannelModel
tx := db.Select("custom_id").Where("root_id =? and device_id =?", rootId, channelId).Take(&channel)
if tx.Error != nil || channel.CustomID == nil {
return "", tx.Error
}
return *channel.CustomID, nil
}

18
dao/dialog.go
View File

@@ -15,7 +15,7 @@ const (
// SipDialogModel 持久化SIP会话
type SipDialogModel struct {
GBModel
DeviceID string
DeviceID string // 保存级联上级的会话, 使用server addr作为id
ChannelID string
CallID string
Dialog *common.RequestWrapper `json:"message,omitempty"`
@@ -85,3 +85,19 @@ func (m *daoDialog) QueryExpiredDialogs(now time.Time) ([]*SipDialogModel, error
}
return dialogs, nil
}
// QueryDialogByCallID 根据callid查询dialog
func (m *daoDialog) QueryDialogByCallID(id string) (*SipDialogModel, error) {
var dialog SipDialogModel
err := db.Where("call_id = ?", id).First(&dialog).Error
if err != nil {
return nil, err
}
return &dialog, nil
}
func (m *daoDialog) UpdateRefreshTime(callid string, refreshTime time.Time) error {
return DBTransaction(func(tx *gorm.DB) error {
return tx.Model(&SipDialogModel{}).Where("call_id = ?", callid).Update("refresh_time", refreshTime).Error
})
}

22
dao/platform.go
View File

@@ -290,3 +290,25 @@ func (d *daoPlatform) SetShareAllChannel(id int, shareAll bool) error {
return tx.Model(&PlatformModel{}).Where("id =?", id).Update("share_all", shareAll).Error
})
}
// QueryPlatformByChannelID 查询某个通道级联到的所有上级
func (d *daoPlatform) QueryPlatformByChannelID(deviceID, channelID string) ([]*PlatformChannelModel, error) {
var platformChannels []*PlatformChannelModel
tx := db.Where("device_id =? and channel_id =? group by server_addr", deviceID, channelID).Find(&platformChannels)
if tx.Error != nil {
return nil, tx.Error
}
return platformChannels, nil
}
// QueryAllSharedPlatforms 查询全部共享的级联列表
func (d *daoPlatform) QueryAllSharedPlatforms() ([]*PlatformModel, error) {
var platforms []*PlatformModel
tx := db.Where("share_all =?", true).Find(&platforms)
if tx.Error != nil {
return nil, tx.Error
}
return platforms, nil
}

5
go.mod
View File

@@ -19,7 +19,6 @@ require (
github.com/discoviking/fsm v0.0.0-20150126104936-f4a273feecca // indirect
github.com/dustin/go-humanize v1.0.1 // indirect
github.com/glebarez/go-sqlite v1.21.2 // indirect
github.com/go-co-op/gocron/v2 v2.16.6 // indirect
github.com/go-ole/go-ole v1.2.6 // indirect
github.com/gobwas/httphead v0.1.0 // indirect
github.com/gobwas/pool v0.2.1 // indirect
@@ -28,6 +27,7 @@ require (
github.com/jinzhu/inflection v1.0.0 // indirect
github.com/jinzhu/now v1.1.5 // indirect
github.com/jonboulle/clockwork v0.5.0 // indirect
github.com/kr/pretty v0.3.1 // indirect
github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect
github.com/mattn/go-colorable v0.1.13 // indirect
github.com/mattn/go-isatty v0.0.20 // indirect
@@ -54,6 +54,7 @@ require (
require (
github.com/glebarez/sqlite v1.11.0
github.com/go-co-op/gocron/v2 v2.16.6
github.com/gorilla/mux v1.8.1
github.com/gorilla/websocket v1.5.3
github.com/lkmio/avformat v0.0.1
@@ -62,4 +63,4 @@ require (
gorm.io/gorm v1.26.1
)
replace github.com/ghettovoice/gosip => github.com/lkmio/gosip v0.0.0-20250907014334-8cd203aeab7a
replace github.com/ghettovoice/gosip => github.com/lkmio/gosip v0.0.0-20251016021306-565c7a2fa4f5

30
main.go
View File

@@ -4,6 +4,7 @@ import (
"crypto/md5"
"encoding/hex"
"encoding/json"
"gb-cms/api"
"gb-cms/common"
"gb-cms/dao"
"gb-cms/hook"
@@ -20,14 +21,8 @@ import (
"time"
)
var (
AdminMD5 string // 明文密码"admin"的MD5值
PwdMD5 string
StartUpTime time.Time
)
func init() {
StartUpTime = time.Now()
api.StartUpTime = time.Now()
logConfig := common.LogConfig{
Level: int(zapcore.DebugLevel),
@@ -58,14 +53,14 @@ func main() {
// 读取或生成密码MD5值
hash := md5.Sum([]byte("admin"))
AdminMD5 = hex.EncodeToString(hash[:])
api.AdminMD5 = hex.EncodeToString(hash[:])
plaintext, md5Hex := ReadTempPwd()
plaintext, md5Hex := api.ReadTempPwd()
if plaintext != "" {
log.Sugar.Infof("temp pwd: %s", plaintext)
}
PwdMD5 = md5Hex
api.PwdMD5 = md5Hex
// 加载黑名单
blacklists, err := dao.Blacklist.Load()
@@ -82,7 +77,7 @@ func main() {
}
// 启动web session超时管理
go TokenManager.Start(5 * time.Minute)
go api.TokenManager.Start(5 * time.Minute)
// 启动设备在线超时管理
stack.OnlineDeviceManager.Start(time.Duration(common.Config.AliveExpires)*time.Second/4, time.Duration(common.Config.AliveExpires)*time.Second, stack.OnExpires)
@@ -103,7 +98,7 @@ func main() {
panic(err)
}
go StartStats()
go api.StartStats()
log.Sugar.Infof("启动sip server成功. addr: %s:%d", config.ListenIP, config.SipPort)
common.Config.SipContactAddr = net.JoinHostPort(config.PublicIP, strconv.Itoa(config.SipPort))
@@ -126,7 +121,7 @@ func main() {
// 启动http服务
httpAddr := net.JoinHostPort(config.ListenIP, strconv.Itoa(config.HttpPort))
log.Sugar.Infof("启动http server. addr: %s", httpAddr)
go startApiServer(httpAddr)
go api.StartApiServer(httpAddr)
// 启动目录刷新任务
go stack.AddScheduledTask(time.Minute, true, stack.RefreshCatalogScheduleTask)
@@ -137,6 +132,7 @@ func main() {
s, _ := gocron.NewScheduler()
defer func() { _ = s.Shutdown() }()
// 删除过期的位置、报警记录
_, _ = s.NewJob(
gocron.CronJob(
"0 3 * * *",
@@ -144,12 +140,12 @@ func main() {
),
gocron.NewTask(
func() {
// 删除过期的位置、报警记录
now := time.Now()
alarmExpireTime := now.Add(time.Duration(common.Config.AlarmReserveDays) * 24 * time.Hour)
positionExpireTime := now.Add(time.Duration(common.Config.PositionReserveDays) * 24 * time.Hour)
alarmExpireTime := now.AddDate(0, 0, -common.Config.AlarmReserveDays)
positionExpireTime := now.AddDate(0, 0, -common.Config.PositionReserveDays)
// 删除过期的报警记录
err := dao.Alarm.DeleteExpired(alarmExpireTime)
err = dao.Alarm.DeleteExpired(alarmExpireTime)
if err != nil {
log.Sugar.Errorf("删除过期的报警记录失败 err: %s", err.Error())
}

82
stack/client.go
View File

@@ -27,7 +27,24 @@ type GBClient interface {
// OnQueryDeviceInfo 被查询设备信息
OnQueryDeviceInfo(sn int)
OnSubscribeCatalog(sn int)
// OnSubscribeCatalog 被订阅目录
OnSubscribeCatalog(request sip.Request, expires int) (sip.Response, error)
// OnSubscribeAlarm 被订阅报警
OnSubscribeAlarm(request sip.Request, expires int) (sip.Response, error)
// OnSubscribePosition 被订阅位置
OnSubscribePosition(req sip.Request, expires int) (sip.Response, error)
CreateRequestByDialogType(t int, method sip.RequestMethod) (sip.Request, error)
SendMessage(body interface{})
BuildRequest(method sip.RequestMethod, contentType *sip.ContentType, body string) (sip.Request, error)
PushCatalog()
NotifyCatalog(sn int, channels []*dao.ChannelModel, messageFactory func() sip.Request)
}
type gbClient struct {
@@ -37,6 +54,17 @@ type gbClient struct {
}
func (g *gbClient) OnQueryCatalog(sn int, channels []*dao.ChannelModel) {
g.NotifyCatalog(sn, channels, func() sip.Request {
request, err := BuildMessageRequest(g.sipUA.Username, g.sipUA.ListenAddr, g.sipUA.ServerID, g.sipUA.ServerAddr, g.sipUA.Transport, "")
if err != nil {
panic(err)
}
return request
})
}
func (g *gbClient) NotifyCatalog(sn int, channels []*dao.ChannelModel, messageFactory func() sip.Request) {
response := CatalogResponse{}
response.SN = sn
response.CmdType = CmdCatalog
@@ -44,7 +72,6 @@ func (g *gbClient) OnQueryCatalog(sn int, channels []*dao.ChannelModel) {
response.SumNum = len(channels)
if response.SumNum < 1 {
g.SendMessage(&response)
return
}
@@ -52,26 +79,46 @@ func (g *gbClient) OnQueryCatalog(sn int, channels []*dao.ChannelModel) {
channel := *channels[i]
// 向上级推送自定义的通道ID
if channel.CustomID != nil {
if channel.CustomID != nil && *channel.CustomID != "" {
channel.DeviceID = *channel.CustomID
}
// 如果设备离线, 状态设置为OFF
_, b := OnlineDeviceManager.Find(channel.RootID)
if b {
channel.Status = common.ON
} else {
channel.Status = common.OFF
}
response.DeviceList.Devices = nil
response.DeviceList.Num = 1 // 一次发一个通道
response.DeviceList.Devices = append(response.DeviceList.Devices, &channel)
response.DeviceList.Devices[0].ParentID = g.sipUA.Username
g.SendMessage(&response)
request := messageFactory()
if request == nil {
continue
}
xmlBody, err := xml.MarshalIndent(&response, " ", "")
if err != nil {
panic(err)
}
request.SetBody(string(xmlBody), true)
common.SetHeader(request, &XmlMessageType)
g.stack.SendRequest(request)
}
}
func (g *gbClient) SendMessage(msg interface{}) {
marshal, err := xml.MarshalIndent(msg, "", " ")
xmlBody, err := xml.MarshalIndent(msg, " ", "")
if err != nil {
panic(err)
}
request, err := BuildMessageRequest(g.sipUA.Username, g.sipUA.ListenAddr, g.sipUA.ServerID, g.sipUA.ServerAddr, g.sipUA.Transport, string(marshal))
request, err := BuildMessageRequest(g.sipUA.Username, g.sipUA.ListenAddr, g.sipUA.ServerID, g.sipUA.ServerAddr, g.sipUA.Transport, string(xmlBody))
if err != nil {
panic(err)
}
@@ -79,6 +126,10 @@ func (g *gbClient) SendMessage(msg interface{}) {
g.sipUA.stack.SendRequest(request)
}
func (g *gbClient) BuildRequest(method sip.RequestMethod, contentType *sip.ContentType, body string) (sip.Request, error) {
return BuildRequest(method, g.sipUA.Username, g.sipUA.Username, g.sipUA.ServerID, g.sipUA.ServerAddr, g.sipUA.Transport, contentType, body)
}
func (g *gbClient) OnQueryDeviceInfo(sn int) {
g.deviceInfo.SN = sn
g.SendMessage(&g.deviceInfo)
@@ -95,8 +146,23 @@ func (g *gbClient) SetDeviceInfo(name, manufacturer, model, firmware string) {
g.deviceInfo.Firmware = firmware
}
func (g *gbClient) OnSubscribeCatalog(sn int) {
func (g *gbClient) OnSubscribeCatalog(request sip.Request, expires int) (sip.Response, error) {
return nil, nil
}
func (g *gbClient) OnSubscribeAlarm(request sip.Request, expires int) (sip.Response, error) {
return nil, nil
}
func (g *gbClient) OnSubscribePosition(req sip.Request, expires int) (sip.Response, error) {
return nil, nil
}
func (g *gbClient) CreateRequestByDialogType(t int, method sip.RequestMethod) (sip.Request, error) {
return nil, nil
}
func (g *gbClient) PushCatalog() {
}
func NewGBClient(params *common.SIPUAOptions, stack common.SipServer) GBClient {

29
stack/client_manager.go
View File

@@ -99,3 +99,32 @@ func RemovePlatform(key string) (GBClient, error) {
platform := PlatformManager.Remove(key)
return platform, nil
}
// FindChannelSharedPlatforms 查找改通道的共享级联列表
func FindChannelSharedPlatforms(deviceId, channelId string) map[string]GBClient {
var platforms = make(map[string]GBClient, 8)
sharedPlatforms, _ := dao.Platform.QueryAllSharedPlatforms()
for _, platform := range sharedPlatforms {
client := PlatformManager.Find(platform.ServerAddr)
if client == nil {
continue
}
platforms[platform.ServerAddr] = client
}
platformChannels, _ := dao.Platform.QueryPlatformByChannelID(deviceId, channelId)
for _, platformChannel := range platformChannels {
platform, _ := dao.Platform.QueryPlatformByID(int(platformChannel.PID))
if platform != nil {
client := PlatformManager.Find(platform.ServerAddr)
if client == nil {
continue
}
platforms[platform.ServerAddr] = client
}
}
return platforms
}

48
stack/device.go
View File

@@ -417,10 +417,58 @@ func (d *Device) Close() {
// 更新在数据库中的状态
d.Status = common.OFF
_ = dao.Device.UpdateDeviceStatus(d.DeviceID, common.OFF)
// 通知级联上级, 该设备下的通道离线
d.PushCatalog()
// 删除设备下的所有会话
_ = dao.Dialog.DeleteDialogs(d.DeviceID)
}
func (d *Device) PushCatalog() {
_, online := OnlineDeviceManager.Find(d.DeviceID)
channels, _ := dao.Channel.QueryChannelsByRootID(d.DeviceID)
if len(channels) < 1 {
return
}
catalog := CatalogResponse{
BaseResponse: BaseResponse{
BaseMessage: BaseMessage{
SN: GetSN(),
DeviceID: d.DeviceID,
CmdType: CmdCatalog,
},
},
}
var event = "OFF"
if online {
event = "ON"
}
for _, channel := range channels {
if online && channel.Status == common.OFF {
// 如果是上线通知, 只通知在线的通道
continue
}
var deviceId = channel.DeviceID
if channel.CustomID != nil && *channel.CustomID != "" {
deviceId = *channel.CustomID
}
catalog.DeviceList.Num = 1
catalog.DeviceList.Devices = append(catalog.DeviceList.Devices, &dao.ChannelModel{
DeviceID: deviceId,
Event: event,
})
}
catalog.SumNum = len(catalog.DeviceList.Devices)
ForwardCatalogNotifyMessage(&catalog)
}
// CreateDialogRequestFromAnswer 根据invite的应答创建Dialog请求
// 应答的to头域需携带tag
func CreateDialogRequestFromAnswer(message sip.Response, uas bool, remoteAddr string) sip.Request {

38
stack/event.go
View File

@@ -36,3 +36,41 @@ func (ev *Event) Equals(other interface{}) bool {
return false
}
type SubscriptionState string
func (ev *SubscriptionState) String() string { return fmt.Sprintf("%s: %s", ev.Name(), ev.Value()) }
func (ev *SubscriptionState) Name() string { return "Subscription-State" }
func (ev SubscriptionState) Value() string { return string(ev) }
func (ev *SubscriptionState) Clone() sip.Header { return ev }
func (ev *SubscriptionState) Equals(other interface{}) bool {
if h, ok := other.(SubscriptionState); ok {
if ev == nil {
return false
}
return *ev == h
}
if h, ok := other.(*SubscriptionState); ok {
if ev == h {
return true
}
if ev == nil && h != nil || ev != nil && h == nil {
return false
}
return *ev == *h
}
return false
}
//const (
// SubscriptionStateActive SubscriptionState = "active"
// SubscriptionStatePending SubscriptionState = "pending"
// SubscriptionStateTerminated SubscriptionState = "terminated"
//)

6
stack/jt_device.go
View File

@@ -62,10 +62,10 @@ func (g *JTDevice) OnInvite(request sip.Request, user string) sip.Response {
func (g *JTDevice) Start() {
log.Sugar.Infof("启动部标设备, deivce: %s transport: %s addr: %s", g.Username, g.sipUA.Transport, g.sipUA.ServerAddr)
g.sipUA.Start()
g.sipUA.SetOnRegisterHandler(g.Online, g.Offline)
g.sipUA.SetOnRegisterHandler(g.OnlineCB, g.OfflineCB)
}
func (g *JTDevice) Online() {
func (g *JTDevice) OnlineCB() {
log.Sugar.Infof("部标设备上线 sim number: %s device: %s server addr: %s", g.simNumber, g.Username, g.ServerAddr)
if err := dao.JTDevice.UpdateOnlineStatus(common.ON, g.Username); err != nil {
@@ -73,7 +73,7 @@ func (g *JTDevice) Online() {
}
}
func (g *JTDevice) Offline() {
func (g *JTDevice) OfflineCB() {
log.Sugar.Infof("部标设备离线 sim number: %s device: %s server addr: %s", g.simNumber, g.Username, g.ServerAddr)
if err := dao.JTDevice.UpdateOnlineStatus(common.OFF, g.Username); err != nil {

29
stack/message_factory.go
View File

@@ -9,7 +9,7 @@ import (
)
const (
XmlHeaderGBK = `<?xml version="1.0" encoding="GB2312"?>` + "\r\n"
XmlHeaderGBK = `<?xml version="1.0"?>` + "\r\n"
)
func BuildSDP(media, userName, sessionName, ip string, port uint16, startTime, stopTime, setup string, speed int, ssrc string, attrs ...string) string {
@@ -67,20 +67,25 @@ func NewSIPRequestBuilderWithTransport(transport string) *sip.RequestBuilder {
}
func BuildMessageRequest(from, fromRealm, to, toAddr, transport, body string) (sip.Request, error) {
gbkBody := AddXMLHeader(body)
return BuildRequest(sip.MESSAGE, from, fromRealm, to, toAddr, transport, &XmlMessageType, gbkBody)
}
func BuildRequest(method sip.RequestMethod, fromUser, realm, toUser, toAddr, transport string, contentType *sip.ContentType, body string) (sip.Request, error) {
builder := NewRequestBuilder(method, fromUser, realm, toUser, toAddr, transport)
if contentType != nil && len(body) > 0 {
builder.SetContentType(contentType)
builder.SetBody(body)
}
return builder.Build()
}
func AddXMLHeader(body string) string {
if !strings.HasPrefix(body, "<?xml") {
body = XmlHeaderGBK + body
}
//gbkBody, _, err := transform.String(simplifiedchinese.GBK.NewEncoder(), body)
//if err != nil {
// panic(err)
//}
gbkBody := body
builder := NewRequestBuilder(sip.MESSAGE, from, fromRealm, to, toAddr, transport)
builder.SetContentType(&XmlMessageType)
builder.SetBody(gbkBody)
return builder.Build()
return body
}
func NewRequestBuilder(method sip.RequestMethod, fromUser, realm, toUser, toAddr, transport string) *sip.RequestBuilder {

186
stack/platform.go
View File

@@ -7,9 +7,12 @@ import (
"gb-cms/log"
"github.com/ghettovoice/gosip/sip"
"github.com/lkmio/avformat/utils"
"net"
"net/http"
"net/netip"
"strconv"
"strings"
"time"
)
const (
@@ -19,6 +22,7 @@ const (
type Platform struct {
*gbClient
registerTimer *time.Timer // 在发起注册时, 启动定时器, 到期后, 如果未上线, 释放资源. 防止重启后, 上级也重启, 资源长时间未释放.
}
// OnBye 被上级挂断
@@ -27,14 +31,28 @@ func (g *Platform) OnBye(request sip.Request) {
g.CloseStream(id.Value(), false, true)
}
func (g *Platform) OnQueryCatalog(sn int, channels []*dao.ChannelModel) {
// 添加本级域
channels = append(channels, &dao.ChannelModel{
DeviceID: g.ServerID,
Setup: common.SetupTypePassive,
// 追加本地域
func (g *Platform) appendSelfDomain(channels []*dao.ChannelModel) []*dao.ChannelModel {
return append(channels, &dao.ChannelModel{
DeviceID: g.ServerID,
Setup: common.SetupTypePassive,
Name: DefaultDomainName,
Manufacturer: DefaultManufacturer,
Model: DefaultModel,
Owner: "Owner",
Address: "Address",
ParentID: "0",
Secrecy: "0",
RegisterWay: "1",
})
}
g.gbClient.OnQueryCatalog(sn, channels)
func (g *Platform) OnQueryCatalog(sn int, channels []*dao.ChannelModel) {
if len(channels) < 1 {
return
}
g.gbClient.OnQueryCatalog(sn, g.appendSelfDomain(channels))
}
// CloseStream 关闭级联会话
@@ -102,19 +120,29 @@ func (g *Platform) OnInvite(request sip.Request, user string) sip.Response {
func (g *Platform) Start() {
log.Sugar.Infof("启动级联设备, deivce: %s transport: %s addr: %s", g.Username, g.sipUA.Transport, g.sipUA.ServerAddr)
g.registerTimer = time.AfterFunc(120*time.Second, func() {
if !g.Online() {
g.release()
}
})
g.sipUA.Start()
g.sipUA.SetOnRegisterHandler(g.Online, g.Offline)
g.sipUA.SetOnRegisterHandler(g.OnlineCB, g.OfflineCB)
}
func (g *Platform) Stop() {
g.sipUA.Stop()
g.sipUA.SetOnRegisterHandler(nil, nil)
// 释放所有推流
g.CloseStreams(true, true)
g.release()
}
func (g *Platform) Online() {
func (g *Platform) OnlineCB() {
if g.registerTimer != nil {
g.registerTimer.Stop()
g.registerTimer = nil
}
log.Sugar.Infof("级联设备上线 device: %s server addr: %s", g.Username, g.ServerAddr)
if err := dao.Platform.UpdateOnlineStatus(common.ON, g.ServerAddr); err != nil {
@@ -122,15 +150,151 @@ func (g *Platform) Online() {
}
}
func (g *Platform) Offline() {
func (g *Platform) OfflineCB() {
log.Sugar.Infof("级联设备离线 device: %s server addr: %s", g.Username, g.ServerAddr)
if err := dao.Platform.UpdateOnlineStatus(common.OFF, g.ServerAddr); err != nil {
log.Sugar.Infof("更新级联设备状态失败 err: %s server addr: %s", err.Error(), g.ServerAddr)
}
g.release()
}
func (g *Platform) release() {
// 取消注册定时器
if g.registerTimer != nil {
g.registerTimer.Stop()
g.registerTimer = nil
}
// 释放所有推流
g.CloseStreams(true, true)
// 删除订阅会话
_ = dao.Dialog.DeleteDialogs(g.ServerAddr)
}
func (g *Platform) OnSubscribeCatalog(request sip.Request, expires int) (sip.Response, error) {
id := request.Source()
return CreateOrDeleteSubscribeDialog(id, request, expires, dao.SipDialogTypeSubscribeCatalog)
}
func (g *Platform) OnSubscribeAlarm(request sip.Request, expires int) (sip.Response, error) {
id := request.Source()
return CreateOrDeleteSubscribeDialog(id, request, expires, dao.SipDialogTypeSubscribeAlarm)
}
func (g *Platform) CreateRequestByDialogType(t int, method sip.RequestMethod) (sip.Request, error) {
model, err := dao.Dialog.QueryDialogsByType(g.ServerAddr, t)
if err != nil {
return nil, err
} else if len(model) < 1 || model[0].Dialog == nil {
return nil, fmt.Errorf("dialog type %d not found", t)
}
host, p, _ := net.SplitHostPort(g.ServerAddr)
remotePort, _ := strconv.Atoi(p)
request := CreateRequestFromDialog(model[0].Dialog.Request, method, host, remotePort)
// 添加头域
expiresSeconds := model[0].RefreshTime.Sub(time.Now()).Seconds()
subscriptionState := SubscriptionState(fmt.Sprintf("active;expires=%.0f;retry-after=0", expiresSeconds))
event := Event("presence")
if dao.SipDialogTypeSubscribeCatalog == t {
event = "catalog"
}
common.SetHeaderIfNotExist(request, &event)
common.SetHeader(request, &subscriptionState)
common.SetHeader(request, &XmlMessageType)
common.SetHeader(request, GlobalContactAddress.AsContactHeader())
return request, nil
}
func (g *Platform) PushCatalog() {
channels, err := dao.Platform.QueryPlatformChannels(g.ServerAddr)
if err != nil {
log.Sugar.Errorf("查询级联设备通道失败 err: %s server addr: %s", err.Error(), g.ServerAddr)
return
} else if len(channels) < 1 {
return
}
for _, channel := range channels {
channel.Event = "ADD"
}
// 因为没有dialog, 可能有的协议栈发送不过去
g.NotifyCatalog(GetSN(), g.appendSelfDomain(channels), func() sip.Request {
request, err := BuildRequest(sip.NOTIFY, g.sipUA.Username, g.sipUA.ListenAddr, g.sipUA.ServerID, g.sipUA.ServerAddr, g.sipUA.Transport, nil, "")
if err != nil {
panic(err)
}
event := Event("catalog")
subscriptionState := SubscriptionState("active")
common.SetHeader(request, &event)
common.SetHeader(request, &subscriptionState)
return request
})
}
func CreateOrDeleteSubscribeDialog(id string, request sip.Request, expires int, t int) (sip.Response, error) {
response := sip.NewResponseFromRequest("", request, 200, "OK", "")
common.SetHeader(response, GlobalContactAddress.AsContactHeader())
if expires < 1 {
// 取消订阅, 删除会话
_, _ = dao.Dialog.DeleteDialogsByType(id, t)
} else {
// 设置to tag
toHeader, _ := response.To()
var tag string
if toHeader.Params != nil {
get, b := toHeader.Params.Get("tag")
if b {
tag = get.String()
}
}
if "" == tag {
common.SetToTag(response)
}
// 首次或刷新订阅, 保存或更新会话
dialog := CreateDialogRequestFromAnswer(response, true, request.Source())
callid, _ := dialog.CallID()
refreshTime := time.Now().Add(time.Duration(expires) * time.Second)
// 已经存在会话, 更新刷新时间
oldDialog, err := dao.Dialog.QueryDialogByCallID(callid.Value())
if err == nil && oldDialog.ID > 0 {
oldDialog.RefreshTime = time.Now().Add(time.Duration(expires) * time.Second)
err = dao.Dialog.UpdateRefreshTime(callid.Value(), refreshTime)
return nil, err
}
// 创建新会话
// 删除之前旧的
_, _ = dao.Dialog.DeleteDialogsByType(id, t)
// 保存会话
err = dao.Dialog.Save(&dao.SipDialogModel{
DeviceID: id,
CallID: callid.Value(),
Dialog: &common.RequestWrapper{Request: dialog},
Type: t,
RefreshTime: refreshTime,
})
if err != nil {
return nil, err
}
}
return response, nil
}
func NewPlatform(options *common.SIPUAOptions, ua common.SipServer) (*Platform, error) {

6
stack/sink.go
View File

@@ -121,9 +121,9 @@ func AddForwardSink(forwardType int, request sip.Request, user string, sink *Sin
response := CreateResponseWithStatusCode(request, http.StatusOK)
// answer添加contact头域
response.RemoveHeader("Contact")
response.AppendHeader(GlobalContactAddress.AsContactHeader())
response.AppendHeader(&SDPMessageType)
common.SetHeader(response, GlobalContactAddress.AsContactHeader())
common.SetHeader(response, &SDPMessageType)
response.SetBody(answer, true)
common.SetToTag(response)

119
stack/sip_handler.go
View File

@@ -1,9 +1,11 @@
package stack
import (
"encoding/xml"
"gb-cms/common"
"gb-cms/dao"
"gb-cms/log"
"github.com/ghettovoice/gosip/sip"
"github.com/lkmio/avformat/utils"
"net"
"strconv"
@@ -45,7 +47,7 @@ func (e *EventHandler) OnRegister(id, transport, addr, userAgent string) (int, G
now := time.Now()
host, p, _ := net.SplitHostPort(addr)
port, _ := strconv.Atoi(p)
device := &dao.DeviceModel{
model := &dao.DeviceModel{
DeviceID: id,
Transport: transport,
RemoteIP: host,
@@ -56,17 +58,24 @@ func (e *EventHandler) OnRegister(id, transport, addr, userAgent string) (int, G
LastHeartbeat: now,
}
if err := dao.Device.SaveDevice(device); err != nil {
if err := dao.Device.SaveDevice(model); err != nil {
log.Sugar.Errorf("保存设备信息到数据库失败 device: %s err: %s", id, err.Error())
return 0, nil, false
} else if d, err := dao.Device.QueryDevice(id); err == nil {
// 查询所有字段
device = d
model = d
}
OnlineDeviceManager.Add(id, now)
count, _ := dao.Channel.QueryChanelCount(id, true)
return 3600, &Device{device}, count < 1 || dao.Device.QueryNeedRefreshCatalog(id, now)
// 级联通知通道上线
device := &Device{model}
if count > 0 {
go device.PushCatalog()
}
return 3600, device, count < 1 || dao.Device.QueryNeedRefreshCatalog(id, now)
}
func (e *EventHandler) OnKeepAlive(id string, addr string) bool {
@@ -129,6 +138,7 @@ func (e *EventHandler) SavePosition(position *dao.PositionModel) {
_ = dao.Device.UpdateDevice(position.DeviceID, conditions)
}
// 不保留位置信令
if common.Config.PositionReserveDays < 1 {
return
}
@@ -153,6 +163,63 @@ func (e *EventHandler) OnNotifyPositionMessage(notify *MobilePositionNotify) {
e.SavePosition(&model)
}
// ForwardCatalogNotifyMessage 转发目录变化到级联上级
func ForwardCatalogNotifyMessage(catalog *CatalogResponse) {
for _, channel := range catalog.DeviceList.Devices {
// 跳过没有级联的通道
platforms := FindChannelSharedPlatforms(catalog.DeviceID, channel.DeviceID)
if len(platforms) < 1 {
continue
}
for _, platform := range platforms {
// 跳过离线的级联设备
if !platform.Online() {
continue
}
// 先查出customid
model, _ := dao.Channel.QueryChannel(catalog.DeviceID, channel.DeviceID)
newCatalog := *catalog
newCatalog.SumNum = 1
newCatalog.DeviceID = platform.GetID()
newCatalog.DeviceList.Devices = []*dao.ChannelModel{channel}
newCatalog.DeviceList.Num = 1
// 优先使用自定义ID
if model != nil && model.CustomID != nil && *model.CustomID != channel.DeviceID {
newCatalog.DeviceList.Devices[0].DeviceID = *model.CustomID
}
// 格式化消息
indent, err := xml.MarshalIndent(&newCatalog, " ", "")
if err != nil {
log.Sugar.Errorf("报警消息格式化失败 err: %s", err.Error())
continue
}
// 已经订阅走订阅
request, err := platform.CreateRequestByDialogType(dao.SipDialogTypeSubscribeCatalog, sip.NOTIFY)
if err == nil {
body := AddXMLHeader(string(indent))
request.SetBody(body, true)
_ = platform.SendRequest(request)
continue
}
// 不基于会话创建一个notify消息
request, err = platform.BuildRequest(sip.NOTIFY, &XmlMessageType, string(indent))
if err != nil {
log.Sugar.Errorf("创建报警消息失败 err: %s", err.Error())
continue
}
_ = platform.SendRequest(request)
}
}
}
func (e *EventHandler) OnNotifyCatalogMessage(catalog *CatalogResponse) {
for _, channel := range catalog.DeviceList.Devices {
if channel.Event == "" {
@@ -185,6 +252,8 @@ func (e *EventHandler) OnNotifyCatalogMessage(catalog *CatalogResponse) {
log.Sugar.Warnf("未知的目录事件 %s 设备ID: %s", channel.Event, channel.DeviceID)
}
}
ForwardCatalogNotifyMessage(catalog)
}
func (e *EventHandler) OnNotifyAlarmMessage(deviceId string, alarm *AlarmNotify) {
@@ -225,4 +294,46 @@ func (e *EventHandler) OnNotifyAlarmMessage(deviceId string, alarm *AlarmNotify)
if err := dao.Alarm.Save(&model); err != nil {
log.Sugar.Errorf("保存报警信息到数据库失败 device: %s err: %s", alarm.DeviceID, err.Error())
}
channel, err := dao.Channel.QueryChannel(deviceId, alarm.DeviceID)
if channel == nil {
log.Sugar.Errorf("查询通道失败 err: %s device: %s channel: %s", err.Error(), deviceId, alarm.DeviceID)
return
}
// 转发报警到级联上级
platforms := FindChannelSharedPlatforms(deviceId, alarm.DeviceID)
if len(platforms) < 1 {
return
}
newAlarmMessage := *alarm
// 优先使用自定义ID
if channel.CustomID != nil && *channel.CustomID != alarm.DeviceID {
newAlarmMessage.DeviceID = *channel.CustomID
}
for _, platform := range platforms {
if !platform.Online() {
continue
}
// 已经订阅走订阅, 没有订阅走报警事件通知
request, err := platform.CreateRequestByDialogType(dao.SipDialogTypeSubscribeAlarm, sip.NOTIFY)
if err == nil {
// 格式化报警消息
indent, err := xml.MarshalIndent(&newAlarmMessage, " ", "")
if err != nil {
log.Sugar.Errorf("报警消息格式化失败 err: %s", err.Error())
continue
}
body := AddXMLHeader(string(indent))
request.SetBody(body, true)
_ = platform.SendRequest(request)
continue
}
platform.SendMessage(&newAlarmMessage)
}
}

59
stack/sip_server.go
View File

@@ -235,6 +235,62 @@ func (s *sipServer) OnNotify(wrapper *SipRequestSource) {
}
}
// OnSubscribe 收到上级订阅请求
func (s *sipServer) OnSubscribe(wrapper *SipRequestSource) {
var code = http.StatusBadRequest
var response sip.Response
defer func() {
if response == nil {
response = CreateResponseWithStatusCode(wrapper.req, code)
}
SendResponse(wrapper.tx, response)
}()
var client GBClient
if wrapper.fromJt {
return
} else {
if client = PlatformManager.Find(wrapper.req.Source()); client == nil {
log2.Sugar.Errorf("处理订阅请求失败, 找不到级联上级. request: %s", wrapper.req.String())
return
}
}
// 解析有效期头域
var expires int
var err error
expiresHeaders := wrapper.req.GetHeaders("Expires")
if len(expiresHeaders) < 1 {
log2.Sugar.Errorf("处理订阅请求失败, 找不到Expires头域. request: %s", wrapper.req.String())
return
} else if expires, err = strconv.Atoi(expiresHeaders[0].Value()); err != nil {
log2.Sugar.Errorf("处理订阅请求失败, 解析Expires头域失败. request: %s err: %s", wrapper.req.String(), err.Error())
return
}
// 处理订阅消息
body := wrapper.req.Body()
if strings.Contains(body, "<CmdType>Alarm</CmdType>") {
// 报警订阅
response, err = client.OnSubscribeAlarm(wrapper.req, expires)
} else if strings.Contains(body, "<CmdType>Catalog</CmdType>") {
// 目录订阅
response, err = client.OnSubscribeCatalog(wrapper.req, expires)
} else if strings.Contains(body, "<CmdType>MobilePosition</CmdType>") {
// 位置订阅
response, err = client.OnSubscribePosition(wrapper.req, expires)
}
if err != nil {
log2.Sugar.Errorf("处理订阅请求失败, 调用OnSubscribe失败. request: %s err: %s", wrapper.req.String(), err.Error())
code = http.StatusInternalServerError
} else if response == nil {
log2.Sugar.Errorf("处理订阅请求失败, 调用OnSubscribe返回空响应. request: %s", wrapper.req.String())
code = http.StatusInternalServerError
}
}
func (s *sipServer) OnMessage(wrapper *SipRequestSource) {
var ok bool
defer func() {
@@ -501,8 +557,7 @@ func StartSipServer(id, listenIP, publicIP string, listenPort int) (common.SipSe
})) == nil)
utils.Assert(ua.OnRequest(sip.CANCEL, filterRequest(func(wrapper *SipRequestSource) {
})) == nil)
utils.Assert(ua.OnRequest(sip.SUBSCRIBE, filterRequest(func(wrapper *SipRequestSource) {
})) == nil)
utils.Assert(ua.OnRequest(sip.SUBSCRIBE, filterRequest(server.OnSubscribe)) == nil)
server.listenAddr = addr
port := sip.Port(listenPort)

19
stack/sip_ua.go
View File

@@ -42,6 +42,10 @@ type SIPUA interface {
SetOnRegisterHandler(online, offline func())
GetDomain() string
Online() bool
SendRequest(request sip.Request) sip.ClientTransaction
}
func EqualSipUAOptions(old, new *common.SIPUAOptions) bool {
@@ -79,6 +83,8 @@ type sipUA struct {
onlineCB func()
offlineCB func()
online bool
}
func (g *sipUA) doRegister(request sip.Request) bool {
@@ -232,7 +238,8 @@ func (g *sipUA) Refresh() time.Duration {
if g.registerOK {
g.registerOKTime = time.Now()
if g.onlineCB != nil {
g.online = true
if g.onlineCB != nil && !expires {
go g.onlineCB()
}
}
@@ -256,6 +263,7 @@ func (g *sipUA) Refresh() time.Duration {
g.registerOK = false
g.registerOKRequest = nil
g.NatAddr = ""
g.online = false
if g.offlineCB != nil {
go g.offlineCB()
@@ -303,6 +311,7 @@ func (g *sipUA) Stop() {
g.doUnregister()
}
g.online = false
g.exited = true
g.cancel()
g.registerOK = false
@@ -318,3 +327,11 @@ func (g *sipUA) SetOnRegisterHandler(online, offline func()) {
func (g *sipUA) GetDomain() string {
return g.ServerAddr
}
func (g *sipUA) Online() bool {
return g.online
}
func (g *sipUA) SendRequest(request sip.Request) sip.ClientTransaction {
return g.stack.SendRequest(request)
}

26
stack/subscribe.go
View File

@@ -75,16 +75,11 @@ func Unsubscribe(deviceId string, t int, event Event, body []byte, remoteIP stri
// 添加事件头
expiresHeader := sip.Expires(0)
contactHeader := sip.ContactHeader{DisplayName: GlobalContactAddress.DisplayName, Address: GlobalContactAddress.Uri, Params: GlobalContactAddress.Params}
request.RemoveHeader("Event")
request.RemoveHeader("Expires")
request.RemoveHeader("Contact")
request.RemoveHeader("Content-Type")
request.AppendHeader(&event)
request.AppendHeader(&expiresHeader)
request.AppendHeader(&contactHeader)
request.AppendHeader(&XmlMessageType)
common.SetHeader(request, &event)
common.SetHeader(request, &expiresHeader)
common.SetHeader(request, GlobalContactAddress.AsContactHeader())
common.SetHeader(request, &XmlMessageType)
if body != nil {
request.SetBody(string(body), true)
@@ -103,16 +98,11 @@ func RefreshSubscribe(deviceId string, t int, event Event, expires int, body []b
request := CreateRequestFromDialog(dialogs[0].Dialog.Request, sip.SUBSCRIBE, remoteIP, remotePort)
expiresHeader := sip.Expires(expires)
contactHeader := sip.ContactHeader{DisplayName: GlobalContactAddress.DisplayName, Address: GlobalContactAddress.Uri, Params: GlobalContactAddress.Params}
request.RemoveHeader("Event")
request.RemoveHeader("Expires")
request.RemoveHeader("Contact")
request.RemoveHeader("Content-Type")
request.AppendHeader(&event)
request.AppendHeader(&expiresHeader)
request.AppendHeader(&contactHeader)
request.AppendHeader(&XmlMessageType)
common.SetHeader(request, &event)
common.SetHeader(request, &expiresHeader)
common.SetHeader(request, GlobalContactAddress.AsContactHeader())
common.SetHeader(request, &XmlMessageType)
if body != nil {
request.SetBody(string(body), true)