history video push to server play

docs/init/rtsp2rtmp-postgresql.sql

@@ -123,3 +123,38 @@ COMMENT ON COLUMN public.camera_share.created IS '创建时间';
 COMMENT ON COLUMN public.camera_share.start_time IS '开始时间';
 COMMENT ON COLUMN public.camera_share.deadline IS '结束时间';
 COMMENT ON COLUMN public.camera_share.camera_id IS '摄像头id';
+
+-- public.camera_record definition
+
+-- Drop table
+
+-- DROP TABLE camera_record;
+
+CREATE TABLE camera_record (
+	id_camera_record varchar(255) NOT NULL,
+	created timestamp NULL, -- 创建时间
+	temp_file_name varchar(255) NULL, -- 临时文件名称
+	fg_temp bool NULL, -- 临时文件标志
+	file_name varchar(255) NULL, -- 文件名称
+	fg_remove bool NULL, -- 文件删除标志
+	duration int4 NULL, -- 文件时长: 单位：毫秒
+	start_time timestamp NULL, -- 开始时间
+	end_time timestamp NULL, -- 结束时间
+	id_camera varchar(255) NOT NULL, -- 摄像头主属性
+	has_audio bool NULL, -- 是否有音频
+	CONSTRAINT pk_camera_record PRIMARY KEY (id_camera_record)
+);
+COMMENT ON TABLE public.camera_record IS '摄像头记录';
+
+-- Column comments
+
+COMMENT ON COLUMN public.camera_record.created IS '创建时间';
+COMMENT ON COLUMN public.camera_record.temp_file_name IS '临时文件名称';
+COMMENT ON COLUMN public.camera_record.fg_temp IS '临时文件标志';
+COMMENT ON COLUMN public.camera_record.file_name IS '文件名称';
+COMMENT ON COLUMN public.camera_record.fg_remove IS '文件删除标志';
+COMMENT ON COLUMN public.camera_record.duration IS '文件时长: 单位：毫秒';
+COMMENT ON COLUMN public.camera_record.start_time IS '开始时间';
+COMMENT ON COLUMN public.camera_record.end_time IS '结束时间';
+COMMENT ON COLUMN public.camera_record.id_camera IS '摄像头主属性';
+COMMENT ON COLUMN public.camera_record.has_audio IS '是否有音频';

resources/conf/conf-prod.yml

@@ -21,9 +21,9 @@ server:
         duration: 1440
     remote:
         server-ip: 127.0.0.1
-        port: 8089
+        port: 9089
-        client-id: demo
+        client-code: demo
-        sign-secret: 123456
+        sign-secret: A012345678901234
-        secret: 123456
+        secret: A012345678901234
     
         

src/rtsp2rtmp/tcpclient/camera_aq.go

@@ -9,16 +9,17 @@ import (
 	base_service "github.com/hkmadao/rtsp2rtmp/src/rtsp2rtmp/web/service/base"
 )
 
