Merge pull request livepeer#22 from livepeer/videoPlayerFix

Video player fix

cmd/rtmp_segment_packer/main.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"flag"
 	"io"
+	"net/url"
 	"strings"
 
 	"time"
@@ -265,17 +266,17 @@ func main() {
 	flag.Set("logtostderr", "true")
 	flag.Parse()
 
-	lpms := lpms.New("1935", "8000", "2435", "7935", "")
+	lpms := lpms.New("1935", "8000", "", "")
 	streamDB := &StreamDB{db: make(map[string]stream.Stream)}
 
 	lpms.HandleRTMPPublish(
 		//getStreamID
-		func(reqPath string) (string, error) {
-			return getStreamIDFromPath(reqPath), nil
+		func(url *url.URL) (string, error) {
+			return getStreamIDFromPath(url.Path), nil
 		},
 		//getStream
-		func(reqPath string) (stream.Stream, stream.Stream, error) {
-			streamID := getStreamIDFromPath(reqPath)
+		func(url *url.URL) (stream.Stream, stream.Stream, error) {
+			streamID := getStreamIDFromPath(url.Path)
 			stream1 := NewSegmentStream(streamID)
 			stream2 := NewSegmentStream(streamID)
 			streamDB.db[streamID] = stream1

cmd/runner/main.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"flag"
 	"net/http"
+	"net/url"
 	"strings"
 	"time"
 
@@ -26,18 +27,18 @@ func main() {
 	flag.Set("logtostderr", "true")
 	flag.Parse()
 
-	lpms := lpms.New("1935", "8000", "2435", "7935", "")
+	lpms := lpms.New("1935", "8000", "", "")
 	streamDB := &StreamDB{db: make(map[string]stream.Stream)}
 	bufferDB := &BufferDB{db: make(map[string]*stream.HLSBuffer)}
 
 	lpms.HandleRTMPPublish(
 		//getStreamID
-		func(reqPath string) (string, error) {
-			return getStreamIDFromPath(reqPath), nil
+		func(url *url.URL) (string, error) {
+			return getStreamIDFromPath(url.Path), nil
 		},
 		//getStream
-		func(reqPath string) (stream.Stream, stream.Stream, error) {
-			rtmpStreamID := getStreamIDFromPath(reqPath)
+		func(url *url.URL) (stream.Stream, stream.Stream, error) {
+			rtmpStreamID := getStreamIDFromPath(url.Path)
 			hlsStreamID := rtmpStreamID + "_hls"
 			rtmpStream := stream.NewVideoStream(rtmpStreamID, stream.RTMP)
 			hlsStream := stream.NewVideoStream(hlsStreamID, stream.HLS)

lpms.go

@@ -7,6 +7,7 @@ import (
 	"context"
 	"encoding/json"
 	"net/http"
+	"net/url"
 
 	"github.com/golang/glog"
 	"github.com/livepeer/lpms/stream"
@@ -37,11 +38,11 @@ type transcodeReq struct {
 }
 
 //New creates a new LPMS server object.  It really just brokers everything to the components.
-func New(rtmpPort string, httpPort string, srsRTMPPort string, srsHTTPPort string, ffmpegPath string) *LPMS {
+func New(rtmpPort, httpPort, ffmpegPath, vodPath string) *LPMS {
 	server := &joy4rtmp.Server{Addr: (":" + rtmpPort)}
-	player := &vidplayer.VidPlayer{RtmpServer: server}
+	player := &vidplayer.VidPlayer{RtmpServer: server, VodPath: vodPath}
 	listener := &vidlistener.VidListener{RtmpServer: server, FfmpegPath: ffmpegPath}
-	return &LPMS{rtmpServer: server, vidPlayer: player, vidListen: listener, srsRTMPPort: srsRTMPPort, srsHTTPPort: srsHTTPPort, httpPort: httpPort, ffmpegPath: ffmpegPath}
+	return &LPMS{rtmpServer: server, vidPlayer: player, vidListen: listener, httpPort: httpPort, ffmpegPath: ffmpegPath}
 }
 
 //Start starts the rtmp and http server
@@ -65,8 +66,8 @@ func (l *LPMS) Start() error {
 
 //HandleRTMPPublish offload to the video listener
 func (l *LPMS) HandleRTMPPublish(
-	getStreamID func(reqPath string) (string, error),
-	getStream func(reqPath string) (stream.Stream, stream.Stream, error),
+	getStreamID func(url *url.URL) (string, error),
+	getStream func(url *url.URL) (stream.Stream, stream.Stream, error),
 	endStream func(rtmpStrmID string, hlsStrmID string)) error {
 
 	return l.vidListen.HandleRTMPPublish(getStreamID, getStream, endStream)

segmenter/video_segmenter.go

@@ -8,6 +8,7 @@ import (
 	"io/ioutil"
 	"os"
 	"os/exec"
+	"path/filepath"
 	"strconv"
 	"time"
 
@@ -64,7 +65,7 @@ func NewFFMpegVideoSegmenter(workDir string, strmID string, localRtmpUrl string,
 }
 
 //RTMPToHLS invokes the FFMpeg command to do the segmenting.  This method blocks unless killed.
-func (s *FFMpegVideoSegmenter) RTMPToHLS(ctx context.Context, opt SegmenterOptions) error {
+func (s *FFMpegVideoSegmenter) RTMPToHLS(ctx context.Context, opt SegmenterOptions, cleanup bool) error {
 	//Set up local workdir
 	if _, err := os.Stat(s.WorkDir); os.IsNotExist(err) {
 		err := os.Mkdir(s.WorkDir, 0700)
@@ -91,7 +92,7 @@ func (s *FFMpegVideoSegmenter) RTMPToHLS(ctx context.Context, opt SegmenterOptio
 
 	var cmd *exec.Cmd
 
-	cmd = exec.Command(path.Join(s.ffmpegPath, "ffmpeg"), "-i", s.LocalRtmpUrl, "-vcodec", "copy", "-acodec", "copy", "-bsf:v", "h264_mp4toannexb", "-f", "segment", "-muxdelay", "0", "-segment_list", plfn, tsfn)
+	cmd = exec.Command(path.Join(s.ffmpegPath, "ffmpeg"), "-i", s.LocalRtmpUrl, "-vcodec", "copy", "-acodec", "copy", "-bsf:v", "h264_mp4toannexb", "-f", "segment", "-segment_time", "8", "-muxdelay", "0", "-segment_list", plfn, tsfn)
 
 	err = cmd.Start()
 	if err != nil {
@@ -109,6 +110,10 @@ func (s *FFMpegVideoSegmenter) RTMPToHLS(ctx context.Context, opt SegmenterOptio
 	case <-ctx.Done():
 		//Can't close RTMP server, joy4 doesn't support it.
 		//server.Stop()
+		glog.Infof("VideoSegmenter stopped for %v", s.StrmID)
+		if cleanup {
+			s.Cleanup()
+		}
 		cmd.Process.Kill()
 		return ctx.Err()
 	}
@@ -129,7 +134,8 @@ func (s *FFMpegVideoSegmenter) PollSegment(ctx context.Context) (*VideoSegment,
 
 	for i := 0; i < PlaylistRetryCount; i++ {
 		pl, _ := m3u8.NewMediaPlaylist(uint(s.curSegment+1), uint(s.curSegment+1))
-		pl.DecodeFrom(bytes.NewReader(readPlaylist(plfn)), true)
+		content := readPlaylist(plfn)
+		pl.DecodeFrom(bytes.NewReader(content), true)
 		for _, plSeg := range pl.Segments {
 			if plSeg.URI == name {
 				length, err = time.ParseDuration(fmt.Sprintf("%vs", plSeg.Duration))
@@ -260,3 +266,10 @@ func (s *FFMpegVideoSegmenter) pollSegment(ctx context.Context, curFn string, ne
 		s.curSegWaitTime = s.curSegWaitTime + sleepTime
 	}
 }
+
+func (s *FFMpegVideoSegmenter) Cleanup() {
+	files, _ := filepath.Glob(path.Join(s.WorkDir, s.StrmID) + "*")
+	for _, fn := range files {
+		os.Remove(fn)
+	}
+}

segmenter/video_segmenter_test.go

@@ -82,7 +82,7 @@ func TestSegmenter(t *testing.T) {
 	defer cancel()
 
 	//Kick off FFMpeg to create segments
-	go func() { se <- func() error { return vs.RTMPToHLS(ctx, opt) }() }()
+	go func() { se <- func() error { return vs.RTMPToHLS(ctx, opt, false) }() }()
 	select {
 	case err := <-se:
 		if err != context.DeadlineExceeded {
@@ -128,7 +128,8 @@ func TestSegmenter(t *testing.T) {
 			t.Errorf("Expecting HLS segment, got %v", seg.Format)
 		}
 
-		if seg.Length != time.Second*2 {
+		timeDiff := seg.Length - time.Second*2
+		if timeDiff > time.Millisecond*500 || timeDiff < -time.Millisecond*500 {
 			t.Errorf("Expecting 2 sec segments, got %v", seg.Length)
 		}
 
@@ -253,8 +254,19 @@ func TestPollSegTimeout(t *testing.T) {
 	os.RemoveAll(workDir)
 	os.Mkdir(workDir, 0700)
 
+	newPl := `#EXTM3U
+#EXT-X-VERSION:3
+#EXT-X-MEDIA-SEQUENCE:0
+#EXT-X-ALLOW-CACHE:YES
+#EXT-X-TARGETDURATION:7
+#EXTINF:2.066000,
+test_0.ts
+#EXTINF:2.066000,
+test_1.ts
+`
+	err := ioutil.WriteFile(workDir+"/test.m3u8", []byte(newPl), 0755)
 	newSeg := `some random data`
-	err := ioutil.WriteFile(workDir+"/test_0.ts", []byte(newSeg), 0755)
+	err = ioutil.WriteFile(workDir+"/test_0.ts", []byte(newSeg), 0755)
 	err = ioutil.WriteFile(workDir+"/test_1.ts", []byte(newSeg), 0755)
 	if err != nil {
 		t.Errorf("Error writing playlist: %v", err)
@@ -272,4 +284,5 @@ func TestPollSegTimeout(t *testing.T) {
 		t.Errorf("Expecting timeout, got %v", err)
 	}
 
+	os.RemoveAll(workDir)
 }

stream/stream_subscriber.go

@@ -131,6 +131,22 @@ func (s *StreamSubscriber) UnsubscribeHLS(muxID string) error {
 	return nil
 }
 
+func (s *StreamSubscriber) UnsubscribeAll() error {
+	if s.hlsSubscribers != nil {
+		for k := range s.hlsSubscribers {
+			delete(s.hlsSubscribers, k)
+		}
+	}
+
+	if s.rtmpSubscribers != nil {
+		for k := range s.rtmpSubscribers {
+			delete(s.rtmpSubscribers, k)
+		}
+	}
+
+	return nil
+}
+
 func (s *StreamSubscriber) StartHLSWorker(ctx context.Context, segWaitTime time.Duration) error {
 	lastSegTimer := time.Now()
 	for {
@@ -170,15 +186,15 @@ func (s *StreamSubscriber) GetRTMPBuffer(subID string) av.Muxer {
 	return s.rtmpSubscribers[subID]
 }
 
-func (s *StreamSubscriber) HLSSubscribers() []string {
+func (s *StreamSubscriber) HLSSubscribersReport() []string {
 	var res []string
 	for sub, v := range s.hlsSubscribers {
 		res = append(res, fmt.Sprintf("%v: %v", reflect.TypeOf(v), sub))
 	}
 	return res
 }
 
-func (s *StreamSubscriber) RTMPSubscribers() []string {
+func (s *StreamSubscriber) RTMPSubscribersReport() []string {
 	var res []string
 	for sub, v := range s.rtmpSubscribers {
 		res = append(res, fmt.Sprintf("%v: %v", reflect.TypeOf(v), sub))

vidlistener/listener.go

@@ -2,6 +2,7 @@ package vidlistener
 
 import (
 	"context"
+	"net/url"
 	"os"
 	"time"
 
@@ -26,20 +27,20 @@ type VidListener struct {
 //HandleRTMPPublish immediately turns the RTMP stream into segmented HLS, and writes it into a stream.
 //It exposes getStreamID so the user can name the stream, and getStream so the user can keep track of all the streams.
 func (self *VidListener) HandleRTMPPublish(
-	getStreamID func(reqPath string) (string, error),
-	getStream func(reqPath string) (rtmpStrm stream.Stream, hlsStrm stream.Stream, err error),
+	getStreamID func(url *url.URL) (string, error),
+	getStream func(url *url.URL) (rtmpStrm stream.Stream, hlsStrm stream.Stream, err error),
 	endStream func(rtmpStrmID string, hlsStrmID string)) error {
 
 	self.RtmpServer.HandlePublish = func(conn *joy4rtmp.Conn) {
-		glog.Infof("RTMP server got upstream")
+		glog.Infof("RTMP server got upstream: %v", conn.URL)
 
-		_, err := getStreamID(conn.URL.Path)
+		_, err := getStreamID(conn.URL)
 		if err != nil {
 			glog.Errorf("RTMP Stream Publish Error: %v", err)
 			return
 		}
 
-		rs, hs, err := getStream(conn.URL.Path)
+		rs, hs, err := getStream(conn.URL)
 		if err != nil {
 			glog.Errorf("RTMP Publish couldn't get a destination stream for %v", conn.URL.Path)
 			return
@@ -75,7 +76,7 @@ func (self *VidListener) segmentStream(ctx context.Context, rs stream.Stream, hs
 	localRtmpUrl := "rtmp://localhost" + self.RtmpServer.Addr + "/stream/" + rs.GetStreamID()
 	s := segmenter.NewFFMpegVideoSegmenter(workDir, hs.GetStreamID(), localRtmpUrl, segOptions.SegLength, self.FfmpegPath)
 	c := make(chan error, 1)
-	go func() { c <- s.RTMPToHLS(ctx, segOptions) }()
+	go func() { c <- s.RTMPToHLS(ctx, segOptions, true) }()
 
 	go func() {
 		c <- func() error {

vidlistener/listener_test.go

@@ -2,6 +2,7 @@ package vidlistener
 
 import (
 	"fmt"
+	"net/url"
 	"os/exec"
 	"testing"
 	"time"
@@ -14,10 +15,10 @@ func TestListener(t *testing.T) {
 	server := &joy4rtmp.Server{Addr: ":1937"}
 	listener := &VidListener{RtmpServer: server}
 	listener.HandleRTMPPublish(
-		func(reqPath string) (string, error) {
+		func(url *url.URL) (string, error) {
 			return "test", nil
 		},
-		func(reqPath string) (stream.Stream, stream.Stream, error) {
+		func(url *url.URL) (stream.Stream, stream.Stream, error) {
 			// return errors.New("Some Error")
 			return stream.NewVideoStream("test", stream.RTMP), stream.NewVideoStream("test", stream.HLS), nil
 		},