Added logger.

writer/cmd/main.go

@@ -1,35 +1,20 @@
 package main
 
 import (
-	"flag"
 	"fmt"
+	logger "git.insit.tech/psa/rtsp_reader-writer/writer/internal/log"
+	"time"
+
 	"git.insit.tech/psa/rtsp_reader-writer/writer/internal/config"
 	"git.insit.tech/psa/rtsp_reader-writer/writer/internal/ingest/rtsp"
-	"git.insit.tech/sas/rtsp_proxy/core/log"
+	log2 "git.insit.tech/sas/rtsp_proxy/core/log"
 )
 
 func main() {
-	directory := flag.String("dir", "/home/psa/GoRepository", "directory")
+	config.LogsDirectory = log2.DirCreator(config.Local, "logs")
-	flag.Parse()
+	logger.Log = log2.MainLogging(
+		fmt.Sprintf("%s/Main_%s.log", config.LogsDirectory, time.Now().Format("15-04-05_02-01-2006")))
 
-	// Parse camera links from YAML file into struct Cameras.
+	rtsp.StartWriter()
-	cams, err := config.ParseCamerasYAML(*directory)
-	if err != nil {
-		panic(err)
-	}
-
-	Log := log.MainLogging("/home/psa/GoRepository/" + "/data/" + "camera54-centr-kirova-kalinina")
-
-	// Connect to each camera.
-	for _, link := range cams {
-		fmt.Printf("process camera:\n %s\n", link)
-
-		go func() {
-			err = rtsp.StartWriter(*directory, 60, link, Log)
-			if err != nil {
-				fmt.Printf("procRTSP function error for camera %s: %s", link, err.Error())
-			}
-		}()
-	}
 	select {}
 }

writer/go.sum

@@ -24,8 +24,6 @@ github.com/golang/snappy v1.0.0 h1:Oy607GVXHs7RtbggtPBnr2RmDArIsAefDwvrdWvRhGs=
 github.com/golang/snappy v1.0.0/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
 github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
 github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
-github.com/grafov/m3u8 v0.12.1 h1:DuP1uA1kvRRmGNAZ0m+ObLv1dvrfNO0TPx0c/enNk0s=
-github.com/grafov/m3u8 v0.12.1/go.mod h1:nqzOkfBiZJENr52zTVd/Dcl03yzphIMbJqkXGu+u080=
 github.com/pion/randutil v0.1.0 h1:CFG1UdESneORglEsnimhUjf33Rwjubwj6xfiOXBa3mA=
 github.com/pion/randutil v0.1.0/go.mod h1:XcJrSMMbbMRhASFVOlj/5hQial/Y8oH/HVo7TBZq+j8=
 github.com/pion/rtcp v1.2.15 h1:LZQi2JbdipLOj4eBjK4wlVoQWfrZbh3Q6eHtWtJBZBo=
@@ -48,8 +46,6 @@ github.com/youpy/go-wav v0.3.2 h1:NLM8L/7yZ0Bntadw/0h95OyUsen+DQIVf9gay+SUsMU=
 github.com/youpy/go-wav v0.3.2/go.mod h1:0FCieAXAeSdcxFfwLpRuEo0PFmAoc+8NU34h7TUvk50=
 github.com/zaf/g711 v1.4.0 h1:XZYkjjiAg9QTBnHqEg37m2I9q3IIDv5JRYXs2N8ma7c=
 github.com/zaf/g711 v1.4.0/go.mod h1:eCDXt3dSp/kYYAoooba7ukD/Q75jvAaS4WOMr0l1Roo=
-github.com/zencoder/go-dash v0.0.0-20201006100653-2f93b14912b2 h1:0iAY2pL6yYhNYpdc1DbFq0p7ocyu5MlgKmkealhz3nk=
-github.com/zencoder/go-dash v0.0.0-20201006100653-2f93b14912b2/go.mod h1:c8Gxxfmh0jmZ6G+ISlpa315WBVkzd8mEhu6gN9mn5Qg=
 go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
 go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
 go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0=

writer/internal/config/config.go

@@ -0,0 +1,7 @@
+package config
+
+var (
+	Local         = "reader-writer"
+	LogsDirectory string
+	Cameras       = "/home/psa/GoRepository/rtsp_reader-writer/writer/internal/config/cameras.yaml"
+)

writer/internal/config/parser.go

@@ -6,10 +6,10 @@ import (
 )
 
 // ParseCamerasYAML parses camera links from YAML file into struct Cameras.
