<commit_before>//The RTMP server.  This will put up a RTMP endpoint when starting up Swarm.
//To integrate with LPMS means your code will become the source / destination of the media server.
//This RTMP endpoint is mainly used for video upload.  The expected url is rtmp://localhost:port/livepeer/stream
package core

import (
	"context"
	"net/http"
	"net/url"

	"github.com/ericxtang/m3u8"
	"github.com/golang/glog"
	"github.com/livepeer/lpms/ffmpeg"
	"github.com/livepeer/lpms/segmenter"
	"github.com/livepeer/lpms/stream"
	"github.com/livepeer/lpms/vidlistener"
	"github.com/livepeer/lpms/vidplayer"

	joy4rtmp "github.com/nareix/joy4/format/rtmp"
)

type LPMS struct {
	rtmpServer *joy4rtmp.Server
	vidPlayer  *vidplayer.VidPlayer
	vidListen  *vidlistener.VidListener
	workDir    string
	httpAddr   string
}

type transcodeReq struct {
	Formats  []string
	Bitrates []string
	Codecin  string
	Codecout []string
	StreamID string
}

type LPMSOpts struct {
	RtmpPort     string
	RtmpDisabled bool
	RtmpHost     string
	HttpPort     string
	HttpDisabled bool
	HttpHost     string
	VodPath      string
	WorkDir      string
}

func defaultLPMSOpts(opts *LPMSOpts) {
	if opts.RtmpPort == "" {
		opts.RtmpPort = "1935"
	}
	if opts.HttpPort == "" {
		opts.HttpPort = "8935"
	}
}

//New creates a new LPMS server object.  It really just brokers everything to the components.
func New(opts *LPMSOpts) *LPMS {
	defaultLPMSOpts(opts)
	var rtmpServer *joy4rtmp.Server
	if !opts.RtmpDisabled {
		rtmpServer = &joy4rtmp.Server{Addr: (opts.RtmpHost + ":" + opts.RtmpPort)}
	}
	var httpAddr string
	if !opts.HttpDisabled {
		httpAddr = opts.HttpHost + ":" + opts.HttpPort
	}
	player := vidplayer.NewVidPlayer(rtmpServer, opts.VodPath)
	listener := &vidlistener.VidListener{RtmpServer: rtmpServer}
	return &LPMS{rtmpServer: rtmpServer, vidPlayer: player, vidListen: listener, workDir: opts.WorkDir, httpAddr: httpAddr}
}

//Start starts the rtmp and http server
func (l *LPMS) Start(ctx context.Context) error {
	ec := make(chan error, 1)
	ffmpeg.InitFFmpeg()
	defer ffmpeg.DeinitFFmpeg()
	if l.rtmpServer != nil {
		go func() {
			glog.V(4).Infof("LPMS Server listening on rtmp://%v", l.rtmpServer.Addr)
			ec <- l.rtmpServer.ListenAndServe()
		}()
	}
	startHTTP := l.httpAddr != ""
	if startHTTP {
		go func() {
			glog.V(4).Infof("HTTP Server listening on http://%v", l.httpAddr)
			ec <- http.ListenAndServe(l.httpAddr, nil)
		}()
	}

	if l.rtmpServer != nil || startHTTP {
		select {
		case err := <-ec:
			glog.Errorf("LPMS Server Error: %v.  Quitting...", err)
			return err
		case <-ctx.Done():
			return ctx.Err()
		}
	}

	return nil
}

//HandleRTMPPublish offload to the video listener
func (l *LPMS) HandleRTMPPublish(
	makeStreamID func(url *url.URL) (strmID string),
	gotStream func(url *url.URL, rtmpStrm stream.RTMPVideoStream) (err error),
	endStream func(url *url.URL, rtmpStrm stream.RTMPVideoStream) error) {

	l.vidListen.HandleRTMPPublish(makeStreamID, gotStream, endStream)
}

//HandleRTMPPlay offload to the video player
func (l *LPMS) HandleRTMPPlay(getStream func(url *url.URL) (stream.RTMPVideoStream, error)) error {
	return l.vidPlayer.HandleRTMPPlay(getStream)
}

//HandleHLSPlay offload to the video player
func (l *LPMS) HandleHLSPlay(
	getMasterPlaylist func(url *url.URL) (*m3u8.MasterPlaylist, error),
	getMediaPlaylist func(url *url.URL) (*m3u8.MediaPlaylist, error),
	getSegment func(url *url.URL) ([]byte, error)) {

	l.vidPlayer.HandleHLSPlay(getMasterPlaylist, getMediaPlaylist, getSegment)
}

