Merge pull request #92 from rufftio/v4

使用 engine 的 stream pasue 和 resume 替代 neverTimeout

README.md

@@ -23,7 +23,6 @@ gb28181:
     autosubposition: false #是否自动订阅定位
     expires: 3600s #订阅周期(单位：秒)，默认3600
     interval: 6s #订阅间隔（单位：秒），默认6
-  prefetchrecord: false 
   udpcachesize:   0 #表示UDP缓存大小，默认为0，不开启。仅当TCP关闭，切缓存大于0时才开启
   sipip:          "" #sip服务器地址 默认 自动适配设备网段
   serial:         "34020000002000000001"

channel.go

@@ -17,53 +17,111 @@ import (
 	"m7s.live/plugin/ps/v4"
 )
 
+var QUERY_RECORD_TIMEOUT = time.Second * 5
+
 type PullStream struct {
 	opt       *InviteOptions
 	channel   *Channel
 	inviteRes sip.Response
 }
 
-func (p *PullStream) Bye() int {
+func (p *PullStream) CreateRequest(method sip.RequestMethod) (req sip.Request) {
 	res := p.inviteRes
-	bye := p.channel.CreateRequst(sip.BYE)
+	req = p.channel.CreateRequst(method)
 	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)
+	req.ReplaceHeaders(from.Name(), []sip.Header{from})
+	req.ReplaceHeaders(to.Name(), []sip.Header{to})
+	req.ReplaceHeaders(callId.Name(), []sip.Header{callId})
+	return
+}
+
+func (p *PullStream) Bye() int {
+	req := p.CreateRequest(sip.BYE)
+	resp, err := p.channel.device.SipRequestForResponse(req)
 	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 http.StatusInternalServerError
 	}
 	return int(resp.StatusCode())
 }
 