-func cameraAq(paramStr string) {
+func cameraAq(commandMessage CommandMessage) {
+	paramStr := commandMessage.Param
 	condition := common.AqCondition{}
 	err := json.Unmarshal([]byte(paramStr), &condition)
 	if err != nil {
-		logs.Error("flvFileMediaInfo message format error: %v", err)
+		logs.Error("cameraAq message format error: %v", err)
 		return
 	}
-	conn, err := connectAndRegister("historyVideoPage")
+	conn, err := connectAndRegister("cameraAq", commandMessage.MessageId)
 	if err != nil {
-		logs.Error("historyVideoPage connect to server error: %v", err)
+		logs.Error("cameraAq connect to server error: %v", err)
 		return
 	}
 

src/rtsp2rtmp/tcpclient/command_receive.go

@@ -1,8 +1,6 @@
 package tcpclient
 
 import (
-	"bytes"
-	"encoding/binary"
 	"encoding/json"
 	"time"
 
@@ -15,13 +13,13 @@ func StartCommandReceiveServer() {
 	go func() {
 		for {
 			commandReceiveConnect()
-			<-time.NewTicker(1 * time.Minute).C
+			<-time.NewTicker(10 * time.Second).C
 		}
 	}()
 }
 
 func commandReceiveConnect() {
-	conn, err := connectAndRegister("keepChannel")
+	conn, err := connectAndRegister("keepChannel", "")
 	if err != nil {
 		logs.Error("keepChannel connect to server error: %v", err)
 		return
@@ -35,14 +33,8 @@ func commandReceiveConnect() {
 			logs.Error("read len error: %v", err)
 			break
 		}
-		var dataLen int32
+		dataLen := utils.BigEndianToUint32(dataLenBytes)
-		buffer := bytes.NewBuffer(dataLenBytes)
+
-		err = binary.Read(buffer, binary.LittleEndian, &dataLen)
-		if err != nil {
-			logs.Error(err)
-			break
-		}
-		logs.Info("receive message dataLen: %d", dataLen)
 		serverRepBytes := make([]byte, dataLen)
 		_, err = conn.Read(serverRepBytes)
 		if err != nil {
@@ -50,7 +42,7 @@ func commandReceiveConnect() {
 			break
 		}
 		secretCommandStr := string(serverRepBytes)
-		logs.Info("receive message: %s", secretCommandStr)
+
 		secretStr, err := config.String("server.remote.secret")
 		if err != nil {
 			logs.Error("get remote secret error: %v", err)
@@ -68,19 +60,24 @@ func commandReceiveConnect() {
 			continue
 		}
 
+		// do response
+		go commandRes(commandMessage)
+	}
+}
+
+func commandRes(commandMessage CommandMessage) {
 	switch commandMessage.MessageType {
 	case "cameraAq":
-			cameraAq(commandMessage.Param)
+		cameraAq(commandMessage)
 	case "historyVideoPage":
-			historyVideoPage(commandMessage.Param)
+		historyVideoPage(commandMessage)
 	case "flvFileMediaInfo":
-			flvFileMediaInfo(commandMessage.Param)
+		flvFileMediaInfo(commandMessage)
 	case "flvPlay":
-			flvPlay(commandMessage.Param)
+		flvPlay(commandMessage)
 	case "flvFetchMoreData":
-			flvFetchMoreData(commandMessage.Param)
+		flvFetchMoreData(commandMessage)
 	default:
 		logs.Error("unsupport commandType: %s", commandMessage.MessageType)
 	}
 }
-}

src/rtsp2rtmp/tcpclient/common.go

@@ -17,22 +17,24 @@ type CommandMessage struct {
 	// "cameraAq" "historyVideoPage" "flvFileMediaInfo" "flvPlay" "flvFetchMoreData" "startPushRtmp" "stopPushRtmp"
 	MessageType string `json:"messageType"`
 	Param       string `json:"param"`
+	MessageId   string `json:"messageId"`
 }
 
 // when connect to server, first send register packet to server
 type RegisterInfo struct {
-	ClientId string `json:"clientId"`
+	ClientCode string `json:"clientCode"`
 	DateStr    string `json:"dateStr"`
 	Sign       string `json:"sign"`
 	// "keepChannel" "cameraAq" "historyVideoPage" "flvFileMediaInfo" "flvPlay" "flvFetchMoreData" "startPushRtmp" "stopPushRtmp"
 	ConnType  string `json:"connType"`
+	MessageId string `json:"messageId"`
 }
 
-func newReisterInfo(connType string) (ri RegisterInfo, err error) {
+func newReisterInfo(connType string, messageId string) (ri RegisterInfo, err error) {
 	currentDateStr := time.Now().Format(time.RFC3339)
-	clientId, err := config.String("server.remote.client-id")
+	clientCode, err := config.String("server.remote.client-code")
 	if err != nil {
-		logs.Error("get remote client-id error: %v\n", err)
+		logs.Error("get remote client-code error: %v\n", err)
 		return
 	}
 	signSecret, err := config.String("server.remote.sign-secret")
@@ -40,7 +42,7 @@ func newReisterInfo(connType string) (ri RegisterInfo, err error) {
 		logs.Error("get remote sign-secret error: %v\n", err)
 		return
 	}
-	planText := fmt.Sprintf("cilentId=%s&dateStr=%s", clientId, currentDateStr)
+	planText := fmt.Sprintf("clientCode=%s&dateStr=%s", clientCode, currentDateStr)
 	signStr, err := utils.EncryptAES([]byte(signSecret), planText)
 	if err != nil {
 		err = fmt.Errorf("buildSign error: %v", err)
@@ -48,15 +50,16 @@ func newReisterInfo(connType string) (ri RegisterInfo, err error) {
 	}
 
 	ri = RegisterInfo{
-		ClientId: clientId,
+		ClientCode: clientCode,
 		ConnType:   connType,
 		DateStr:    currentDateStr,
 		Sign:       signStr,
+		MessageId:  messageId,
 	}
 	return
 }
 
-func connectAndRegister(connType string) (conn net.Conn, err error) {
+func connectAndRegister(connType string, messageId string) (conn net.Conn, err error) {
 	serverIp, err := config.String("server.remote.server-ip")
 	if err != nil {
 		logs.Error("get remote server-ip error: %v. \n", err)
@@ -72,12 +75,9 @@ func connectAndRegister(connType string) (conn net.Conn, err error) {
 		logs.Error(err)
 		return
 	}
-	defer func() {
-		conn.Close()
-	}()
 
 	// register to server
-	ri, err := newReisterInfo(connType)
+	ri, err := newReisterInfo(connType, messageId)
 	if err != nil {
 		logs.Error(err)
 		return

src/rtsp2rtmp/tcpclient/flv_data_push.go

@@ -51,16 +51,17 @@ func (flvPush FlvPush) Write(p []byte) (n int, err error) {
 	return len(p), nil
 }
 
-func flvPlay(paramStr string) {
+func flvPlay(commandMessage CommandMessage) {
+	paramStr := commandMessage.Param
 	playParam := PlayParam{}
 	err := json.Unmarshal([]byte(paramStr), &playParam)
 	if err != nil {
 		logs.Error("flvPlay message format error: %v", err)
 		return
 	}
-	conn, err := connectAndRegister("flvFileMediaInfo")
+	conn, err := connectAndRegister("flvPlay", commandMessage.MessageId)
 	if err != nil {
-		logs.Error("flvFileMediaInfo connect to server error: %v", err)
+		logs.Error("flvPlay connect to server error: %v", err)
 		return
 	}
 	camera_record, err := base_service.CameraRecordSelectById(playParam.IdCameraRecord)
@@ -79,7 +80,9 @@ func flvPlay(paramStr string) {
 		fileName = camera_record.TempFileName
 	}
 
-	ffr := fileflvreader.NewFileFlvReader(playParam.SeekSecond, conn, fileName)
+	flvPush := FlvPush{conn: conn}
+
+	ffr := fileflvreader.NewFileFlvReader(playParam.SeekSecond, flvPush, fileName)
 	_, ok := playerMap.Load(playParam.PlayerId)
 	if ok {
 		logs.Error("playerId: %s exists", playParam.PlayerId)

src/rtsp2rtmp/tcpclient/flv_fetch_more_data.go

@@ -14,14 +14,15 @@ type FetchMoreDataParam struct {
 	SeekSecond uint64 `json:"seekSecond"`
 }
 
-func flvFetchMoreData(paramStr string) {
+func flvFetchMoreData(commandMessage CommandMessage) {
+	paramStr := commandMessage.Param
 	param := FetchMoreDataParam{}
 	err := json.Unmarshal([]byte(paramStr), &param)
 	if err != nil {
-		logs.Error("flvPlay message format error: %v", err)
+		logs.Error("flvFetchMoreData message format error: %v", err)
 		return
 	}
-	conn, err := connectAndRegister("flvFetchMoreData")
+	conn, err := connectAndRegister("flvFetchMoreData", commandMessage.MessageId)
 	if err != nil {
 		logs.Error("flvFetchMoreData connect to server error: %v", err)
 		return

src/rtsp2rtmp/tcpclient/flv_media_info.go

@@ -15,7 +15,8 @@ type FlvFileMediaInfoParam struct {
 	IdCameraRecord string `json:"idCameraRecord"`
 }
 
-func flvFileMediaInfo(paramStr string) {
+func flvFileMediaInfo(commandMessage CommandMessage) {
+	paramStr := commandMessage.Param
 	param := FlvFileMediaInfoParam{}
 	err := json.Unmarshal([]byte(paramStr), &param)
 	if err != nil {
@@ -23,7 +24,7 @@ func flvFileMediaInfo(paramStr string) {
 		return
 	}
 	idCameraRecord := param.IdCameraRecord
-	conn, err := connectAndRegister("flvFileMediaInfo")
+	conn, err := connectAndRegister("flvFileMediaInfo", commandMessage.MessageId)
 	if err != nil {
 		logs.Error("flvFileMediaInfo connect to server error: %v", err)
 		return

src/rtsp2rtmp/tcpclient/history_video_page.go

@@ -11,14 +11,15 @@ import (
 	base_service "github.com/hkmadao/rtsp2rtmp/src/rtsp2rtmp/web/service/base"
 )
 
-func historyVideoPage(paramStr string) {
+func historyVideoPage(commandMessage CommandMessage) {
+	paramStr := commandMessage.Param
 	pageInfoInput := common.AqPageInfoInput{}
 	err := json.Unmarshal([]byte(paramStr), &pageInfoInput)
 	if err != nil {
-		logs.Error("flvFileMediaInfo message format error: %v", err)
+		logs.Error("historyVideoPage message format error: %v", err)
 		return
 	}
-	conn, err := connectAndRegister("historyVideoPage")
+	conn, err := connectAndRegister("historyVideoPage", commandMessage.MessageId)
 	if err != nil {
 		logs.Error("historyVideoPage connect to server error: %v", err)
 		return

src/rtsp2rtmp/utils/util.go

@@ -40,13 +40,18 @@ func Float64ToByteBigEndian(float float64) []byte {
 func Int32ToByteBigEndian(number int32) []byte {
 	bytes := make([]byte, 4)
 	bytes[0] = byte(number >> (3 * 8))
-	bytes[0] = byte(number >> (2 * 8))
+	bytes[1] = byte(number >> (2 * 8))
-	bytes[0] = byte(number >> (1 * 8))
+	bytes[2] = byte(number >> (1 * 8))
-	bytes[0] = byte(number)
+	bytes[3] = byte(number)
 
 	return bytes
 }
 
+func BigEndianToUint32(bytes []byte) (dataLen uint32) {
+	dataLen = binary.BigEndian.Uint32(bytes)
+	return
+}
+
 func ByteToFloat64(bytes []byte) float64 {
 	bits := binary.LittleEndian.Uint64(bytes)