//SegmentRTMPToHLS takes a rtmp stream and re-packages it into a HLS stream with the specified segmenter options
func (l *LPMS) SegmentRTMPToHLS(ctx context.Context, rs stream.RTMPVideoStream, hs stream.HLSVideoStream, segOptions segmenter.SegmenterOptions) error {
	//Invoke Segmenter
	localRtmpUrl := "rtmp://localhost" + l.rtmpServer.Addr + "/stream/" + rs.GetStreamID()
	glog.V(4).Infof("Segment RTMP Req: %v", localRtmpUrl)

	s := segmenter.NewFFMpegVideoSegmenter(l.workDir, hs.GetStreamID(), localRtmpUrl, segOptions)<commit_msg>vendor: Update LPMS to ef2615b.<commit_after>//The RTMP server.  This will put up a RTMP endpoint when starting up Swarm.
//To integrate with LPMS means your code will become the source / destination of the media server.
//This RTMP endpoint is mainly used for video upload.  The expected url is rtmp://localhost:port/livepeer/stream
package core

import (
	"context"
	"net/http"
	"net/url"
	"strings"

	"github.com/ericxtang/m3u8"
	"github.com/golang/glog"
	"github.com/livepeer/lpms/ffmpeg"
	"github.com/livepeer/lpms/segmenter"
	"github.com/livepeer/lpms/stream"
	"github.com/livepeer/lpms/vidlistener"
	"github.com/livepeer/lpms/vidplayer"

	joy4rtmp "github.com/nareix/joy4/format/rtmp"
)

type LPMS struct {
	rtmpServer *joy4rtmp.Server
	vidPlayer  *vidplayer.VidPlayer
	vidListen  *vidlistener.VidListener
	workDir    string
	httpAddr   string
}

type transcodeReq struct {
	Formats  []string
	Bitrates []string
	Codecin  string
	Codecout []string
	StreamID string
}

type LPMSOpts struct {
	RtmpPort     string
	RtmpDisabled bool
	RtmpHost     string
	HttpPort     string
	HttpDisabled bool
	HttpHost     string
	VodPath      string
	WorkDir      string
}

func defaultLPMSOpts(opts *LPMSOpts) {
	if opts.RtmpPort == "" {
		opts.RtmpPort = "1935"
	}
	if opts.HttpPort == "" {
		opts.HttpPort = "8935"
	}
}

//New creates a new LPMS server object.  It really just brokers everything to the components.
func New(opts *LPMSOpts) *LPMS {
	defaultLPMSOpts(opts)
	var rtmpServer *joy4rtmp.Server
	if !opts.RtmpDisabled {
		rtmpServer = &joy4rtmp.Server{Addr: (opts.RtmpHost + ":" + opts.RtmpPort)}
	}
	var httpAddr string
	if !opts.HttpDisabled {
		httpAddr = opts.HttpHost + ":" + opts.HttpPort
	}
	player := vidplayer.NewVidPlayer(rtmpServer, opts.VodPath)
	listener := &vidlistener.VidListener{RtmpServer: rtmpServer}
	return &LPMS{rtmpServer: rtmpServer, vidPlayer: player, vidListen: listener, workDir: opts.WorkDir, httpAddr: httpAddr}
}

//Start starts the rtmp and http server
func (l *LPMS) Start(ctx context.Context) error {
	ec := make(chan error, 1)
	ffmpeg.InitFFmpeg()
	defer ffmpeg.DeinitFFmpeg()
	if l.rtmpServer != nil {
		go func() {
			glog.V(4).Infof("LPMS Server listening on rtmp://%v", l.rtmpServer.Addr)
			ec <- l.rtmpServer.ListenAndServe()
		}()
	}
	startHTTP := l.httpAddr != ""
	if startHTTP {
		go func() {
			glog.V(4).Infof("HTTP Server listening on http://%v", l.httpAddr)
			ec <- http.ListenAndServe(l.httpAddr, nil)
		}()
	}

	if l.rtmpServer != nil || startHTTP {
		select {
		case err := <-ec:
			glog.Errorf("LPMS Server Error: %v.  Quitting...", err)
			return err
		case <-ctx.Done():
			return ctx.Err()
		}
	}

	return nil
}

//HandleRTMPPublish offload to the video listener
func (l *LPMS) HandleRTMPPublish(
	makeStreamID func(url *url.URL) (strmID string),
	gotStream func(url *url.URL, rtmpStrm stream.RTMPVideoStream) (err error),
	endStream func(url *url.URL, rtmpStrm stream.RTMPVideoStream) error) {

	l.vidListen.HandleRTMPPublish(makeStreamID, gotStream, endStream)
}

//HandleRTMPPlay offload to the video player
func (l *LPMS) HandleRTMPPlay(getStream func(url *url.URL) (stream.RTMPVideoStream, error)) error {
	return l.vidPlayer.HandleRTMPPlay(getStream)
}

//HandleHLSPlay offload to the video player
func (l *LPMS) HandleHLSPlay(
	getMasterPlaylist func(url *url.URL) (*m3u8.MasterPlaylist, error),
	getMediaPlaylist func(url *url.URL) (*m3u8.MediaPlaylist, error),
	getSegment func(url *url.URL) ([]byte, error)) {

	l.vidPlayer.HandleHLSPlay(getMasterPlaylist, getMediaPlaylist, getSegment)
}

//SegmentRTMPToHLS takes a rtmp stream and re-packages it into a HLS stream with the specified segmenter options
func (l *LPMS) SegmentRTMPToHLS(ctx context.Context, rs stream.RTMPVideoStream, hs stream.HLSVideoStream, segOptions segmenter.SegmenterOptions) error {
	// set localhost if necessary. Check more problematic addrs? [::] ?
	rtmpAddr := l.rtmpServer.Addr
	if strings.HasPrefix(rtmpAddr, "0.0.0.0") {
		rtmpAddr = "127.0.0.1" + rtmpAddr[len("0.0.0.0"):]
	}
	localRtmpUrl := "rtmp://" + rtmpAddr + "/stream/" + rs.GetStreamID()

	glog.V(4).Infof("Segment RTMP Req: %v", localRtmpUrl)

	//Invoke Segmenter
	s := segmenter.NewFFMpegVideoSegmenter(l.workDir, hs.GetStreamID(), localRtmpUrl, segOptions)
{'fraction_non_alphanumeric': 0.12056737588652482, 'fraction_numerical': 0.007693232359658613, 'mean_word_length': 4.347043701799486, 'pattern_counts': {'":': 4, '<': 11, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 3, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4033096', 'n_tokens_mistral': 3126, 'n_tokens_neox': 3026, 'n_words': 742}