-type ChannelEx struct {
-	device          *Device      // 所属设备
-	status          atomic.Int32 // 通道状态,0:空闲,1:正在invite,2:正在播放
-	LiveSubSP       string       // 实时子码流，通过rtsp
-	Records         []*Record
-	RecordStartTime string
-	RecordEndTime   string
-	recordStartTime time.Time
-	recordEndTime   time.Time
-	GpsTime         time.Time //gps时间
-	Longitude       string    //经度
-	Latitude        string    //纬度
-	*log.Logger     `json:"-" yaml:"-"`
+func (p *PullStream) info(body string) int {
+	d := p.channel.device
+	req := p.CreateRequest(sip.INFO)
+	contentType := sip.ContentType("Application/MANSRTSP")
+	req.AppendHeader(&contentType)
+	req.SetBody(body, true)
+
+	resp, err := d.SipRequestForResponse(req)
+	if err != nil {
+		log.Warnf("Send info to stream error: %v, stream=%s, body=%s", err, p.opt.StreamPath, body)
+		return getSipRespErrorCode(err)
+	}
+	return int(resp.StatusCode())
+}
+
+// 暂停播放
+func (p *PullStream) Pause() int {
+	body := fmt.Sprintf(`PAUSE RTSP/1.0
+CSeq: %d
+PauseTime: now
+`, p.channel.device.sn)
+	return p.info(body)
+}
+
+// 恢复播放
+func (p *PullStream) Resume() int {
+	d := p.channel.device
+	body := fmt.Sprintf(`PLAY RTSP/1.0
+CSeq: %d
+Range: npt=now-
+`, d.sn)
+	return p.info(body)
+}
+
+// 跳转到播放时间
+// second: 相对于起始点调整到第 sec 秒播放
+func (p *PullStream) PlayAt(second uint) int {
+	d := p.channel.device
+	body := fmt.Sprintf(`PLAY RTSP/1.0
+CSeq: %d
+Range: npt=%d-
+`, d.sn, second)
+	return p.info(body)
+}
+
+// 快进/快退播放
+// speed 取值： 0.25 0.5 1 2 4 或者其对应的负数表示倒放
+func (p *PullStream) PlayForward(speed float32) int {
+	d := p.channel.device
+	body := fmt.Sprintf(`PLAY RTSP/1.0
+CSeq: %d
+Scale: %0.6f
+`, d.sn, speed)
+	return p.info(body)
+}
+
+type Channel struct {
+	device      *Device      // 所属设备
+	status      atomic.Int32 // 通道状态,0:空闲,1:正在invite,2:正在播放
+	LiveSubSP   string       // 实时子码流，通过rtsp
+	GpsTime     time.Time    //gps时间
+	Longitude   string       //经度
+	Latitude    string       //纬度
+	*log.Logger `json:"-" yaml:"-"`
+	ChannelInfo
 }
 
 // Channel 通道
-type Channel struct {
-	DeviceID     string
+type ChannelInfo struct {
+	DeviceID     string // 通道ID
 	ParentID     string
 	Name         string
 	Manufacturer string
@@ -77,8 +135,6 @@ type Channel struct {
 	RegisterWay  int
 	Secrecy      int
 	Status       string
-	Children     []*Channel `json:"-" yaml:"-"`
-	ChannelEx               //自定义属性
 }
 
 func (channel *Channel) CreateRequst(Method sip.RequestMethod) (req sip.Request) {
@@ -140,13 +196,9 @@ func (channel *Channel) CreateRequst(Method sip.RequestMethod) (req sip.Request)
 	req.SetDestination(d.NetAddr)
 	return req
 }
-func (channel *Channel) QueryRecord(startTime, endTime string) int {
+
+func (channel *Channel) QueryRecord(startTime, endTime string) ([]*Record, error) {
 	d := channel.device
-	channel.RecordStartTime = startTime
-	channel.RecordEndTime = endTime
-	channel.recordStartTime, _ = time.Parse(TIME_LAYOUT, startTime)
-	channel.recordEndTime, _ = time.Parse(TIME_LAYOUT, endTime)
-	channel.Records = nil
 	request := d.CreateRequest(sip.MESSAGE)
 	contentType := sip.ContentType("Application/MANSCDP+xml")
 	request.AppendHeader(&contentType)
@@ -161,12 +213,21 @@ func (channel *Channel) QueryRecord(startTime, endTime string) int {
 		<Type>all</Type>
 		</Query>`, d.sn, channel.DeviceID, startTime, endTime)
 	request.SetBody(body, true)
+
+	resultCh := RecordQueryLink.WaitResult(d.ID, channel.DeviceID, d.sn, QUERY_RECORD_TIMEOUT)
 	resp, err := d.SipRequestForResponse(request)
 	if err != nil {
-		return http.StatusRequestTimeout
+		return nil, fmt.Errorf("query error: %s", err)
 	}
-	return int(resp.StatusCode())
+	if resp.StatusCode() != http.StatusOK {
+		return nil, fmt.Errorf("query error, status=%d", resp.StatusCode())
+	}
+	// RecordQueryLink 中加了超时机制，该结果一定会返回
+	// 所以此处不用再增加超时等保护机制
+	r := <-resultCh
+	return r.list, r.err
 }
+
 func (channel *Channel) Control(PTZCmd string) int {
 	d := channel.device
 	request := d.CreateRequest(sip.MESSAGE)
@@ -264,6 +325,8 @@ func (channel *Channel) Invite(opt *InviteOptions) (code int, err error) {
 	}
 	if opt.StreamPath != "" {
 		streamPath = opt.StreamPath
+	} else {
+		opt.StreamPath = streamPath
 	}
 	if opt.dump == "" {
 		opt.dump = conf.DumpPath
@@ -326,7 +389,7 @@ func (channel *Channel) Invite(opt *InviteOptions) (code int, err error) {
 	code = int(inviteRes.StatusCode())
 	channel.Info("invite response", zap.Int("status code", code))
 
-	if code == OK {
+	if code == http.StatusOK {
 		ds := strings.Split(inviteRes.Body(), "\r\n")
 		for _, l := range ds {
 			if ls := strings.Split(l, "="); len(ls) > 1 {
@@ -368,8 +431,51 @@ func (channel *Channel) Bye(streamPath string) int {
 	return http.StatusNotFound
 }
 
+func (channel *Channel) Pause(streamPath string) int {
+	if s, loaded := PullStreams.Load(streamPath); loaded {
+		r := s.(*PullStream).Pause()
+		if s := Streams.Get(streamPath); s != nil {
+			s.Pause()
+		}
+		return r
+	}
+	return http.StatusNotFound
+}
+
+func (channel *Channel) Resume(streamPath string) int {
+	if s, loaded := PullStreams.Load(streamPath); loaded {
+		r := s.(*PullStream).Resume()
+		if s := Streams.Get(streamPath); s != nil {
+			s.Resume()
+		}
+		return r
+	}
+	return http.StatusNotFound
+}
+
+func (channel *Channel) PlayAt(streamPath string, second uint) int {
+	if s, loaded := PullStreams.Load(streamPath); loaded {
+		r := s.(*PullStream).PlayAt(second)
+		if s := Streams.Get(streamPath); s != nil {
+			s.Resume()
+		}
+		return r
+	}
+	return http.StatusNotFound
+}
+
+func (channel *Channel) PlayForward(streamPath string, speed float32) int {
+	if s, loaded := PullStreams.Load(streamPath); loaded {
+		return s.(*PullStream).PlayForward(speed)
+	}
+	if s := Streams.Get(streamPath); s != nil {
+		s.Resume()
+	}
+	return http.StatusNotFound
+}
+
 func (channel *Channel) TryAutoInvite(opt *InviteOptions) {
-	if conf.InviteMode == 1 && channel.CanInvite() {
+	if channel.CanInvite() {
 		go channel.Invite(opt)
 	}
 }
@@ -402,3 +508,11 @@ func (channel *Channel) CanInvite() bool {
 
 	return false
 }
+
+func getSipRespErrorCode(err error) int {
+	if re, ok := err.(*sip.RequestError); ok {
+		return int(re.Code)
+	} else {
+		return http.StatusInternalServerError
+	}
+}

const.go

@@ -1,60 +1,5 @@
 package gb28181
 
-const (
-	Trying                      = 100
-	Ringing                     = 180
-	CallIsBeingForwarded        = 181
-	Queued                      = 182
-	SessionProgress             = 183
-	OK                          = 200
-	Accepted                    = 202
-	MultipleChoices             = 300
-	MovedPermanently            = 301
-	MovedTemporarily            = 302
-	UseProxy                    = 305
-	AlternativeService          = 380
-	BadRequest                  = 400
-	Unauthorized                = 401
-	PaymentRequired             = 402
-	Forbidden                   = 403
-	NotFound                    = 404
-	MethodNotAllowed            = 405
-	NotAcceptable               = 406
-	ProxyAuthenticationRequired = 407
-	RequestTimeout              = 408
-	Gone                        = 410
-	RequestEntityTooLarge       = 413
-	RequestURITooLong           = 414
-	UnsupportedMediaType        = 415
-	UnsupportedURIScheme        = 416
-	BadExtension                = 420
-	ExtensionRequired           = 421
-	IntervalTooBrief            = 423
-	TemporarilyUnavailable      = 480
-	CallTransactionDoesNotExist = 481
-	LoopDetected                = 482
-	TooManyHops                 = 483
-	AddressIncomplete           = 484
-	Ambiguous                   = 485
-	BusyHere                    = 486
-	RequestTerminated           = 487
-	NotAcceptableHere           = 488
-	BadEvent                    = 489
-	RequestPending              = 491
-	Undecipherable              = 493
-	ServerInternalError         = 500
-	NotImplemented              = 501
-	BadGateway                  = 502
-	ServiceUnavailable          = 503
-	ServerTim                   = 504
-	VersionNotSupported         = 505
-	MessageTooLarge             = 513
-	BusyEverywhere              = 600
-	Decline                     = 603
-	DoesNotExistAnywhere        = 604
-	SessionNotAcceptable        = 606
-)
-
 var reasons = map[int]string{
 	100: "Trying",
 	180: "Ringing",
@@ -113,3 +58,9 @@ var reasons = map[int]string{
 func Explain(statusCode int) string {
 	return reasons[statusCode]
 }
+
+const (
+	INVIDE_MODE_MANUAL = iota
+	INVIDE_MODE_AUTO
+	INVIDE_MODE_ONSUBSCRIBE
+)

device.go

@@ -24,7 +24,6 @@ const TIME_LAYOUT = "2006-01-02T15:04:05"
 
 // Record 录像
 type Record struct {
-	//channel   *Channel
 	DeviceID  string
 	Name      string
 	FilePath  string
@@ -76,14 +75,14 @@ type Device struct {
 func (d *Device) MarshalJSON() ([]byte, error) {
 	type Alias Device
 	data := &struct {
-		Channels []*Channel
+		Channels []*ChannelInfo
 		*Alias
 	}{
 		Alias: (*Alias)(d),
 	}
 	d.channelMap.Range(func(key, value interface{}) bool {
 		c := value.(*Channel)
-		data.Channels = append(data.Channels, c)
+		data.Channels = append(data.Channels, &c.ChannelInfo)
 		return true
 	})
 	return json.Marshal(data)
@@ -201,32 +200,31 @@ func (c *GB28181Config) SaveDevices() {
 	}
 }
 
-func (d *Device) addOrUpdateChannel(channel *Channel) {
-	if old, ok := d.channelMap.Load(channel.DeviceID); ok {
-		channel.ChannelEx = old.(*Channel).ChannelEx
+func (d *Device) addOrUpdateChannel(info ChannelInfo) (c *Channel) {
+	if old, ok := d.channelMap.Load(info.DeviceID); ok {
+		c = old.(*Channel)
+		c.ChannelInfo = info
+	} else {
+		c = &Channel{
+			device:      d,
+			ChannelInfo: info,
+			Logger:      d.Logger.With(zap.String("channel", info.DeviceID)),
+		}
+		if s := engine.Streams.Get(fmt.Sprintf("%s/%s/rtsp", c.device.ID, c.DeviceID)); s != nil {
+			c.LiveSubSP = s.Path
+		} else {
+			c.LiveSubSP = ""
+		}
+		d.channelMap.Store(info.DeviceID, c)
 	}
-	channel.device = d
-	channel.Logger = d.Logger.With(zap.String("channel", channel.DeviceID), zap.String("name", channel.Name))
-	d.channelMap.Store(channel.DeviceID, channel)
+	return
 }
 
 func (d *Device) deleteChannel(DeviceID string) {
 	d.channelMap.Delete(DeviceID)
 }
 
-func (d *Device) CheckSubStream() {
-	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) {
-
+func (d *Device) UpdateChannels(list ...ChannelInfo) {
 	for _, c := range list {
 		if _, ok := conf.Ignores[c.DeviceID]; ok {
 			continue
@@ -249,32 +247,18 @@ func (d *Device) UpdateChannels(list []*Channel) {
 			}
 		}
 		//本设备增加通道
-		d.addOrUpdateChannel(c)
+		channel := d.addOrUpdateChannel(c)
 
-		//预取和邀请
-		if conf.PreFetchRecord {
-			n := time.Now()
-			n = time.Date(n.Year(), n.Month(), n.Day(), 0, 0, 0, 0, time.Local)
-			if len(c.Records) == 0 || (n.Format(TIME_LAYOUT) == c.RecordStartTime &&
-				n.Add(time.Hour*24-time.Second).Format(TIME_LAYOUT) == c.RecordEndTime) {
-				go c.QueryRecord(n.Format(TIME_LAYOUT), n.Add(time.Hour*24-time.Second).Format(TIME_LAYOUT))
-			}
+		if conf.InviteMode == INVIDE_MODE_AUTO {
+			channel.TryAutoInvite(&InviteOptions{})
 		}
-		c.TryAutoInvite(&InviteOptions{})
 		if s := engine.Streams.Get("sub/" + c.DeviceID); s != nil {
-			c.LiveSubSP = s.Path
+			channel.LiveSubSP = s.Path
 		} else {
-			c.LiveSubSP = ""
+			channel.LiveSubSP = ""
 		}
 	}
 }
-func (d *Device) UpdateRecord(channelId string, list []*Record) {
-	d.channelMap.Range(func(key, value any) bool {
-		c := value.(*Channel)
-		c.Records = append(c.Records, list...)
-		return true
-	})
-}
 
 func (d *Device) CreateRequest(Method sip.RequestMethod) (req sip.Request) {
 	d.sn++
@@ -358,7 +342,7 @@ func (d *Device) Subscribe() int {
 
 	response, err := d.SipRequestForResponse(request)
 	if err == nil && response != nil {
-		if response.StatusCode() == OK {
+		if response.StatusCode() == http.StatusOK {
 			callId, _ := request.CallID()
 			d.subscriber.CallID = string(*callId)
 		} else {
@@ -409,7 +393,7 @@ func (d *Device) QueryDeviceInfo() {
 			// 	d.SipIP = received.String()
 			// }
 			d.Info("QueryDeviceInfo", zap.Uint16("status code", uint16(response.StatusCode())))
-			if response.StatusCode() == OK {
+			if response.StatusCode() == http.StatusOK {
 				break
 			}
 		}
@@ -437,7 +421,7 @@ func (d *Device) MobilePositionSubscribe(id string, expires time.Duration, inter
 
 	response, err := d.SipRequestForResponse(mobilePosition)
 	if err == nil && response != nil {
-		if response.StatusCode() == OK {
+		if response.StatusCode() == http.StatusOK {
 			callId, _ := mobilePosition.CallID()
 			d.subscriber.CallID = callId.String()
 		} else {
@@ -452,9 +436,9 @@ func (d *Device) MobilePositionSubscribe(id string, expires time.Duration, inter
 func (d *Device) UpdateChannelPosition(channelId string, gpsTime string, lng string, lat string) {
 	if v, ok := d.channelMap.Load(channelId); ok {
 		c := v.(*Channel)
-		c.ChannelEx.GpsTime = time.Now() //时间取系统收到的时间，避免设备时间和格式问题
-		c.ChannelEx.Longitude = lng
-		c.ChannelEx.Latitude = lat
+		c.GpsTime = time.Now() //时间取系统收到的时间，避免设备时间和格式问题
+		c.Longitude = lng
+		c.Latitude = lat
 		c.Debug("update channel position success")
 	} else {
 		//如果未找到通道，则更新到设备上
@@ -483,7 +467,7 @@ func (d *Device) UpdateChannelStatus(deviceList []*notifyMessage) {
 			d.channelOffline(v.DeviceID)
 		case "ADD":
 			d.Debug("receive channel add notify")
-			channel := Channel{
+			channel := ChannelInfo{
 				DeviceID:     v.DeviceID,
 				ParentID:     v.ParentID,
 				Name:         v.Name,
@@ -499,7 +483,7 @@ func (d *Device) UpdateChannelStatus(deviceList []*notifyMessage) {
 				Secrecy:      v.Secrecy,
 				Status:       v.Status,
 			}
-			d.addOrUpdateChannel(&channel)
+			d.addOrUpdateChannel(channel)
 		case "DEL":
 			//删除
 			d.Debug("receive channel delete notify")
@@ -507,7 +491,7 @@ func (d *Device) UpdateChannelStatus(deviceList []*notifyMessage) {
 		case "UPDATE":
 			d.Debug("receive channel update notify")
 			// 更新通道
-			channel := &Channel{
+			channel := ChannelInfo{
 				DeviceID:     v.DeviceID,
 				ParentID:     v.ParentID,
 				Name:         v.Name,
@@ -523,8 +507,7 @@ func (d *Device) UpdateChannelStatus(deviceList []*notifyMessage) {
 				Secrecy:      v.Secrecy,
 				Status:       v.Status,
 			}
-			channels := []*Channel{channel}
-			d.UpdateChannels(channels)
+			d.UpdateChannels(channel)
 		}
 	}
 }

handle.go

@@ -52,8 +52,11 @@ func (a *Authorization) getDigest(raw string) string {
 }
 
 func (c *GB28181Config) OnRegister(req sip.Request, tx sip.ServerTransaction) {
-	from, _ := req.From()
-
+	from, ok := req.From()
+	if !ok {
+		GB28181Plugin.Error("OnRegister", zap.String("error", "no from"))
+		return
+	}
 	id := from.Address.User().String()
 	GB28181Plugin.Info("OnRegister", zap.String("id", id), zap.String("source", req.Source()), zap.String("destination", req.Destination()))
 	if len(id) != 20 {
@@ -162,13 +165,15 @@ func (c *GB28181Config) OnMessage(req sip.Request, tx sip.ServerTransaction) {
 		temp := &struct {
 			XMLName      xml.Name
 			CmdType      string
+			SN           int // 请求序列号，一般用于对应 request 和 response
 			DeviceID     string
 			DeviceName   string
 			Manufacturer string
 			Model        string
 			Channel      string
-			DeviceList   []*Channel `xml:"DeviceList>Item"`
-			RecordList   []*Record  `xml:"RecordList>Item"`
+			DeviceList   []ChannelInfo `xml:"DeviceList>Item"`
+			RecordList   []*Record     `xml:"RecordList>Item"`
+			SumNum       int           // 录像结果的总数 SumNum，录像结果会按照多条消息返回，可用于判断是否全部返回
 		}{}
 		decoder := xml.NewDecoder(bytes.NewReader([]byte(req.Body())))
 		decoder.CharsetReader = charset.NewReaderLabel
@@ -188,22 +193,21 @@ func (c *GB28181Config) OnMessage(req sip.Request, tx sip.ServerTransaction) {
 				go d.syncChannels()
 			} else {
 				d.channelMap.Range(func(key, value interface{}) bool {
-					channel := value.(*Channel)
-					channel.TryAutoInvite(&InviteOptions{})
+					if conf.InviteMode == INVIDE_MODE_AUTO {
+						value.(*Channel).TryAutoInvite(&InviteOptions{})
+					}
 					return true
 				})
 			}
-			//为什么要查找子码流?
-			//d.CheckSubStream()
 			//在KeepLive 进行位置订阅的处理，如果开启了自动订阅位置，则去订阅位置
 			if c.Position.AutosubPosition && time.Since(d.GpsTime) > c.Position.Interval*2 {
 				d.MobilePositionSubscribe(d.ID, c.Position.Expires, c.Position.Interval)
 				GB28181Plugin.Debug("Mobile Position Subscribe", zap.String("deviceID", d.ID))
 			}
 		case "Catalog":
-			d.UpdateChannels(temp.DeviceList)
+			d.UpdateChannels(temp.DeviceList...)
 		case "RecordInfo":
-			d.UpdateRecord(temp.DeviceID, temp.RecordList)
+			RecordQueryLink.Put(d.ID, temp.DeviceID, temp.SN, temp.SumNum, temp.RecordList)
 		case "DeviceInfo":
 			// 主设备信息
 			d.Name = temp.DeviceName

link.go

@@ -0,0 +1,124 @@
+package gb28181
+
+import (
+	"fmt"
+	"sync"
+	"time"
+
+	"go.uber.org/zap"
+)
+
+// 对于录像查询，通过 queryKey (即 deviceId + channelId + sn) 唯一区分一次请求和响应
+// 并将其关联起来，以实现异步响应的目的
+// 提供单例实例供调用
+var RecordQueryLink = NewRecordQueryLink(time.Second * 60)
+
+type recordQueryLink struct {
+	pendingResult map[string]recordQueryResult // queryKey 查询结果缓存
+	pendingResp   map[string]recordQueryResp   // queryKey 待回复的查询请求
+	timeout       time.Duration                // 查询结果的过期时间
+	sync.RWMutex
+}
+
+type recordQueryResult struct {
+	time     time.Time
+	err      error
+	sum      int
+	finished bool
+	list     []*Record
+}
+type recordQueryResp struct {
+	respChan  chan<- recordQueryResult
+	timeout   time.Duration
+	startTime time.Time
+}
+
+func NewRecordQueryLink(resultTimeout time.Duration) *recordQueryLink {
+	c := &recordQueryLink{
+		timeout:       resultTimeout,
+		pendingResult: make(map[string]recordQueryResult),
+		pendingResp:   make(map[string]recordQueryResp),
+	}
+	go c.cleanTimeout()
+	return c
+}
+
+// 唯一区分一次录像查询
+func recordQueryKey(deviceId, channelId string, sn int) string {
+	return fmt.Sprintf("%s-%s-%d", deviceId, channelId, sn)
+}
+
+// 定期清理过期的查询结果和请求
+func (c *recordQueryLink) cleanTimeout() {
+	tick := time.NewTicker(time.Millisecond * 100)
+	for {
+		<-tick.C
+		for k, s := range c.pendingResp {
+			if time.Since(s.startTime) > s.timeout {
+				if r, ok := c.pendingResult[k]; ok {
+					c.notify(k, r)
+				} else {
+					c.notify(k, recordQueryResult{err: fmt.Errorf("query time out")})
+				}
+			}
+		}
+		for k, r := range c.pendingResult {
+			if time.Since(r.time) > c.timeout {
+				delete(c.pendingResult, k)
+			}
+		}
+	}
+}
+
+func (c *recordQueryLink) Put(deviceId, channelId string, sn int, sum int, record []*Record) {
+	key, r := c.doPut(deviceId, channelId, sn, sum, record)
+	if r.finished {
+		c.notify(key, r)
+	}
+}
+
+func (c *recordQueryLink) doPut(deviceId, channelId string, sn, sum int, record []*Record) (key string, r recordQueryResult) {
+	c.Lock()
+	defer c.Unlock()
+	key = recordQueryKey(deviceId, channelId, sn)
+	if v, ok := c.pendingResult[key]; ok {
+		r = v
+	} else {
+		r = recordQueryResult{time: time.Now(), sum: sum, list: make([]*Record, 0)}
+	}
+
+	r.list = append(r.list, record...)
+	if len(r.list) == sum {
+		r.finished = true
+	}
+	c.pendingResult[key] = r
+	GB28181Plugin.Logger.Debug("put record",
+		zap.String("key", key),
+		zap.Int("sum", sum),
+		zap.Int("count", len(r.list)))
+	return
+}
+
+func (c *recordQueryLink) WaitResult(
+	deviceId, channelId string, sn int,
+	timeout time.Duration) (resultCh <-chan recordQueryResult) {
+
+	key := recordQueryKey(deviceId, channelId, sn)
+	c.Lock()
+	defer c.Unlock()
+	respCh := make(chan recordQueryResult, 1)
+	resultCh = respCh
+	c.pendingResp[key] = recordQueryResp{startTime: time.Now(), timeout: timeout, respChan: respCh}
+	return
+}
+
+func (c *recordQueryLink) notify(key string, r recordQueryResult) {
+	if s, ok := c.pendingResp[key]; ok {
+		s.respChan <- r
+	}
+	c.Lock()
+	defer c.Unlock()
+	delete(c.pendingResp, key)
+	delete(c.pendingResult, key)
+	GB28181Plugin.Logger.Debug("record notify", zap.String("key", key))
+}

main.go

@@ -19,11 +19,9 @@ type GB28181PositionConfig struct {
 }
 
 type GB28181Config struct {
-	// 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"`
+	InviteMode int    `default:"1"` //邀请模式，0:手动拉流，1:预拉流，2:按需拉流
+	InviteIDs  string //按照国标gb28181协议允许邀请的设备类型:132 摄像机 NVR
+	ListenAddr string `default:"0.0.0.0"`
 	//sip服务器的配置
 	SipNetwork string   `default:"udp"` //传输协议，默认UDP，可选TCP
 	SipIP      string   //sip 服务器公网IP
@@ -99,12 +97,19 @@ func (c *GB28181Config) OnEvent(event any) {
 		go c.initRoutes()
 		c.startServer()
 	case *Stream:
-		if c.InviteMode == 2 {
+		if c.InviteMode == INVIDE_MODE_ONSUBSCRIBE {
 			if channel := FindChannel(e.AppName, e.StreamName); channel != nil {
 				channel.TryAutoInvite(&InviteOptions{})
 			}
 		}
+	case SEpublish:
+		if channel := FindChannel(e.Target.AppName, strings.TrimSuffix(e.Target.StreamName, "/rtsp")); channel != nil {
+			channel.LiveSubSP = e.Target.Path
+		}
 	case SEclose:
+		if channel := FindChannel(e.Target.AppName, strings.TrimSuffix(e.Target.StreamName, "/rtsp")); channel != nil {
+			channel.LiveSubSP = ""
+		}
 		if v, ok := PullStreams.LoadAndDelete(e.Target.Path); ok {
 			go v.(*PullStream).Bye()
 		}

ptz.go

@@ -0,0 +1,47 @@
+package gb28181
+
+import "fmt"
+
+var (
+	name2code = map[string]uint8{
+		"stop":      0,
+		"right":     1,
+		"left":      2,
+		"down":      4,
+		"downright": 5,
+		"downleft":  6,
+		"up":        8,
+		"upright":   9,
+		"upleft":    10,
+		"zoomin":    16,
+		"zoomout":   32,
+	}
+)
+
+func toPtzStrByCmdName(cmdName string, horizontalSpeed, verticalSpeed, zoomSpeed uint8) (string, error) {
+	c, err := toPtzCode(cmdName)
+	if err != nil {
+		return "", err
+	}
+	return toPtzStr(c, horizontalSpeed, verticalSpeed, zoomSpeed), nil
+}
+
+func toPtzStr(cmdCode, horizontalSpeed, verticalSpeed, zoomSpeed uint8) string {
+	checkCode := uint16(0xA5+0x0F+0x01+cmdCode+horizontalSpeed+verticalSpeed+(zoomSpeed&0xF0)) % 0x100
+
+	return fmt.Sprintf("A50F01%02X%02X%02X%01X0%02X",
+		cmdCode,
+		horizontalSpeed,
+		verticalSpeed,
+		zoomSpeed>>4, // 根据 GB28181 协议，zoom 只取 4 bit
+		checkCode,
+	)
+}
+
+func toPtzCode(cmd string) (uint8, error) {
+	if code, ok := name2code[cmd]; ok {
+		return code, nil
+	} else {
+		return 0, fmt.Errorf("invalid ptz cmd %q", cmd)
+	}
+}

restful.go

@@ -1,15 +1,23 @@
 package gb28181
 
 import (
+	"encoding/json"
+	"fmt"
 	"net/http"
 	"strconv"
+	"strings"
 	"time"
 
 	"m7s.live/engine/v4/util"
 )
 
+var (
+	playScaleValues = map[float32]bool{0.25: true, 0.5: true, 1: true, 2: true, 4: true}
+)
+
 func (c *GB28181Config) API_list(w http.ResponseWriter, r *http.Request) {
 	util.ReturnJson(func() (list []*Device) {
+		list = make([]*Device, 0)
 		Devices.Range(func(key, value interface{}) bool {
 			device := value.(*Device)
 			if time.Since(device.UpdateTime) > c.RegisterValidity {
@@ -24,12 +32,23 @@ func (c *GB28181Config) API_list(w http.ResponseWriter, r *http.Request) {
 }
 
 func (c *GB28181Config) API_records(w http.ResponseWriter, r *http.Request) {
-	id := r.URL.Query().Get("id")
-	channel := r.URL.Query().Get("channel")
-	startTime := r.URL.Query().Get("startTime")
-	endTime := r.URL.Query().Get("endTime")
+	query := r.URL.Query()
+	id := query.Get("id")
+	channel := query.Get("channel")
+	startTime := query.Get("startTime")
+	endTime := query.Get("endTime")
+	trange := strings.Split(query.Get("range"), "-")
+	if len(trange) == 2 {
+		startTime = trange[0]
+		endTime = trange[1]
+	}
 	if c := FindChannel(id, channel); c != nil {
-		w.WriteHeader(c.QueryRecord(startTime, endTime))
+		res, err := c.QueryRecord(startTime, endTime)
+		if err == nil {
+			WriteJSONOk(w, res)
+		} else {
+			WriteJSON(w, err.Error(), http.StatusInternalServerError)
+		}
 	} else {
 		http.NotFound(w, r)
 	}
@@ -46,6 +65,44 @@ func (c *GB28181Config) API_control(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
+func (c *GB28181Config) API_ptz(w http.ResponseWriter, r *http.Request) {
+	q := r.URL.Query()
+	id := q.Get("id")
+	channel := q.Get("channel")
+	cmd := q.Get("cmd")   // 命令名称，见 ptz.go name2code 定义
+	hs := q.Get("hSpeed") // 水平速度
+	vs := q.Get("vSpeed") // 垂直速度
+	zs := q.Get("zSpeed") // 缩放速度
+
+	hsN, err := strconv.ParseUint(hs, 10, 8)
+	if err != nil {
+		WriteJSON(w, "hSpeed parameter is invalid", 400)
+		return
+	}
+	vsN, err := strconv.ParseUint(vs, 10, 8)
+	if err != nil {
+		WriteJSON(w, "vSpeed parameter is invalid", 400)
+		return
+	}
+	zsN, err := strconv.ParseUint(zs, 10, 8)
+	if err != nil {
+		WriteJSON(w, "zSpeed parameter is invalid", 400)
+		return
+	}
+
+	ptzcmd, err := toPtzStrByCmdName(cmd, uint8(hsN), uint8(vsN), uint8(zsN))
+	if err != nil {
+		WriteJSON(w, err.Error(), 400)
+		return
+	}
+	if c := FindChannel(id, channel); c != nil {
+		code := c.Control(ptzcmd)
+		WriteJSON(w, "device received", code)
+	} else {
+		WriteJSON(w, fmt.Sprintf("device %q channel %q not found", id, channel), 404)
+	}
+}
+
 func (c *GB28181Config) API_invite(w http.ResponseWriter, r *http.Request) {
 	query := r.URL.Query()
 	id := query.Get("id")
@@ -57,11 +114,18 @@ func (c *GB28181Config) API_invite(w http.ResponseWriter, r *http.Request) {
 		MediaPort:  uint16(port),
 		StreamPath: streamPath,
 	}
-	opt.Validate(query.Get("startTime"), query.Get("endTime"))
+	startTime := query.Get("startTime")
+	endTime := query.Get("endTime")
+	trange := strings.Split(query.Get("range"), "-")
+	if len(trange) == 2 {
+		startTime = trange[0]
+		endTime = trange[1]
+	}
+	opt.Validate(startTime, endTime)
 	if c := FindChannel(id, channel); c == nil {
 		http.NotFound(w, r)
 	} else if opt.IsLive() && c.status.Load() > 0 {
-		w.WriteHeader(304) //直播流已存在
+		http.Error(w, "live stream already exists", http.StatusNotModified)
 	} else if code, err := c.Invite(&opt); err == nil {
 		w.WriteHeader(code)
 	} else {
@@ -80,6 +144,63 @@ func (c *GB28181Config) API_bye(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
+func (c *GB28181Config) API_play_pause(w http.ResponseWriter, r *http.Request) {
+	id := r.URL.Query().Get("id")
+	channel := r.URL.Query().Get("channel")
+	streamPath := r.URL.Query().Get("streamPath")
+	if c := FindChannel(id, channel); c != nil {
+		w.WriteHeader(c.Pause(streamPath))
+	} else {
+		http.NotFound(w, r)
+	}
+}
+
+func (c *GB28181Config) API_play_resume(w http.ResponseWriter, r *http.Request) {
+	id := r.URL.Query().Get("id")
+	channel := r.URL.Query().Get("channel")
+	streamPath := r.URL.Query().Get("streamPath")
+	if c := FindChannel(id, channel); c != nil {
+		w.WriteHeader(c.Resume(streamPath))
+	} else {
+		http.NotFound(w, r)
+	}
+}
+
+func (c *GB28181Config) API_play_seek(w http.ResponseWriter, r *http.Request) {
+	id := r.URL.Query().Get("id")
+	channel := r.URL.Query().Get("channel")
+	streamPath := r.URL.Query().Get("streamPath")
+	secStr := r.URL.Query().Get("second")
+	sec, err := strconv.ParseUint(secStr, 10, 32)
+	if err != nil {
+		WriteJSON(w, "second parameter is invalid: "+err.Error(), 400)
+		return
+	}
+	if c := FindChannel(id, channel); c != nil {
+		w.WriteHeader(c.PlayAt(streamPath, uint(sec)))
+	} else {
+		http.NotFound(w, r)
+	}
+}
+
+func (c *GB28181Config) API_play_forward(w http.ResponseWriter, r *http.Request) {
+	id := r.URL.Query().Get("id")
+	channel := r.URL.Query().Get("channel")
+	streamPath := r.URL.Query().Get("streamPath")
+	speedStr := r.URL.Query().Get("speed")
+	speed, err := strconv.ParseFloat(speedStr, 32)
+	secondErrMsg := "speed parameter is invalid, should be one of 0.25,0.5,1,2,4"
+	if err != nil || !playScaleValues[float32(speed)] {
+		WriteJSON(w, secondErrMsg, 400)
+		return
+	}
+	if c := FindChannel(id, channel); c != nil {
+		w.WriteHeader(c.PlayForward(streamPath, float32(speed)))
+	} else {
+		http.NotFound(w, r)
+	}
+}
+
 func (c *GB28181Config) API_position(w http.ResponseWriter, r *http.Request) {
 	//CORS(w, r)
 	query := r.URL.Query()
@@ -135,3 +256,13 @@ func (c *GB28181Config) API_get_position(w http.ResponseWriter, r *http.Request)
 		return
 	}, c.Position.Interval, w, r)
 }
+
+func WriteJSONOk(w http.ResponseWriter, data interface{}) {
+	WriteJSON(w, data, 200)
+}
+
+func WriteJSON(w http.ResponseWriter, data interface{}, status int) {
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(status)
+	json.NewEncoder(w).Encode(data)
+}