package streamer import ( "bufio" "context" "fmt" "io" "live-streamer/config" "log" "os" "os/exec" "path/filepath" "strings" "sync" "time" ) type Streamer struct { videoList []config.InputItem currentVideoIndex int cmd *exec.Cmd logFile *os.File ctx context.Context cancel context.CancelFunc mu sync.Mutex } func NewStreamer(videoList []config.InputItem) *Streamer { logDir := "logs" if err := os.MkdirAll(logDir, 0755); err != nil { log.Printf("Error creating log directory: %v\n", err) } logPath := filepath.Join(logDir, fmt.Sprintf("ffmpeg_%s.log", time.Now().Format("2006-01-02_15-04-05"))) logFile, err := os.OpenFile(logPath, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0644) if err != nil { log.Fatalf("Error opening log file: %v\n", err) } return &Streamer{ videoList: videoList, currentVideoIndex: 0, cmd: nil, logFile: logFile, ctx: nil, } } func (s *Streamer) Add(videoPath string) { s.videoList = append(s.videoList, config.InputItem{Path: videoPath}) } func (s *Streamer) Remove(videoPath string) { for i, item := range s.videoList { if item.Path == videoPath { s.videoList = append(s.videoList[:i], s.videoList[i+1:]...) if s.currentVideoIndex >= len(s.videoList) { s.currentVideoIndex = 0 } if s.currentVideoIndex == i { s.Stop() } break } } } func (s *Streamer) Prev() { s.currentVideoIndex-- if s.currentVideoIndex < 0 { s.currentVideoIndex = len(s.videoList) - 1 } s.Stop() } func (s *Streamer) Next() { s.currentVideoIndex++ if s.currentVideoIndex >= len(s.videoList) { s.currentVideoIndex = 0 } s.Stop() } func (s *Streamer) Stream() { for { if len(s.videoList) == 0 { time.Sleep(time.Second) continue } s.start() } } func (s *Streamer) buildFFmpegArgs(videoItem config.InputItem) []string { videoPath := videoItem.Path args := []string{"-re"} if videoItem.Start != "" { args = append(args, "-ss", videoItem.Start) } args = append(args, "-i", videoPath) if videoItem.End != "" { args = append(args, "-to", videoItem.End) } args = append(args, "-c:v", config.GlobalConfig.Play.VideoCodec, "-preset", config.GlobalConfig.Play.Preset, "-crf", fmt.Sprintf("%d", config.GlobalConfig.Play.CRF), "-maxrate", config.GlobalConfig.Play.MaxRate, "-bufsize", config.GlobalConfig.Play.BufSize, "-vf", fmt.Sprintf("scale=%s", config.GlobalConfig.Play.Scale), "-r", fmt.Sprintf("%d", config.GlobalConfig.Play.FrameRate), "-c:a", config.GlobalConfig.Play.AudioCodec, "-b:a", config.GlobalConfig.Play.AudioBitrate, "-ar", fmt.Sprintf("%d", config.GlobalConfig.Play.AudioSampleRate), "-f", config.GlobalConfig.Play.OutputFormat, "-stats", "-loglevel", "info", ) if config.GlobalConfig.Play.CustomArgs != "" { customArgs := strings.Fields(config.GlobalConfig.Play.CustomArgs) args = append(args, customArgs...) } args = append(args, fmt.Sprintf("%s/%s", config.GlobalConfig.Output.RTMPServer, config.GlobalConfig.Output.StreamKey)) // log.Println("ffmpeg args: ", args) return args } func (s *Streamer) start() { s.Stop() s.ctx, s.cancel = context.WithCancel(context.Background()) currentVideo := s.videoList[s.currentVideoIndex] videoPath := currentVideo.Path log.Println("start stream: ", videoPath) s.mu.Lock() s.cmd = exec.CommandContext(s.ctx, "ffmpeg", s.buildFFmpegArgs(currentVideo)...) s.mu.Unlock() pipe, err := s.cmd.StderrPipe() if err != nil { log.Printf("failed to get pipe: %v", err) return } reader := bufio.NewReader(pipe) writer := bufio.NewWriter(s.logFile) if err := s.cmd.Start(); err != nil { log.Printf("starting ffmpeg error: %v\n", err) return } go s.log(reader, writer) <-s.ctx.Done() log.Printf("stop stream: %s", videoPath) if currentVideo == s.videoList[s.currentVideoIndex] { s.Next() } } func (s *Streamer) Stop() { if s.cancel != nil { done := make(chan error) go func() { done <- s.cmd.Wait() }() s.cancel() s.mu.Lock() if s.cmd != nil && s.cmd.Process != nil { select { case <-done: break case <-time.After(3 * time.Second): // log.Printf("ffmpeg process is still running, killing it...\n") if !s.cmd.ProcessState.Exited() { _ = s.cmd.Process.Kill() } break } s.cmd = nil } s.mu.Unlock() } } func (s *Streamer) GetCurrentVideo() string { return s.videoList[s.currentVideoIndex].Path } func (s *Streamer) GetVideoList() []config.InputItem { return s.videoList } func (s *Streamer) GetVideoListPath() []string { var videoList []string for _, item := range s.videoList { videoList = append(videoList, item.Path) } return videoList } func (s *Streamer) GetCurrentIndex() int { return s.currentVideoIndex } func (s *Streamer) Close() { if s.logFile != nil { s.logFile.Close() s.logFile = nil } s.Stop() } func (s *Streamer) log(reader *bufio.Reader, writer *bufio.Writer) { defer func() { if s.logFile != nil { if err := s.logFile.Sync(); err != nil { log.Printf("syncing log file error: %v\n", err) } } }() buf := make([]byte, 1024) for { n, err := reader.Read(buf) if err != nil { if err != io.EOF { log.Printf("reading ffmpeg error: %v\n", err) } break } if n > 0 { timestamp := time.Now().Format("2006-01-02 15:04:05") logLine := fmt.Sprintf("[%s] %s", timestamp, string(buf[:n])) if s.logFile != nil { if _, err := writer.WriteString(logLine); err != nil { log.Printf("writing to log file error: %v\n", err) } if err := writer.Flush(); err != nil { log.Printf("flushing writer error: %v\n", err) } } } } }