Fixed leak of goroutine and channel in Video.cleanup()

go.mod

@@ -1,3 +1,7 @@
 module github.com/AlexEidt/Vidio
 
-go 1.16
+go 1.21.1
+
+toolchain go1.21.4
+
+require github.com/nla-is/datax-go/v2 v2.0.0-alpha.5 // indirect

go.sum

@@ -0,0 +1,10 @@
+github.com/ebitengine/purego v0.5.0 h1:JrMGKfRIAM4/QVKaesIIT7m/UVjTj5GYhRSQYwfVdpo=
+github.com/ebitengine/purego v0.5.0/go.mod h1:ah1In8AOtksoNK6yk5z1HTJeUkC1Ez4Wk2idgGslMwQ=
+github.com/fxamacker/cbor/v2 v2.5.0 h1:oHsG0V/Q6E/wqTS2O1Cozzsy69nqCiguo5Q1a1ADivE=
+github.com/fxamacker/cbor/v2 v2.5.0/go.mod h1:TA1xS00nchWmaBnEIxPSE5oHLuJBAVvqrtAnWBwBCVo=
+github.com/nla-is/datax-go/v2 v2.0.0-alpha.5 h1:eTzwuQdlbta+YWIcrIM69JuaGMCfUMzl7OEmP0SnrRU=
+github.com/nla-is/datax-go/v2 v2.0.0-alpha.5/go.mod h1:PcxU+Rc92jHGwoddnqIxQDXF1XYWTLgIAxhvIfFzQzM=
+github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM=
+github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg=
+golang.org/x/sys v0.7.0 h1:3jlCCIQZPdOYu1h8BkNvLz8Kgwtae2cagcG/VamtZRU=
+golang.org/x/sys v0.7.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=

utils.go

@@ -43,6 +43,7 @@ func ffprobe(filename, stype string) ([]map[string]string, error) {
 	// Extract video information with ffprobe.
 	cmd := exec.Command(
 		"ffprobe",
+		"-rtsp_transport", "tcp",
 		"-show_streams",
 		"-select_streams", stype,
 		"-print_format", "compact",

video.go

@@ -27,6 +27,9 @@ type Video struct {
 	metadata    map[string]string // Video metadata.
 	pipe        io.ReadCloser     // Stdout pipe for ffmpeg process.
 	cmd         *exec.Cmd         // ffmpeg command.
+
+	closeCleanupChan chan struct{} // exit from cleanup goroutine to avoid chan and goroutine leak
+	cleanupClosed    bool
 }
 
 func (video *Video) FileName() string {
@@ -107,9 +110,15 @@ func NewVideo(filename string) (*Video, error) {
 	return streams[0], err
 }
 
+func isStream(url string) bool {
+	return strings.HasPrefix(url, "http://") ||
+		strings.HasPrefix(url, "https://") ||
+		strings.HasPrefix(url, "rtsp://")
+}
+
 // Read all video streams from the given file.
 func NewVideoStreams(filename string) ([]*Video, error) {
-	if !exists(filename) {
+	if !isStream(filename) && !exists(filename) {
 		return nil, fmt.Errorf("vidio: video file %s does not exist", filename)
 	}
 	// Check if ffmpeg and ffprobe are installed on the users machine.
@@ -150,6 +159,8 @@ func NewVideoStreams(filename string) ([]*Video, error) {
 			stream:     i,
 			hasstreams: hasstream,
 			metadata:   data,
+
+			closeCleanupChan: make(chan struct{}, 1),
 		}
 
 		video.addVideoData(data)
@@ -199,6 +210,7 @@ func (video *Video) init() error {
 	// ffmpeg command to pipe video data to stdout in 8-bit RGBA format.
 	cmd := exec.Command(
 		"ffmpeg",
+		"-rtsp_transport", "tcp",
 		"-i", video.filename,
 		"-f", "image2pipe",
 		"-loglevel", "quiet",
@@ -384,6 +396,11 @@ func (video *Video) ReadFrames(n ...int) ([]*image.RGBA, error) {
 
 // Closes the pipe and stops the ffmpeg process.
 func (video *Video) Close() {
+	if !video.cleanupClosed {
+		video.cleanupClosed = true
+		video.closeCleanupChan <- struct{}{}
+		close(video.closeCleanupChan)
+	}
 	if video.pipe != nil {
 		video.pipe.Close()
 	}
@@ -398,13 +415,18 @@ func (video *Video) cleanup() {
 	c := make(chan os.Signal, 1)
 	signal.Notify(c, os.Interrupt, syscall.SIGTERM)
 	go func() {
-		<-c
-		if video.pipe != nil {
-			video.pipe.Close()
-		}
-		if video.cmd != nil {
-			video.cmd.Process.Kill()
+		select {
+		case <-c:
+			if video.pipe != nil {
+				video.pipe.Close()
+			}
+			if video.cmd != nil {
+				video.cmd.Process.Kill()
+			}
+			os.Exit(1)
+		case <-video.closeCleanupChan:
+			signal.Stop(c)
+			close(c)
 		}
-		os.Exit(1)
 	}()
 }