-func ParseCamerasYAML(dir string) (map[string]string, error) {
+func ParseCamerasYAML() (map[string]string, error) {
 	var CamerasYAML map[string]string
 
-	data, err := os.ReadFile(dir + "/rtsp_reader-writer/writer/internal/config/source.yaml")
+	data, err := os.ReadFile(Cameras)
 	if err != nil {
 		return CamerasYAML, err
 	}

writer/internal/ingest/formats/aac.go

@@ -27,7 +27,6 @@ func ProcessAAC(
 	aacMedia *description.Media,
 	aacRTPDec *rtpmpeg4audio.Decoder,
 	pkt *rtp.Packet,
-	t string,
 ) (
 	int64,
 	[][]byte,
@@ -35,13 +34,13 @@ func ProcessAAC(
 	// Decode timestamp.
 	pts, ok := c.PacketPTS2(aacMedia, pkt)
 	if !ok {
-		return 0, nil, fmt.Errorf("[%v]: waiting for timestamp\n", t)
+		return 0, nil, fmt.Errorf("waiting for timestamp\n")
 	}
 
 	// Extract access unit from RTP packets.
 	aus, err := aacRTPDec.Decode(pkt)
 	if err != nil {
-		return 0, nil, fmt.Errorf("[%v]: decoding RTP packet error: %w", t, err)
+		return 0, nil, fmt.Errorf("decoding RTP packet error: %w", err)
 	}
 
 	return pts, aus, nil

writer/internal/ingest/formats/av1.go

@@ -32,7 +32,6 @@ func ProcessAV1(
 	pkt *rtp.Packet,
 	av1Dec *AV1Decoder,
 	firstRandomReceived bool,
-	t string,
 ) (
 	int64,
 	*image.RGBA,
@@ -40,31 +39,31 @@ func ProcessAV1(
 	// Decode timestamp.
 	pts, ok := c.PacketPTS2(av1Media, pkt)
 	if !ok {
-		return 0, nil, fmt.Errorf("[%v]: waiting for timestamp\n", t)
+		return 0, nil, fmt.Errorf("waiting for timestamp\n")
 	}
 
 	// Extract AV1 temporal units from RTP packets.
 	tu, err := av1RTPDec.Decode(pkt)
 	if err != rtpav1.ErrNonStartingPacketAndNoPrevious && err != rtpav1.ErrMorePacketsNeeded {
-		return 0, nil, fmt.Errorf("[%v]: decoding RTP packet error: %w", t, err)
+		return 0, nil, fmt.Errorf("decoding RTP packet error: %w", err)
 	}
 
 	// Wait for a random access unit.
 	IsRandomAccess2, _ := av1.IsRandomAccess(tu)
 	if !firstRandomReceived && !IsRandomAccess2 {
-		return 0, nil, fmt.Errorf("[%v]: waiting for a random access unit\n", t)
+		return 0, nil, fmt.Errorf("waiting for a random access unit\n")
 	}
 	firstRandomReceived = true
 
 	// Convert AV1 temporal units into RGBA frames.
 	img, err := av1Dec.Decode(tu)
 	if err != nil {
-		return 0, nil, fmt.Errorf("[%v]: convert into RGBA frames error\n", t)
+		return 0, nil, fmt.Errorf("convert into RGBA frames error\n")
 	}
 
 	// Wait for a frame.
 	if img == nil {
-		return 0, nil, fmt.Errorf("[%v]: frame not found\n", t)
+		return 0, nil, fmt.Errorf("frame not found\n")
 	}
 
 	return pts, img, nil

writer/internal/ingest/formats/g711.go

@@ -22,19 +22,19 @@ func FindG711Format(desc *description.Session) (*format.G711, *description.Media
 }
 
 // ProcessG711 processes G711 flow and returns PTS and AU.
-func ProcessG711(c *gortsplib.Client, g711Media *description.Media, g711RTPDec *rtplpcm.Decoder, pkt *rtp.Packet, t string,
+func ProcessG711(c *gortsplib.Client, g711Media *description.Media, g711RTPDec *rtplpcm.Decoder, pkt *rtp.Packet,
 ) (
 	int64, []byte, error) {
 	// Decode timestamp.
 	pts, ok := c.PacketPTS2(g711Media, pkt)
 	if !ok {
-		return 0, nil, fmt.Errorf("[%v]: waiting for timestamp\n", t)
+		return 0, nil, fmt.Errorf("waiting for timestamp\n")
 	}
 
 	// Extract access unit from RTP packets.
 	au, err := g711RTPDec.Decode(pkt)
 	if err != nil {
-		return 0, nil, fmt.Errorf("[%v]: decoding RTP packet error: %w", t, err)
+		return 0, nil, fmt.Errorf("decoding RTP packet error: %w", err)
 	}
 
 	return pts, au, nil

writer/internal/ingest/formats/h264.go

@@ -28,7 +28,6 @@ func ProcessH264(
 	h264Media *description.Media,
 	h264RTPDec *rtph264.Decoder,
 	pkt *rtp.Packet,
-	t string,
 ) (
 	int64,
 	[][]byte,
@@ -36,14 +35,14 @@ func ProcessH264(
 	// Decode timestamp.
 	pts, ok := c.PacketPTS2(h264Media, pkt)
 	if !ok {
-		return 0, nil, fmt.Errorf("[%v]: waiting for timestamp\n", t)
+		return 0, nil, fmt.Errorf("waiting for timestamp\n")
 	}
 
 	// Extract access unit from RTP packets.
 	au, err := h264RTPDec.Decode(pkt)
 	if err != nil {
 		if err != rtph264.ErrNonStartingPacketAndNoPrevious && err != rtph264.ErrMorePacketsNeeded {
-			return 0, nil, fmt.Errorf("[%v]: decoding RTP packet error: %w", t, err)
+			return 0, nil, fmt.Errorf("decoding RTP packet error: %w", err)
 		}
 	}
 
@@ -58,7 +57,6 @@ func ProcessH264RGBA(
 	h264Dec *H264Decoder,
 	pkt *rtp.Packet,
 	firstRandomAccess bool,
-	t string,
 ) (
 	int64,
 	*image.RGBA,
@@ -66,13 +64,13 @@ func ProcessH264RGBA(
 	// Decode timestamp.
 	pts, ok := c.PacketPTS2(h264Media, pkt)
 	if !ok {
-		return 0, nil, fmt.Errorf("[%v]: waiting for timestamp\n", t)
+		return 0, nil, fmt.Errorf("waiting for timestamp\n")
 	}
 
 	// Extract access units from RTP packets.
 	au, err := h264RTPDec.Decode(pkt)
 	if err != rtph264.ErrNonStartingPacketAndNoPrevious && err != rtph264.ErrMorePacketsNeeded {
-		return 0, nil, fmt.Errorf("[%v]: decoding RTP packet error: %w", t, err)
+		return 0, nil, fmt.Errorf("decoding RTP packet error: %w", err)
 	}
 
 	// Wait for a random access unit.
@@ -84,12 +82,12 @@ func ProcessH264RGBA(
 	// Convert H264 access units into RGBA frames.
 	img, err := h264Dec.Decode(au)
 	if err != nil {
-		return 0, nil, fmt.Errorf("[%v]: convert into RGBA frames error\n", t)
+		return 0, nil, fmt.Errorf("convert into RGBA frames error\n")
 	}
 
 	// Wait for a frame.
 	if img == nil {
-		return 0, nil, fmt.Errorf("[%v]: frame not found\n", t)
+		return 0, nil, fmt.Errorf("frame not found\n")
 	}
 
 	return pts, img, nil

writer/internal/ingest/formats/h265.go

@@ -31,7 +31,6 @@ func ProcessH265RGBA(
 	h265Dec *H265Decoder,
 	pkt *rtp.Packet,
 	firstRandomAccess bool,
-	t string,
 ) (
 	int64,
 	*image.RGBA,
@@ -39,13 +38,13 @@ func ProcessH265RGBA(
 	// Decode timestamp.
 	pts, ok := c.PacketPTS2(h265Media, pkt)
 	if !ok {
-		return 0, nil, fmt.Errorf("[%v]: waiting for timestamp\n", t)
+		return 0, nil, fmt.Errorf("waiting for timestamp\n")
 	}
 
 	// Extract access units from RTP packets.
 	au, err := h265RTPDec.Decode(pkt)
 	if err != rtph265.ErrNonStartingPacketAndNoPrevious && err != rtph265.ErrMorePacketsNeeded {
-		return 0, nil, fmt.Errorf("[%v]: decoding RTP packet error: %w", t, err)
+		return 0, nil, fmt.Errorf("decoding RTP packet error: %w", err)
 	}
 
 	// Wait for a random access unit.
@@ -57,12 +56,12 @@ func ProcessH265RGBA(
 	// Convert H265 access units into RGBA frames.
 	img, err := h265Dec.Decode(au)
 	if err != nil {
-		return 0, nil, fmt.Errorf("[%v]: convert into RGBA frames error\n", t)
+		return 0, nil, fmt.Errorf("convert into RGBA frames error\n")
 	}
 
 	// Wait for a frame.
 	if img == nil {
-		return 0, nil, fmt.Errorf("[%v]: frame not found\n", t)
+		return 0, nil, fmt.Errorf("frame not found\n")
 	}
 
 	return pts, img, nil

writer/internal/ingest/formats/lpcm.go

@@ -27,7 +27,6 @@ func ProcessLPCM(
 	lpcmMedia *description.Media,
 	lpcmRTPDec *rtplpcm.Decoder,
 	pkt *rtp.Packet,
-	t string,
 ) (
 	int64,
 	[]byte,
@@ -35,13 +34,13 @@ func ProcessLPCM(
 	// Decode timestamp.
 	pts, ok := c.PacketPTS2(lpcmMedia, pkt)
 	if !ok {
-		return 0, nil, fmt.Errorf("[%v]: waiting for timestamp\n", t)
+		return 0, nil, fmt.Errorf("waiting for timestamp\n")
 	}
 
 	// Extract LPCM samples from RTP packets.
 	samples, err := lpcmRTPDec.Decode(pkt)
 	if err != nil {
-		return 0, nil, fmt.Errorf("[%v]: decoding RTP packet error: %w", t, err)
+		return 0, nil, fmt.Errorf("decoding RTP packet error: %w", err)
 	}
 
 	return pts, samples, err

writer/internal/ingest/formats/mjpeg.go

@@ -30,7 +30,6 @@ func ProcessMJPEGRGBA(
 	mjpegMedia *description.Media,
 	mjpegRTPDec *rtpmjpeg.Decoder,
 	pkt *rtp.Packet,
-	t string,
 ) (
 	int64,
 	image.Image,
@@ -38,19 +37,19 @@ func ProcessMJPEGRGBA(
 	// Decode timestamp.
 	pts, ok := c.PacketPTS2(mjpegMedia, pkt)
 	if !ok {
-		return 0, nil, fmt.Errorf("[%v]: waiting for timestamp\n", t)
+		return 0, nil, fmt.Errorf("waiting for timestamp\n")
 	}
 
 	// Extract JPEG images from RTP packets.
 	enc, err := mjpegRTPDec.Decode(pkt)
 	if err != rtpmjpeg.ErrNonStartingPacketAndNoPrevious && err != rtpmjpeg.ErrMorePacketsNeeded {
-		return 0, nil, fmt.Errorf("[%v]: decoding RTP packet error: %w", t, err)
+		return 0, nil, fmt.Errorf("decoding RTP packet error: %w", err)
 	}
 
 	// Convert JPEG images into RGBA frames.
 	img, err := jpeg.Decode(bytes.NewReader(enc))
 	if err != nil {
-		return 0, nil, fmt.Errorf("[%v]: convert into RGBA frames error\n", t)
+		return 0, nil, fmt.Errorf("convert into RGBA frames error\n")
 	}
 
 	return pts, img, nil

writer/internal/ingest/formats/opus.go

@@ -27,7 +27,6 @@ func ProcessOPUS(
 	opusMedia *description.Media,
 	opusRTPDec *rtpsimpleaudio.Decoder,
 	pkt *rtp.Packet,
-	t string,
 ) (
 	int64,
 	[]byte,
@@ -35,13 +34,13 @@ func ProcessOPUS(
 	// Decode timestamp.
 	pts, ok := c.PacketPTS2(opusMedia, pkt)
 	if !ok {
-		return 0, nil, fmt.Errorf("[%v]: waiting for timestamp\n", t)
+		return 0, nil, fmt.Errorf("waiting for timestamp\n")
 	}
 
 	// Extract Opus packets from RTP packets.
 	op, err := opusRTPDec.Decode(pkt)
 	if err != nil {
-		return 0, nil, fmt.Errorf("[%v]: decoding RTP packet error: %w", t, err)
+		return 0, nil, fmt.Errorf("decoding RTP packet error: %w", err)
 	}
 
 	return pts, op, nil

writer/internal/ingest/formats/vp8.go

@@ -29,7 +29,6 @@ func ProcessVP8RGBA(
 	vp8RTPDec *rtpvp8.Decoder,
 	vp8Dec *VPDecoder,
 	pkt *rtp.Packet,
-	t string,
 ) (
 	int64,
 	*image.RGBA,
@@ -37,24 +36,24 @@ func ProcessVP8RGBA(
 	// Decode timestamp.
 	pts, ok := c.PacketPTS2(vp8Media, pkt)
 	if !ok {
-		return 0, nil, fmt.Errorf("[%v]: waiting for timestamp\n", t)
+		return 0, nil, fmt.Errorf("waiting for timestamp\n")
 	}
 
 	// Extract access units from RTP packets.
 	au, err := vp8RTPDec.Decode(pkt)
 	if err != rtpvp8.ErrNonStartingPacketAndNoPrevious && err != rtpvp8.ErrMorePacketsNeeded {
-		return 0, nil, fmt.Errorf("[%v]: decoding RTP packet error: %w", t, err)
+		return 0, nil, fmt.Errorf("decoding RTP packet error: %w", err)
 	}
 
 	// Convert VP8 access units into RGBA frames.
 	img, err := vp8Dec.Decode(au)
 	if err != nil {
-		return 0, nil, fmt.Errorf("[%v]: convert into RGBA frames error\n", t)
+		return 0, nil, fmt.Errorf("convert into RGBA frames error\n")
 	}
 
 	// Wait for a frame.
 	if img == nil {
-		return 0, nil, fmt.Errorf("[%v]: frame not found\n", t)
+		return 0, nil, fmt.Errorf("frame not found\n")
 	}
 
 	return pts, img, nil

writer/internal/ingest/formats/vp9.go

@@ -30,7 +30,6 @@ func ProcessVP9RGBA(
 	vp9RTPDec *rtpvp9.Decoder,
 	vp9Dec *VPDecoder,
 	pkt *rtp.Packet,
-	t string,
 ) (
 	int64,
 	*image.RGBA,
@@ -38,24 +37,24 @@ func ProcessVP9RGBA(
 	// Decode timestamp.
 	pts, ok := c.PacketPTS2(vp9Media, pkt)
 	if !ok {
-		return 0, nil, fmt.Errorf("[%v]: waiting for timestamp\n", t)
+		return 0, nil, fmt.Errorf("waiting for timestamp\n")
 	}
 
 	// Extract access units from RTP packets.
 	au, err := vp9RTPDec.Decode(pkt)
 	if err != rtpvp8.ErrNonStartingPacketAndNoPrevious && err != rtpvp8.ErrMorePacketsNeeded {
-		return 0, nil, fmt.Errorf("[%v]: decoding RTP packet error: %w", t, err)
+		return 0, nil, fmt.Errorf("decoding RTP packet error: %w", err)
 	}
 
 	// Convert VP8 access units into RGBA frames.
 	img, err := vp9Dec.Decode(au)
 	if err != nil {
-		return 0, nil, fmt.Errorf("[%v]: convert into RGBA frames error\n", t)
+		return 0, nil, fmt.Errorf("convert into RGBA frames error\n")
 	}
 
 	// Wait for a frame.
 	if img == nil {
-		return 0, nil, fmt.Errorf("[%v]: frame not found\n", t)
+		return 0, nil, fmt.Errorf("frame not found\n")
 	}
 
 	return pts, img, nil

writer/internal/ingest/rtsp/operator.go

@@ -4,22 +4,25 @@ import (
 	"log"
 	"strings"
 	"time"
+
+	"go.uber.org/zap"
 )
 
-// cutURI returns the last part of the URI after "/".
+// lastPartURI returns the last part of the URI after "/".
-func cutURI(URI string) (CutURI string) {
+func lastPartURI(URI string) (CutURI string) {
-	splitted := strings.Split(URI, "/")
+	split := strings.Split(URI, "/")
-	return splitted[len(splitted)-1]
+	return split[len(split)-1]
 }
 
 // waitPeriod waits for the next period.
-func waitPeriod(period int) {
+func waitPeriod(period int, cam *zap.Logger) {
 	periodTD := time.Duration(period) * time.Second
 
 	now := time.Now()
 	nextSegment := now.Truncate(periodTD).Add(periodTD)
 	waitDuration := nextSegment.Sub(now)
-	log.Printf("waiting for start recording: %v\n", waitDuration)
+	cam.Info("waiting for start recording:", zap.Duration("time", waitDuration))
+	log.Println("waiting for start recording: ", waitDuration)
 	time.Sleep(waitDuration)
 }
 

writer/internal/log/logger.go

@@ -2,4 +2,6 @@ package log
 
 import "go.uber.org/zap"
 
-var Log *zap.Logger
+var (
+	Log *zap.Logger
+)

writer/pkg/storage/file_manager.go

@@ -6,9 +6,9 @@ import (
 )
 
 // CreateFileName creates FileName structure.
-func CreateFileName(dir string, resolutions []string, cuttedURI string, period int) *common.FileName {
+func CreateFileName(dirData string, resolutions []string, cutURI string, period int) *common.FileName {
 	fn := common.FileName{
-		Path:     dir + "/data/" + cuttedURI + "/" + resolutions[0],
+		Path:     dirData + "/" + cutURI + "/" + resolutions[0],
 		TimeNow:  time.Now().Format("15-04-05_02-01-2006"),
 		Name:     "videoFragment",
 		Number:   -1,