feat: 适配倍速播放接口

api.go

@@ -88,7 +88,7 @@ type RecordParams struct {
 type StreamIDParams struct {
 	StreamID common.StreamID `json:"streamid"`
 	Command  string          `json:"command"`
-	Scale    int             `json:"scale"`
+	Scale    float64         `json:"scale"`
 }
 
 type PageQuery struct {
@@ -1680,6 +1680,30 @@ func (api *ApiServer) OnRecordStop(writer http.ResponseWriter, request *http.Req
 	common.HttpForwardTo("/api/v1/record/stop", writer, request)
 }
 
-func (api *ApiServer) OnPlaybackControl(params *StreamParams, w http.ResponseWriter, req *http.Request) (interface{}, error) {
+func (api *ApiServer) OnPlaybackControl(params *StreamIDParams, w http.ResponseWriter, req *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{device}
+	s.ScalePlayback(stream.Dialog, params.Scale)
+	err = stack.MSSpeedSet(string(params.StreamID), params.Scale)
+	if err != nil {
+		return nil, err
+	}
+
 	return "OK", nil
 }

common/http_request.go

@@ -53,18 +53,35 @@ func parseQueryParams(c func(key string) string, v interface{}) (interface{}, er
 		switch fieldValue.Kind() {
 		case reflect.String:
 			fieldValue.SetString(value)
+			break
 		case reflect.Int:
 			intValue, err := strconv.Atoi(value)
 			if err != nil {
 				return nil, err
 			}
 			fieldValue.SetInt(int64(intValue))
+			break
 		case reflect.Bool:
 			boolValue, err := strconv.ParseBool(value)
 			if err != nil {
 				return nil, err
 			}
 			fieldValue.SetBool(boolValue)
+			break
+		case reflect.Float64:
+			floatValue, err := strconv.ParseFloat(value, 64)
+			if err != nil {
+				return nil, err
+			}
+			fieldValue.SetFloat(floatValue)
+			break
+		case reflect.Float32:
+			floatValue, err := strconv.ParseFloat(value, 32)
+			if err != nil {
+				return nil, err
+			}
+			fieldValue.SetFloat(floatValue)
+			break
 		}
 	}
 

dao/device.go

@@ -125,7 +125,7 @@ func (d *daoDevice) RefreshHeartbeat(deviceId string, now time.Time, addr string
 	return DBTransaction(func(tx *gorm.DB) error {
 		host, p, _ := net.SplitHostPort(addr)
 		port, _ := strconv.Atoi(p)
-		return tx.Model(&DeviceModel{}).Select("LastHeartbeat", "Status", "RemoteAddr").Where("device_id =?", deviceId).Updates(&DeviceModel{
+		return tx.Model(&DeviceModel{}).Select("LastHeartbeat", "Status", "RemoteIP", "RemotePort").Where("device_id =?", deviceId).Updates(&DeviceModel{
 			LastHeartbeat: now,
 			Status:        common.ON,
 			RemoteIP:      host,

stack/device.go

@@ -43,6 +43,8 @@ var (
 	XmlMessageType sip.ContentType = "Application/MANSCDP+xml"
 
 	SDPMessageType sip.ContentType = "application/sdp"
+
+	RTSPMessageType sip.ContentType = "application/RTSP"
 )
 
 type GBDevice interface {
@@ -456,7 +458,6 @@ func CreateDialogRequestFromAnswer(message sip.Response, uas bool, remoteAddr st
 	id, _ := message.CallID()
 
 	requestLine := &sip.SipUri{}
-	requestLine.SetUser(from.Address.User())
 	host, port, _ := net.SplitHostPort(remoteAddr)
 	portInt, _ := strconv.Atoi(port)
 	sipPort := sip.Port(portInt)
@@ -467,9 +468,11 @@ func CreateDialogRequestFromAnswer(message sip.Response, uas bool, remoteAddr st
 
 	builder := NewSIPRequestBuilderWithTransport(message.Transport())
 	if uas {
+		requestLine.SetUser(from.Address.User())
 		builder.SetFrom(sip.NewAddressFromToHeader(to))
 		builder.SetTo(sip.NewAddressFromFromHeader(from))
 	} else {
+		requestLine.SetUser(to.Address.User())
 		builder.SetFrom(sip.NewAddressFromFromHeader(from))
 		builder.SetTo(sip.NewAddressFromToHeader(to))
 	}

stack/live.go

@@ -88,7 +88,7 @@ func (d *Device) Invite(inviteType common.InviteType, streamId common.StreamID,
 	}()
 
 	// 告知流媒体服务创建国标源, 返回收流地址信息
-	ip, port, urls, ssrc, msErr := MSCreateGBSource(string(streamId), setup, "", string(inviteType), speed)
+	ip, port, urls, ssrc, msErr := MSCreateGBSource(string(streamId), setup, "", string(inviteType), float64(speed))
 	if msErr != nil {
 		log.Sugar.Errorf("创建GBSource失败 err: %s", msErr.Error())
 		return nil, nil, msErr

stack/media_server.go

@@ -49,15 +49,15 @@ type SinkDetails struct {
 }
 
 type SDP struct {
-	SessionName string `json:"session_name,omitempty"` // play/download/playback/talk/broadcast
-	Addr        string `json:"addr,omitempty"`         // 连接地址
-	SSRC        string `json:"ssrc,omitempty"`
-	Setup       string `json:"setup,omitempty"`     // active/passive
-	Transport   string `json:"transport,omitempty"` // tcp/udp
-	Speed       int    `json:"speed"`
-	StartTime   int    `json:"start_time,omitempty"`
-	EndTime     int    `json:"end_time,omitempty"`
-	FileSize    int    `json:"file_size,omitempty"`
+	SessionName string  `json:"session_name,omitempty"` // play/download/playback/talk/broadcast
+	Addr        string  `json:"addr,omitempty"`         // 连接地址
+	SSRC        string  `json:"ssrc,omitempty"`
+	Setup       string  `json:"setup,omitempty"`     // active/passive
+	Transport   string  `json:"transport,omitempty"` // tcp/udp
+	Speed       float64 `json:"speed"`
+	StartTime   int     `json:"start_time,omitempty"`
+	EndTime     int     `json:"end_time,omitempty"`
+	FileSize    int     `json:"file_size,omitempty"`
 }
 
 type SourceSDP struct {
@@ -103,7 +103,7 @@ func SendWithUrlParams(path string, body interface{}, values url.Values) (*http.
 	return client.Do(request)
 }
 
-func MSCreateGBSource(id, setup string, ssrc string, sessionName string, speed int) (string, uint16, []string, string, error) {
+func MSCreateGBSource(id, setup string, ssrc string, sessionName string, speed float64) (string, uint16, []string, string, error) {
 	v := &SourceSDP{
 		Source: id,
 		SDP: SDP{
@@ -274,3 +274,15 @@ func MSQueryStreamInfo(header http.Header, queryParams string) (*http.Response,
 
 	return client.Do(proxyReq)
 }
+
+func MSSpeedSet(id string, speed float64) error {
+	v := &SourceSDP{
+		Source: id,
+		SDP: SDP{
+			Speed: speed,
+		},
+	}
+
+	_, err := Send("api/v1/gb28181/speed/set", v)
+	return err
+}

stack/playback.go

@@ -0,0 +1,34 @@
+package stack
+
+import (
+	"fmt"
+	"gb-cms/common"
+	"github.com/ghettovoice/gosip/sip"
+)
+
+const (
+	RTSPBodyFormat = "PLAY RTSP/1.0\r\n" +
+		"CSeq: %d\r\n" +
+		"Scale: %.1f\r\n"
+)
+
+func (d *Device) ScalePlayback(dialog sip.Request, speed float64) {
+	infoRequest := CreateRequestFromDialog(dialog, sip.INFO)
+	sn := GetSN()
+	body := fmt.Sprintf(RTSPBodyFormat, sn, speed)
+	infoRequest.SetBody(body, true)
+	infoRequest.RemoveHeader("Content-Type")
+	infoRequest.AppendHeader(&RTSPMessageType)
+	infoRequest.RemoveHeader("Contact")
+	infoRequest.AppendHeader(GlobalContactAddress.AsContactHeader())
+
+	// 替換到device的真實地址
+	recipient := infoRequest.Recipient()
+	if uri, ok := recipient.(*sip.SipUri); ok {
+		sipPort := sip.Port(d.RemotePort)
+		uri.FHost = d.RemoteIP
+		uri.FPort = &sipPort
+	}
+
+	common.SipStack.SendRequest(infoRequest)
+}