From 37fd121d11b461c2ded243f04898642aea20b3c9 Mon Sep 17 00:00:00 2001 From: langhuihui <178529795@qq.com> Date: Thu, 11 May 2023 21:12:41 +0800 Subject: [PATCH] feat: change to use ps plugin --- channel.go | 152 ++++++++++++++++++------------- device.go | 65 ++++++-------- handle.go | 10 ++- inviteoption.go | 14 +-- main.go | 46 ++++++---- publisher.go | 235 ------------------------------------------------ restful.go | 50 ++--------- server.go | 121 ++----------------------- 8 files changed, 179 insertions(+), 514 deletions(-) delete mode 100644 publisher.go diff --git a/channel.go b/channel.go index 63c843a..71cc52f 100644 --- a/channel.go +++ b/channel.go @@ -5,27 +5,55 @@ import ( "net/http" "strconv" "strings" - "sync" "time" + "sync/atomic" + "github.com/ghettovoice/gosip/sip" "go.uber.org/zap" . "m7s.live/engine/v4" "m7s.live/plugin/gb28181/v4/utils" + "m7s.live/plugin/ps/v4" ) +type PullStream struct { + opt *InviteOptions + channel *Channel + inviteRes sip.Response +} + +func (p *PullStream) Bye() int { + res := p.inviteRes + bye := p.channel.CreateRequst(sip.BYE) + from, _ := res.From() + to, _ := res.To() + callId, _ := res.CallID() + bye.ReplaceHeaders(from.Name(), []sip.Header{from}) + bye.ReplaceHeaders(to.Name(), []sip.Header{to}) + bye.ReplaceHeaders(callId.Name(), []sip.Header{callId}) + resp, err := p.channel.device.SipRequestForResponse(bye) + if p.opt.IsLive() { + p.channel.status.Store(0) + // defer p.channel.TryAutoInvite(p.opt) + } + if p.opt.recyclePort != nil { + p.opt.recyclePort(p.opt.MediaPort) + } + if err != nil { + return ServerInternalError + } + return int(resp.StatusCode()) +} + type ChannelEx struct { - device *Device - RecordPublisher *GBPublisher `json:"-" yaml:"-"` - LivePublisher *GBPublisher - LiveSubSP string //实时子码流 + device *Device // 所属设备 + status atomic.Int32 // 通道状态,0:空闲,1:正在invite,2:正在播放 + LiveSubSP string // 实时子码流，通过rtsp Records []*Record RecordStartTime string RecordEndTime string recordStartTime time.Time recordEndTime time.Time - liveInviteLock *sync.Mutex - tcpPortIndex uint16 GpsTime time.Time //gps时间 Longitude string //经度 Latitude string //纬度 @@ -203,18 +231,26 @@ f = v/a/编码格式/码率大小/采样率 f字段中视、音频参数段之间不需空格分割。 可使用f字段中的分辨率参数标识同一设备不同分辨率的码流。 */ + func (channel *Channel) Invite(opt *InviteOptions) (code int, err error) { if opt.IsLive() { - if !channel.liveInviteLock.TryLock() { + if !channel.status.CompareAndSwap(0, 1) { return 304, nil } defer func() { - if code != OK { - channel.liveInviteLock.Unlock() + if err != nil { + channel.status.Store(0) + if conf.InviteMode == 1 { + // 5秒后重试 + time.AfterFunc(time.Second*5, func() { + channel.Invite(opt) + }) + } + } else { + channel.status.Store(2) } }() } - channel.Bye(opt.IsLive()) d := channel.device streamPath := fmt.Sprintf("%s/%s", d.ID, channel.DeviceID) s := "Play" @@ -223,37 +259,36 @@ func (channel *Channel) Invite(opt *InviteOptions) (code int, err error) { s = "Playback" streamPath = fmt.Sprintf("%s/%s/%d-%d", d.ID, channel.DeviceID, opt.Start, opt.End) } + if opt.StreamPath != "" { + streamPath = opt.StreamPath + } if opt.dump == "" { opt.dump = conf.DumpPath } - publisher := &GBPublisher{ - InviteOptions: opt, - channel: channel, - } - publisher.DisableReorder = !conf.RtpReorder protocol := "" + networkType := "udp" + resuePort := true if conf.IsMediaNetworkTCP() { + networkType = "tcp" protocol = "TCP/" if conf.tcpPorts.Valid { - opt.MediaPort, err = publisher.ListenTCP() - if err != nil { - return ServerInternalError, err - } - } else if opt.MediaPort == 0 { - opt.MediaPort = conf.MediaPort + opt.MediaPort, err = conf.tcpPorts.GetPort() + opt.recyclePort = conf.tcpPorts.Recycle + resuePort = false } - publisher.DisableReorder = true } else { if conf.udpPorts.Valid { - opt.MediaPort, err = publisher.ListenUDP() - if err != nil { - code = ServerInternalError - return - } - } else if opt.MediaPort == 0 { - opt.MediaPort = conf.MediaPort + opt.MediaPort, err = conf.udpPorts.GetPort() + opt.recyclePort = conf.udpPorts.Recycle + resuePort = false } } + if err != nil { + return http.StatusInternalServerError, err + } + if opt.MediaPort == 0 { + opt.MediaPort = conf.MediaPort + } sdpInfo := []string{ "v=0", @@ -266,7 +301,6 @@ func (channel *Channel) Invite(opt *InviteOptions) (code int, err error) { "a=recvonly", "a=rtpmap:96 PS/90000", "y=" + opt.ssrc, - "", } if conf.IsMediaNetworkTCP() { sdpInfo = append(sdpInfo, "a=setup:passive", "a=connection:new") @@ -275,22 +309,22 @@ func (channel *Channel) Invite(opt *InviteOptions) (code int, err error) { contentType := sip.ContentType("application/sdp") invite.AppendHeader(&contentType) - invite.SetBody(strings.Join(sdpInfo, "\r\n"), true) + invite.SetBody(strings.Join(sdpInfo, "\r\n")+"\r\n", true) subject := sip.GenericHeader{ HeaderName: "Subject", Contents: fmt.Sprintf("%s:%s,%s:0", channel.DeviceID, opt.ssrc, conf.Serial), } invite.AppendHeader(&subject) - publisher.inviteRes, err = d.SipRequestForResponse(invite) + inviteRes, err := d.SipRequestForResponse(invite) if err != nil { plugin.Error(fmt.Sprintf("SIP->Invite %s :%s invite error: %s", channel.DeviceID, invite.String(), err.Error())) return http.StatusInternalServerError, err } - code = int(publisher.inviteRes.StatusCode()) + code = int(inviteRes.StatusCode()) plugin.Info(fmt.Sprintf("Channel :%s invite response status code: %d", channel.DeviceID, code)) if code == OK { - ds := strings.Split(publisher.inviteRes.Body(), "\r\n") + ds := strings.Split(inviteRes.Body(), "\r\n") for _, l := range ds { if ls := strings.Split(l, "="); len(ls) > 1 { if ls[0] == "y" && len(ls[1]) > 0 { @@ -303,46 +337,42 @@ func (channel *Channel) Invite(opt *InviteOptions) (code int, err error) { } } } - // if conf.UdpCacheSize > 0 && !conf.IsMediaNetworkTCP() { - // publisher.udpCache = utils.NewPqRtp() - // } - if err = plugin.Publish(streamPath, publisher); err != nil { - code = ServerInternalError - return + err = ps.Receive(streamPath, opt.dump, fmt.Sprintf("%s:%d", networkType, opt.MediaPort), opt.SSRC, resuePort) + if err == nil { + PullStreams.Store(streamPath, &PullStream{ + opt: opt, + channel: channel, + inviteRes: inviteRes, + }) + err = srv.Send(sip.NewAckRequest("", invite, inviteRes, "", nil)) } - ack := sip.NewAckRequest("", invite, publisher.inviteRes, "", nil) - srv.Send(ack) - } else if channel.CanInvite() { - time.AfterFunc(time.Second*5, func() { - channel.TryAutoInvite() - }) } return } -func (channel *Channel) Bye(live bool) int { +func (channel *Channel) Bye(streamPath string) int { d := channel.device - streamPath := fmt.Sprintf("%s/%s", d.ID, channel.DeviceID) - if s := Streams.Get(streamPath); s != nil { - s.Close() + if streamPath == "" { + streamPath = fmt.Sprintf("%s/%s", d.ID, channel.DeviceID) } - if live && channel.LivePublisher != nil { - return channel.LivePublisher.Bye() + if s, loaded := PullStreams.LoadAndDelete(streamPath); loaded { + s.(*PullStream).Bye() + if s := Streams.Get(streamPath); s != nil { + s.Close() + } + return http.StatusOK } - if !live && channel.RecordPublisher != nil { - return channel.RecordPublisher.Bye() - } - return 404 + return http.StatusNotFound } -func (channel *Channel) TryAutoInvite() { - if conf.AutoInvite && channel.CanInvite() { - go channel.Invite(&InviteOptions{}) +func (channel *Channel) TryAutoInvite(opt *InviteOptions) { + if conf.InviteMode == 1 && channel.CanInvite() { + go channel.Invite(opt) } } func (channel *Channel) CanInvite() bool { - if channel.LivePublisher != nil || len(channel.DeviceID) != 20 || channel.Status == "OFF" { + if channel.status.Load() != 0 || len(channel.DeviceID) != 20 || channel.Status == "OFF" { return false } diff --git a/device.go b/device.go index f7df038..ec93ded 100644 --- a/device.go +++ b/device.go @@ -10,8 +10,6 @@ import ( "sync" "time" - "golang.org/x/exp/maps" - "go.uber.org/zap" "m7s.live/engine/v4" "m7s.live/plugin/gb28181/v4/utils" @@ -62,8 +60,7 @@ type Device struct { sipIP string //设备对应网卡的服务器ip mediaIP string //设备对应网卡的服务器ip NetAddr string - ChannelMap map[string]*Channel - channelMutex sync.RWMutex + channelMap sync.Map subscriber struct { CallID string Timeout time.Time @@ -76,13 +73,18 @@ type Device struct { func (d *Device) MarshalJSON() ([]byte, error) { type Alias Device - return json.Marshal(&struct { + data := &struct { Channels []*Channel *Alias }{ - Channels: maps.Values(d.ChannelMap), - Alias: (*Alias)(d), + Alias: (*Alias)(d), + } + d.channelMap.Range(func(key, value interface{}) bool { + c := value.(*Channel) + data.Channels = append(data.Channels, c) + return true }) + return json.Marshal(data) } func (c *GB28181Config) RecoverDevice(d *Device, req sip.Request) { from, _ := req.From() @@ -116,9 +118,6 @@ func (c *GB28181Config) RecoverDevice(d *Device, req sip.Request) { d.mediaIP = mediaIp d.NetAddr = deviceIp d.UpdateTime = time.Now() - if d.ChannelMap == nil { - d.ChannelMap = make(map[string]*Channel) - } } func (c *GB28181Config) StoreDevice(id string, req sip.Request) (d *Device) { @@ -164,7 +163,6 @@ func (c *GB28181Config) StoreDevice(id string, req sip.Request) (d *Device) { sipIP: sipIP, mediaIP: mediaIp, NetAddr: deviceIp, - ChannelMap: make(map[string]*Channel), } Devices.Store(id, d) c.SaveDevices() @@ -200,35 +198,27 @@ func (c *GB28181Config) SaveDevices() { } func (d *Device) addOrUpdateChannel(channel *Channel) { - d.channelMutex.Lock() - defer d.channelMutex.Unlock() + if old, ok := d.channelMap.Load(channel.DeviceID); ok { + channel.ChannelEx = old.(*Channel).ChannelEx + } channel.device = d - if old, ok := d.ChannelMap[channel.DeviceID]; ok { - //复制锁指针 - channel.ChannelEx = old.ChannelEx - } - if channel.liveInviteLock == nil { - channel.liveInviteLock = &sync.Mutex{} - } - d.ChannelMap[channel.DeviceID] = channel + d.channelMap.Store(channel.DeviceID, channel) } func (d *Device) deleteChannel(DeviceID string) { - d.channelMutex.Lock() - defer d.channelMutex.Unlock() - delete(d.ChannelMap, DeviceID) + d.channelMap.Delete(DeviceID) } func (d *Device) CheckSubStream() { - d.channelMutex.Lock() - defer d.channelMutex.Unlock() - for _, c := range d.ChannelMap { + d.channelMap.Range(func(key, value any) bool { + c := value.(*Channel) if s := engine.Streams.Get("sub/" + c.DeviceID); s != nil { c.LiveSubSP = s.Path } else { c.LiveSubSP = "" } - } + return true + }) } func (d *Device) UpdateChannels(list []*Channel) { @@ -265,7 +255,7 @@ func (d *Device) UpdateChannels(list []*Channel) { go c.QueryRecord(n.Format(TIME_LAYOUT), n.Add(time.Hour*24-time.Second).Format(TIME_LAYOUT)) } } - c.TryAutoInvite() + c.TryAutoInvite(&InviteOptions{}) if s := engine.Streams.Get("sub/" + c.DeviceID); s != nil { c.LiveSubSP = s.Path } else { @@ -274,11 +264,11 @@ func (d *Device) UpdateChannels(list []*Channel) { } } func (d *Device) UpdateRecord(channelId string, list []*Record) { - d.channelMutex.RLock() - if c, ok := d.ChannelMap[channelId]; ok { + d.channelMap.Range(func(key, value any) bool { + c := value.(*Channel) c.Records = append(c.Records, list...) - } - d.channelMutex.RUnlock() + return true + }) } func (d *Device) CreateRequest(Method sip.RequestMethod) (req sip.Request) { @@ -455,7 +445,8 @@ func (d *Device) MobilePositionSubscribe(id string, expires time.Duration, inter // UpdateChannelPosition 更新通道GPS坐标 func (d *Device) UpdateChannelPosition(channelId string, gpsTime string, lng string, lat string) { - if c, ok := d.ChannelMap[channelId]; ok { + if v, ok := d.channelMap.Load(channelId); ok { + c := v.(*Channel) c.ChannelEx.GpsTime = time.Now() //时间取系统收到的时间，避免设备时间和格式问题 c.ChannelEx.Longitude = lng c.ChannelEx.Latitude = lat @@ -534,7 +525,8 @@ func (d *Device) UpdateChannelStatus(deviceList []*notifyMessage) { } func (d *Device) channelOnline(DeviceID string) { - if c, ok := d.ChannelMap[DeviceID]; ok { + if v, ok := d.channelMap.Load(DeviceID); ok { + c := v.(*Channel) c.Status = "ON" plugin.Sugar().Debugf("通道[%s]在线\n", c.Name) } else { @@ -543,7 +535,8 @@ func (d *Device) channelOnline(DeviceID string) { } func (d *Device) channelOffline(DeviceID string) { - if c, ok := d.ChannelMap[DeviceID]; ok { + if v, ok := d.channelMap.Load(DeviceID); ok { + c := v.(*Channel) c.Status = "OFF" plugin.Sugar().Debugf("通道[%s]离线\n", c.Name) } else { diff --git a/handle.go b/handle.go index c7275d3..37fa78d 100644 --- a/handle.go +++ b/handle.go @@ -185,12 +185,14 @@ func (c *GB28181Config) OnMessage(req sip.Request, tx sip.ServerTransaction) { case "Keepalive": d.LastKeepaliveAt = time.Now() //callID !="" 说明是订阅的事件类型信息 - if d.ChannelMap == nil || len(d.ChannelMap) == 0 { + if d.lastSyncTime.IsZero() { go d.syncChannels() } else { - for _, ch := range d.ChannelMap { - ch.TryAutoInvite() - } + d.channelMap.Range(func(key, value interface{}) bool { + channel := value.(*Channel) + channel.TryAutoInvite(&InviteOptions{}) + return true + }) } //为什么要查找子码流? //d.CheckSubStream() diff --git a/inviteoption.go b/inviteoption.go index db4fc2c..369eaf6 100644 --- a/inviteoption.go +++ b/inviteoption.go @@ -8,12 +8,14 @@ import ( ) type InviteOptions struct { - Start int - End int - dump string - ssrc string - SSRC uint32 - MediaPort uint16 + Start int + End int + dump string + ssrc string + SSRC uint32 + MediaPort uint16 + StreamPath string + recyclePort func(p uint16) (err error) } func (o InviteOptions) IsLive() bool { diff --git a/main.go b/main.go index ce7d2e8..8572ce3 100644 --- a/main.go +++ b/main.go @@ -4,11 +4,11 @@ import ( "fmt" "os" "strings" + "sync" "time" myip "github.com/husanpao/ip" . "m7s.live/engine/v4" - "m7s.live/engine/v4/config" "m7s.live/engine/v4/util" ) @@ -19,18 +19,19 @@ type GB28181PositionConfig struct { } type GB28181Config struct { - AutoInvite bool `default:"true"` + // AutoInvite bool `default:"true"` + InviteMode int `default:"1"` //邀请模式，0:手动拉流，1:预拉流，2:按需拉流 PreFetchRecord bool InviteIDs string //按照国标gb28181协议允许邀请的设备类型:132 摄像机 NVR ListenAddr string `default:"0.0.0.0"` //sip服务器的配置 - SipNetwork string `default:"udp"` //传输协议，默认UDP，可选TCP - SipIP string //sip 服务器公网IP - SipPort uint16 `default:"5060"` //sip 服务器端口，默认 5060 - Serial string `default:"34020000002000000001"` //sip 服务器 id, 默认 34020000002000000001 - Realm string `default:"3402000000"` //sip 服务器域，默认 3402000000 - Username string //sip 服务器账号 - Password string //sip 服务器密码 + SipNetwork string `default:"udp"` //传输协议，默认UDP，可选TCP + SipIP string //sip 服务器公网IP + SipPort uint16 `default:"5060"` //sip 服务器端口，默认 5060 + Serial string `default:"34020000002000000001"` //sip 服务器 id, 默认 34020000002000000001 + Realm string `default:"3402000000"` //sip 服务器域，默认 3402000000 + Username string //sip 服务器账号 + Password string //sip 服务器密码 Port struct { // 新配置方式 Sip string `default:"udp:5060"` Media string `default:"tcp:58200"` @@ -52,14 +53,15 @@ type GB28181Config struct { // WaitKeyFrame bool //是否等待关键帧，如果等待，则在收到第一个关键帧之前，忽略所有媒体流 RemoveBanInterval time.Duration `default:"600s"` //移除禁止设备间隔 // UdpCacheSize int //udp缓存大小 - LogLevel string `default:"info"` //trace, debug, info, warn, error, fatal, panic - routes map[string]string - DumpPath string //dump PS流本地文件路径 - RtpReorder bool `default:"true"` - config.Publish - Server + LogLevel string `default:"info"` //trace, debug, info, warn, error, fatal, panic + routes map[string]string + DumpPath string //dump PS流本地文件路径 + Ignores map[string]struct{} + tcpPorts PortManager + udpPorts PortManager Position GB28181PositionConfig //关于定位的配置参数 + } func (c *GB28181Config) initRoutes() { @@ -73,8 +75,9 @@ func (c *GB28181Config) initRoutes() { } plugin.Info(fmt.Sprintf("LocalAndInternalIPs detail: %s", c.routes)) } + func (c *GB28181Config) OnEvent(event any) { - switch event.(type) { + switch e := event.(type) { case FirstConfig: if c.Port.Sip != "udp:5060" { protocol, ports := util.Conf2Listener(c.Port.Sip) @@ -95,6 +98,16 @@ func (c *GB28181Config) OnEvent(event any) { c.ReadDevices() go c.initRoutes() c.startServer() + case *Stream: + if c.InviteMode == 2 { + if channel := FindChannel(e.AppName, e.StreamName); channel != nil { + channel.TryAutoInvite(&InviteOptions{}) + } + } + case SEclose: + if v, ok := PullStreams.LoadAndDelete(e.Target.Path); ok { + go v.(*PullStream).Bye() + } } } @@ -105,3 +118,4 @@ func (c *GB28181Config) IsMediaNetworkTCP() bool { var conf GB28181Config var plugin = InstallPlugin(&conf) +var PullStreams sync.Map //拉流 diff --git a/publisher.go b/publisher.go deleted file mode 100644 index b9eae8b..0000000 --- a/publisher.go +++ /dev/null @@ -1,235 +0,0 @@ -package gb28181 - -import ( - "encoding/binary" - "fmt" - "io" - "net" - "os" - "path/filepath" - "time" - - "github.com/ghettovoice/gosip/sip" - "github.com/pion/rtp" - "go.uber.org/zap" - . "m7s.live/engine/v4" - "m7s.live/engine/v4/util" -) - -type GBPublisher struct { - PSPublisher - *InviteOptions - channel *Channel - inviteRes sip.Response - // udpCache *utils.PriorityQueueRtp - dumpFile *os.File - dumpPrint io.Writer - lastReceive time.Time -} - -func (p *GBPublisher) PrintDump(s string) { - if p.dumpPrint != nil { - p.dumpPrint.Write([]byte(s)) - } -} - -func (p *GBPublisher) OnEvent(event any) { - if p.channel == nil { - // p.parser.EsHandler = p - p.IO.OnEvent(event) - return - } - switch event.(type) { - case IPublisher: - if p.IsLive() { - p.Type = "GB28181 Live" - p.channel.LivePublisher = p - } else { - p.Type = "GB28181 Playback" - p.channel.RecordPublisher = p - } - // p.parser.EsHandler = p - conf.publishers.Add(p.SSRC, p) - if err := error(nil); p.dump != "" { - fp := filepath.Join(p.dump, p.Stream.Path) - os.MkdirAll(filepath.Dir(fp), 0766) - if p.dumpFile, err = os.OpenFile(fp, os.O_WRONLY|os.O_CREATE, 0644); err != nil { - p.Error("open dump file failed", zap.Error(err)) - } - } - case SEwaitPublish: - //掉线自动重新拉流 - if p.IsLive() { - if p.channel.LivePublisher != nil { - p.channel.LivePublisher = nil - p.channel.liveInviteLock.Unlock() - } - go p.channel.Invite(&InviteOptions{}) - } - case SEclose, SEKick: - if p.IsLive() { - if p.channel.LivePublisher != nil { - p.channel.LivePublisher = nil - p.channel.liveInviteLock.Unlock() - } - } else { - p.channel.RecordPublisher = nil - } - conf.publishers.Delete(p.SSRC) - if p.dumpFile != nil { - p.dumpFile.Close() - } - go p.Bye() - } - p.Publisher.OnEvent(event) -} - -func (p *GBPublisher) Bye() int { - res := p.inviteRes - if res == nil { - return 404 - } - defer p.Stop() - p.inviteRes = nil - bye := p.channel.CreateRequst(sip.BYE) - from, _ := res.From() - to, _ := res.To() - callId, _ := res.CallID() - bye.ReplaceHeaders(from.Name(), []sip.Header{from}) - bye.ReplaceHeaders(to.Name(), []sip.Header{to}) - bye.ReplaceHeaders(callId.Name(), []sip.Header{callId}) - resp, err := p.channel.device.SipRequestForResponse(bye) - if err != nil { - p.Error("Bye", zap.Error(err)) - return ServerInternalError - } - return int(resp.StatusCode()) -} - -func (p *GBPublisher) Replay(f *os.File) (err error) { - var rtpPacket rtp.Packet - defer f.Close() - if p.dumpPrint != nil { - p.PrintDump(` - `) - p.PrintDump("") - defer p.PrintDump("
") - } - var t uint16 - for l := make([]byte, 6); !p.IsClosed(); time.Sleep(time.Millisecond * time.Duration(t)) { - _, err = f.Read(l) - if err != nil { - return - } - payload := make([]byte, util.ReadBE[int](l[:4])) - t = util.ReadBE[uint16](l[4:]) - p.PrintDump(fmt.Sprintf("[%d %d]", t, len(payload))) - _, err = f.Read(payload) - if err != nil { - return - } - rtpPacket.Unmarshal(payload) - p.PushPS(&rtpPacket) - } - return -} - -func (p *GBPublisher) ListenUDP() (port uint16, err error) { - var rtpPacket rtp.Packet - networkBuffer := 1048576 - port, err = conf.udpPorts.GetPort() - if err != nil { - return - } - addr := fmt.Sprintf(":%d", port) - mediaAddr, _ := net.ResolveUDPAddr("udp", addr) - conn, err := net.ListenUDP("udp", mediaAddr) - if err != nil { - conf.udpPorts.Recycle(port) - plugin.Error("listen media server udp err", zap.String("addr", addr), zap.Error(err)) - return 0, err - } - p.SetIO(conn) - go func() { - defer conn.Close() - bufUDP := make([]byte, networkBuffer) - plugin.Info("Media udp server start.", zap.Uint16("port", port)) - defer plugin.Info("Media udp server stop", zap.Uint16("port", port)) - defer conf.udpPorts.Recycle(port) - dumpLen := make([]byte, 6) - conn.SetReadDeadline(time.Now().Add(time.Second * 10)) - for n, _, err := conn.ReadFromUDP(bufUDP); err == nil; n, _, err = conn.ReadFromUDP(bufUDP) { - ps := bufUDP[:n] - if err := rtpPacket.Unmarshal(ps); err != nil { - plugin.Error("Decode rtp error:", zap.Error(err)) - } - p.writeDump(ps, dumpLen) - p.PushPS(&rtpPacket) - conn.SetReadDeadline(time.Now().Add(time.Second * 10)) - } - }() - return -} -func (p *GBPublisher) writeDump(ps util.Buffer, dumpLen []byte) { - if p.dumpFile != nil { - util.PutBE(dumpLen[:4], ps.Len()) - if p.lastReceive.IsZero() { - util.PutBE(dumpLen[4:], 0) - } else { - util.PutBE(dumpLen[4:], uint16(time.Since(p.lastReceive).Milliseconds())) - } - p.lastReceive = time.Now() - p.dumpFile.Write(dumpLen) - p.dumpFile.Write(ps) - } -} -func (p *GBPublisher) ListenTCP() (port uint16, err error) { - port, err = conf.tcpPorts.GetPort() - if err != nil { - return - } - addr := fmt.Sprintf(":%d", port) - mediaAddr, _ := net.ResolveTCPAddr("tcp", addr) - listen, err := net.ListenTCP("tcp", mediaAddr) - if err != nil { - defer conf.tcpPorts.Recycle(port) - plugin.Error("listen media server tcp err", zap.String("addr", addr), zap.Error(err)) - return 0, err - } - go func() { - plugin.Info("Media tcp server start.", zap.Uint16("port", port)) - defer conf.tcpPorts.Recycle(port) - defer plugin.Info("Media tcp server stop", zap.Uint16("port", port)) - conn, err := listen.Accept() - listen.Close() - p.SetIO(conn) - if err != nil { - plugin.Error("Accept err=", zap.Error(err)) - return - } - var rtpPacket rtp.Packet - ps := make(util.Buffer, 1024) - dumpLen := make([]byte, 6) - defer conn.Close() - for err == nil { - if _, err = io.ReadFull(conn, dumpLen[:2]); err != nil { - return - } - ps.Relloc(int(binary.BigEndian.Uint16(dumpLen[:2]))) - if _, err = io.ReadFull(conn, ps); err != nil { - return - } - if err := rtpPacket.Unmarshal(ps); err != nil { - plugin.Error("gb28181 decode rtp error:", zap.Error(err)) - } else if !p.IsClosed() { - p.writeDump(ps, dumpLen) - p.PushPS(&rtpPacket) - } - } - }() - return -} diff --git a/restful.go b/restful.go index 0df1ec6..4583b0a 100644 --- a/restful.go +++ b/restful.go @@ -2,9 +2,7 @@ package gb28181 import ( "net/http" - "os" "strconv" - "strings" "time" "m7s.live/engine/v4/util" @@ -52,15 +50,17 @@ func (c *GB28181Config) API_invite(w http.ResponseWriter, r *http.Request) { query := r.URL.Query() id := query.Get("id") channel := query.Get("channel") + streamPath := query.Get("streamPath") port, _ := strconv.Atoi(query.Get("mediaPort")) opt := InviteOptions{ - dump: query.Get("dump"), - MediaPort: uint16(port), + dump: query.Get("dump"), + MediaPort: uint16(port), + StreamPath: streamPath, } opt.Validate(query.Get("startTime"), query.Get("endTime")) if c := FindChannel(id, channel); c == nil { http.NotFound(w, r) - } else if opt.IsLive() && c.LivePublisher != nil { + } else if opt.IsLive() && c.status.Load() > 0 { w.WriteHeader(304) //直播流已存在 } else if code, err := c.Invite(&opt); err == nil { w.WriteHeader(code) @@ -69,48 +69,12 @@ func (c *GB28181Config) API_invite(w http.ResponseWriter, r *http.Request) { } } -func (c *GB28181Config) API_replay(w http.ResponseWriter, r *http.Request) { - dump := r.URL.Query().Get("dump") - printOut := r.URL.Query().Get("print") - streamPath := r.URL.Query().Get("streamPath") - if dump == "" { - dump = c.DumpPath - } - f, err := os.OpenFile(dump, os.O_RDONLY, 0644) - if err != nil { - http.Error(w, err.Error(), http.StatusInternalServerError) - } else { - if streamPath == "" { - if strings.HasPrefix(dump, "/") { - streamPath = "replay" + dump - } else { - streamPath = "replay/" + dump - } - } - var pub GBPublisher - pub.SetIO(f) - if err = plugin.Publish(streamPath, &pub); err == nil { - if printOut != "" { - pub.dumpPrint = w - pub.SetParentCtx(r.Context()) - err = pub.Replay(f) - } else { - go pub.Replay(f) - w.Write([]byte("ok")) - } - } else { - http.Error(w, err.Error(), http.StatusInternalServerError) - } - } -} - func (c *GB28181Config) API_bye(w http.ResponseWriter, r *http.Request) { - // CORS(w, r) id := r.URL.Query().Get("id") channel := r.URL.Query().Get("channel") - live := r.URL.Query().Get("live") + streamPath := r.URL.Query().Get("streamPath") if c := FindChannel(id, channel); c != nil { - w.WriteHeader(c.Bye(live != "false")) + w.WriteHeader(c.Bye(streamPath)) } else { http.NotFound(w, r) } diff --git a/server.go b/server.go index 858c8f5..4ebc0cd 100755 --- a/server.go +++ b/server.go @@ -1,20 +1,14 @@ package gb28181 import ( - "bufio" "context" - "encoding/binary" "fmt" - "io" - "net" "strconv" "strings" "time" "github.com/logrusorgru/aurora" - "github.com/pion/rtp" "go.uber.org/zap" - "m7s.live/engine/v4/util" "m7s.live/plugin/gb28181/v4/utils" "github.com/ghettovoice/gosip" @@ -24,21 +18,14 @@ import ( var srv gosip.Server -type Server struct { - Ignores map[string]struct{} - publishers util.Map[uint32, *GBPublisher] - tcpPorts PortManager - udpPorts PortManager -} - const MaxRegisterCount = 3 func FindChannel(deviceId string, channelId string) (c *Channel) { if v, ok := Devices.Load(deviceId); ok { d := v.(*Device) - d.channelMutex.RLock() - c = d.ChannelMap[channelId] - d.channelMutex.RUnlock() + if v, ok := d.channelMap.Load(channelId); ok { + return v.(*Channel) + } } return } @@ -129,7 +116,6 @@ func RequestForResponse(transport string, request sip.Request, } func (c *GB28181Config) startServer() { - c.publishers.Init() addr := c.ListenAddr + ":" + strconv.Itoa(int(c.SipPort)) logger := utils.NewZapLogger(plugin.Logger, "GB SIP Server", nil) @@ -151,108 +137,17 @@ func (c *GB28181Config) startServer() { plugin.Info(fmt.Sprint(aurora.Green("Server gb28181 start at"), aurora.BrightBlue(addr))) } - go c.startMediaServer() + if c.MediaNetwork == "tcp" { + c.tcpPorts.Init(c.MediaPortMin, c.MediaPortMax) + } else { + c.udpPorts.Init(c.MediaPortMin, c.MediaPortMax) + } if c.Username != "" || c.Password != "" { go c.removeBanDevice() } } -func (c *GB28181Config) startMediaServer() { - if c.MediaNetwork == "tcp" { - c.tcpPorts.Init(c.MediaPortMin, c.MediaPortMax) - if !c.tcpPorts.Valid { - c.listenMediaTCP() - } - } else { - c.udpPorts.Init(c.MediaPortMin, c.MediaPortMax) - if !c.udpPorts.Valid { - c.listenMediaUDP() - } - } -} - -func (c *GB28181Config) processTcpMediaConn(conn net.Conn) { - var rtpPacket rtp.Packet - reader := bufio.NewReader(conn) - defer conn.Close() - var err error - dumpLen := make([]byte, 6) - ps := make(util.Buffer, 1024) - for err == nil { - if _, err = io.ReadFull(reader, dumpLen[:2]); err != nil { - return - } - ps.Relloc(int(binary.BigEndian.Uint16(dumpLen[:2]))) - if _, err = io.ReadFull(reader, ps); err != nil { - return - } - if err := rtpPacket.Unmarshal(ps); err != nil { - plugin.Error("gb28181 decode rtp error:", zap.Error(err)) - } else if publisher := c.publishers.Get(rtpPacket.SSRC); publisher != nil && publisher.Publisher.Err() == nil { - publisher.writeDump(ps, dumpLen) - publisher.PushPS(&rtpPacket) - } else { - plugin.Info("gb28181 publisher not found", zap.Uint32("ssrc", rtpPacket.SSRC)) - } - } -} - -func (c *GB28181Config) listenMediaTCP() { - addr := ":" + strconv.Itoa(int(c.MediaPort)) - mediaAddr, _ := net.ResolveTCPAddr("tcp", addr) - listen, err := net.ListenTCP("tcp", mediaAddr) - - if err != nil { - plugin.Error("MediaServer listened　tcp err", zap.String("addr", addr), zap.Error(err)) - return - } - plugin.Sugar().Infof("MediaServer started tcp at %s", addr) - defer listen.Close() - defer plugin.Info("MediaServer stopped tcp at", zap.Uint16("port", c.MediaPort)) - - for { - conn, err := listen.Accept() - if err != nil { - plugin.Error("Accept err=", zap.Error(err)) - } - go c.processTcpMediaConn(conn) - } -} - -func (c *GB28181Config) listenMediaUDP() { - var rtpPacket rtp.Packet - networkBuffer := 1048576 - - addr := ":" + strconv.Itoa(int(c.MediaPort)) - mediaAddr, _ := net.ResolveUDPAddr("udp", addr) - conn, err := net.ListenUDP("udp", mediaAddr) - - if err != nil { - plugin.Error(" MediaServer started listening udp err", zap.String("addr", addr), zap.Error(err)) - return - } - bufUDP := make([]byte, networkBuffer) - plugin.Sugar().Infof("MediaServer started at udp %s", addr) - defer plugin.Sugar().Infof("MediaServer stopped at udp %s", addr) - dumpLen := make([]byte, 6) - for n, _, err := conn.ReadFromUDP(bufUDP); err == nil; n, _, err = conn.ReadFromUDP(bufUDP) { - ps := bufUDP[:n] - if err := rtpPacket.Unmarshal(ps); err != nil { - plugin.Error("Decode rtp error:", zap.Error(err)) - } - t := time.Now() - if publisher := c.publishers.Get(rtpPacket.SSRC); publisher != nil && publisher.Publisher.Err() == nil { - publisher.writeDump(ps, dumpLen) - publisher.PushPS(&rtpPacket) - } - x := time.Since(t) - if x > time.Millisecond { - fmt.Println(x) - } - } -} - // func queryCatalog(config *transaction.Config) { // t := time.NewTicker(time.Duration(config.CatalogInterval) * time.Second) // for range t.C {