New logic to write .insit files.

writer/internal/ingest/rtsp/rtsp.go

@@ -20,6 +20,7 @@ import (
 	"github.com/bluenviron/gortsplib/v4/pkg/base"
 	"github.com/bluenviron/gortsplib/v4/pkg/description"
 	"github.com/bluenviron/gortsplib/v4/pkg/format"
+	"github.com/bluenviron/mediacommon/pkg/codecs/h264"
 	"github.com/bluenviron/mediacommon/v2/pkg/codecs/g711"
 	"github.com/pion/rtp"
 	"go.uber.org/zap"
@@ -49,7 +50,10 @@ func StartWriter() {
 
 // rtsp processes RTSP protocol.
 func rtsp(dir string, period int, link string, number int) error {
-	// Увеличиваем счетчик при старте обработки камеры
+
+	var marker bool
+
+	// Calculate the number of active cameras (number of active rtsp functions).
 	metrics.ActiveCameras.Inc()
 	defer metrics.ActiveCameras.Dec()
 
@@ -599,6 +603,25 @@ func rtsp(dir string, period int, link string, number int) error {
 					cam.Warn("process packet error:", zap.Error(err))
 				}
 
+				for _, nalu := range au {
+					typ := h264.NALUType(nalu[0] & 0x1F)
+
+					switch typ {
+					case h264.NALUTypeSPS:
+						continue
+
+					case h264.NALUTypePPS:
+						continue
+
+					case h264.NALUTypeAccessUnitDelimiter:
+						continue
+
+					case h264.NALUTypeIDR:
+						marker = true
+						time.Sleep(time.Millisecond)
+					}
+				}
+
 				if au != nil {
 					// Add appropriate lines to the interleaved packet.
 					seg.Packet.Type = storage.PacketTypeH264
@@ -612,6 +635,8 @@ func rtsp(dir string, period int, link string, number int) error {
 					}
 				}
 
+				marker = false
+
 				//// Encode the access unit into MPEG-TS.
 				//err = currentMpegtsMuxer.WriteH264(pts, au)
 				//if err != nil {
@@ -687,55 +712,56 @@ func rtsp(dir string, period int, link string, number int) error {
 		// Rotate files.
 		go func() {
 			for range ticker.C {
-				/*
+				fileChanged := false
-					// Logic for rotation files.
+				for {
-					currentMpegtsMuxer.Close()
+					if marker {
-					currentMpegtsMuxer.FileName = fn.SetNumNTime()
+						file.Close()
 
-					err = currentMpegtsMuxer.Initialize()
+						file, err = os.Create(fn.SetNumNTime("insit"))
-					if err != nil {
+						if err != nil {
-						log.Printf("[%v-%v]: init muxer error: %v\n", videoFormat, audioFormat, err)
+							cam.Error("creating file error:", zap.Error(err))
-						return
+							logger.Log.Error("creating file error:", zap.Error(err))
+							return
+						}
+
+						seg = storage.Segment{
+							Date:     strconv.FormatInt(time.Now().Unix(), 10),
+							Duration: strconv.Itoa(period),
+							Packet:   storage.InterleavedPacket{},
+						}
+
+						// Write StreamID.
+						if err := binary.Write(file, binary.LittleEndian, int32(len(cutURI))); err != nil {
+							cam.Error("write StreamID length error:", zap.Error(err))
+							logger.Log.Error("write StreamID length error:", zap.Error(err))
+							return
+						}
+						if _, err := file.Write([]byte(cutURI)); err != nil {
+							cam.Error("write StreamID error:", zap.Error(err))
+							logger.Log.Error("write StreamID error:", zap.Error(err))
+							return
+						}
+
+						// Write header of the file.
+						err = storage.WriteHeader(file, seg)
+						if err != nil {
+							cam.Error("write header error:", zap.Error(err))
+							logger.Log.Error("write header error:", zap.Error(err))
+							return
+						}
+
+						cam.Info("new file for recording created")
+						log.Println("new file for recording created")
+
+						fileChanged = true
+						break
 					}
-				*/
 
-				file.Close()
+					if fileChanged {
-
+						fileChanged = false
-				file, err = os.Create(fn.SetNumNTime("insit"))
+						break
-				if err != nil {
+					}
-					cam.Error("creating file error:", zap.Error(err))
-					logger.Log.Error("creating file error:", zap.Error(err))
-					return
 				}
-
-				seg = storage.Segment{
-					Date:     strconv.FormatInt(time.Now().Unix(), 10),
-					Duration: strconv.Itoa(period),
-					Packet:   storage.InterleavedPacket{},
-				}
-
-				// Write StreamID.
-				if err := binary.Write(file, binary.LittleEndian, int32(len(cutURI))); err != nil {
-					cam.Error("write StreamID length error:", zap.Error(err))
-					logger.Log.Error("write StreamID length error:", zap.Error(err))
-					return
-				}
-				if _, err := file.Write([]byte(cutURI)); err != nil {
-					cam.Error("write StreamID error:", zap.Error(err))
-					logger.Log.Error("write StreamID error:", zap.Error(err))
-					return
-				}
-
-				// Write header of the file.
-				err = storage.WriteHeader(file, seg)
-				if err != nil {
-					cam.Error("write header error:", zap.Error(err))
-					logger.Log.Error("write header error:", zap.Error(err))
-					return
-				}
-
-				cam.Info("new file for recording created")
-				log.Println("new file for recording created")
 			}
 		}()