实现HLS播放拉流计数

api.go

@@ -359,6 +359,17 @@ func (api *ApiServer) onTS(source string, w http.ResponseWriter, r *http.Request
 		return
 	}
 
+	sid := r.URL.Query().Get(hls.SessionIdKey)
+	var sink stream.Sink
+	if sid != "" {
+		sink = stream.SinkManager.Find(stream.SinkId(sid))
+	}
+	if sink == nil {
+		log.Sugar.Errorf("hls session with id '%s' has expired.", sid)
+		w.WriteHeader(http.StatusForbidden)
+		return
+	}
+
 	index := strings.LastIndex(source, "_")
 	if index < 0 || index == len(source)-1 {
 		w.WriteHeader(http.StatusBadRequest)
@@ -366,19 +377,19 @@ func (api *ApiServer) onTS(source string, w http.ResponseWriter, r *http.Request
 	}
 
 	seq := source[index+1:]
-	sourceId := source[:index]
+	tsPath := stream.AppConfig.Hls.TSPath(sink.SourceId(), seq)
-	tsPath := stream.AppConfig.Hls.TSPath(sourceId, seq)
 	if _, err := os.Stat(tsPath); err != nil {
 		w.WriteHeader(http.StatusNotFound)
 		return
 	}
 
-	//链路复用无法获取http断开回调
+	sink.(*hls.M3U8Sink).RefreshPlayTime()
-	//Hijack需要自行解析http
+	w.Header().Set("Content-Type", "video/MP2T")
 	http.ServeFile(w, r, tsPath)
 }
 
 func (api *ApiServer) onHLS(sourceId string, w http.ResponseWriter, r *http.Request) {
+	log.Sugar.Infof("请求m3u8")
 	if !stream.AppConfig.Hls.Enable {
 		log.Sugar.Warnf("处理hls请求失败 server未开启hls")
 		http.Error(w, "hls disable", http.StatusInternalServerError)
@@ -386,21 +397,36 @@ func (api *ApiServer) onHLS(sourceId string, w http.ResponseWriter, r *http.Requ
 	}
 
 	w.Header().Set("Content-Type", "application/vnd.apple.mpegurl")
-	//m3u8和ts会一直刷新, 每个请求只hook一次.
+	//hls_sid是流媒体服务器让播放端, 携带的会话id, 如果没有携带说明是第一次请求播放.
-	sinkId := api.generateSinkId(r.RemoteAddr)
+	//播放端不要使用hls_sid这个key, 否则会一直拉流失败
+	sid := r.URL.Query().Get(hls.SessionIdKey)
+	if sid == "" {
+		//让播放端携带会话id
+		sid = utils.RandStringBytes(10)
+
+		query := r.URL.Query()
+		query.Add(hls.SessionIdKey, sid)
+		path := fmt.Sprintf("/%s.m3u8?%s", sourceId, query.Encode())
+
+		response := "#EXTM3U\r\n" +
+			"#EXT-X-STREAM-INF:BANDWIDTH=1,AVERAGE-BANDWIDTH=1\r\n" +
+			path + "\r\n"
+		w.Write([]byte(response))
+		return
+	}
+
+	sink := stream.SinkManager.Find(sid)
+	if sink != nil {
+		w.Write([]byte(sink.(*hls.M3U8Sink).GetM3U8String()))
+		return
+	}
 
-	//hook成功后, 如果还没有m3u8文件，等生成m3u8文件
-	//后续直接返回当前m3u8文件
-	if stream.SinkManager.Exist(sinkId) {
-		http.ServeFile(w, r, stream.AppConfig.Hls.M3U8Path(sourceId))
-	} else {
 	context := r.Context()
 	done := make(chan int, 0)
-
+	sink = hls.NewM3U8Sink(sid, sourceId, func(m3u8 []byte) {
-		sink := hls.NewM3U8Sink(sinkId, sourceId, func(m3u8 []byte) {
 		w.Write(m3u8)
 		done <- 0
-		})
+	}, sid)
 
 	sink.SetUrlValues(r.URL.Query())
 	_, state := stream.PreparePlaySink(sink)
@@ -420,7 +446,6 @@ func (api *ApiServer) onHLS(sourceId string, w http.ResponseWriter, r *http.Requ
 		break
 	}
 }
-}
 
 func (api *ApiServer) onRtc(sourceId string, w http.ResponseWriter, r *http.Request) {
 	v := struct {

hls/hls_sink.go

@@ -1,31 +1,70 @@
 package hls
 
 import (
+	"fmt"
+	"github.com/lkmio/lkm/log"
 	"github.com/lkmio/lkm/stream"
+	"strings"
+	"time"
 )
 
-type tsSink struct {
+const (
+	SessionIdKey = "hls_sid"
+)
+
+type M3U8Sink struct {
 	stream.BaseSink
+	cb               func(m3u8 []byte) //生成m3u8文件的发送回调
+	sessionId        string
+	playtime         time.Time
+	playTimer        *time.Timer
+	m3u8StringFormat *string
 }
 
-func NewTSSink(id stream.SinkId, sourceId string) stream.Sink {
+func (s *M3U8Sink) SendM3U8Data(data *string) error {
-	return &tsSink{stream.BaseSink{Id_: id, SourceId_: sourceId, Protocol_: stream.ProtocolHls}}
+	s.m3u8StringFormat = data
-}
+	s.cb([]byte(s.GetM3U8String()))
-
-func (s *tsSink) Input(data []byte) error {
 	return nil
 }
 
-type m3u8Sink struct {
+func (s *M3U8Sink) Start() {
-	stream.BaseSink
+	timeout := time.Duration(stream.AppConfig.IdleTimeout)
-	cb func(m3u8 []byte)
+	if timeout < time.Second {
+		timeout = time.Duration(stream.AppConfig.Hls.Duration) * 2 * 3 * time.Second
 	}
 
-func (s *m3u8Sink) Input(data []byte) error {
+	s.playTimer = time.AfterFunc(timeout, func() {
-	s.cb(data)
+		sub := time.Now().Sub(s.playtime)
-	return nil
+		if sub > timeout {
+			log.Sugar.Errorf("长时间没有拉取TS切片 sink:%d 超时", s.Id_)
+			s.Close()
+			return
 		}
 
-func NewM3U8Sink(id stream.SinkId, sourceId string, cb func(m3u8 []byte)) stream.Sink {
+		s.playTimer.Reset(timeout)
-	return &m3u8Sink{stream.BaseSink{Id_: id, SourceId_: sourceId, Protocol_: stream.ProtocolHls}, cb}
+	})
+}
+
+func (s *M3U8Sink) GetM3U8String() string {
+	param := fmt.Sprintf("?%s=%s", SessionIdKey, s.sessionId)
+	all := strings.ReplaceAll(string(*s.m3u8StringFormat), "%s", param)
+	log.Sugar.Infof("m3u8 list:%s", all)
+	return all
+}
+
+func (s *M3U8Sink) RefreshPlayTime() {
+	s.playtime = time.Now()
+}
+
+func (s *M3U8Sink) Close() {
+	stream.SinkManager.Remove(s.Id_)
+	s.BaseSink.Close()
+}
+
+func NewM3U8Sink(id stream.SinkId, sourceId string, cb func(m3u8 []byte), sessionId string) stream.Sink {
+	return &M3U8Sink{
+		BaseSink:  stream.BaseSink{Id_: id, SourceId_: sourceId, Protocol_: stream.ProtocolHls},
+		cb:        cb,
+		sessionId: sessionId,
+	}
 }

hls/hls_stream.go

@@ -35,7 +35,8 @@ type transStream struct {
 	duration       int      //切片时长, 单位秒
 	playlistLength int      //最大切片文件个数
 
-	m3u8Sinks map[stream.SinkId]stream.Sink
+	m3u8Sinks        map[stream.SinkId]*M3U8Sink //等待响应m3u8文件的sink
+	m3u8StringFormat string                      //一个协程写, 多个协程读, 不用加锁保护
 }
 
 func (t *transStream) Input(packet utils.AVPacket) error {
@@ -89,16 +90,14 @@ func (t *transStream) WriteHeader() error {
 }
 
 func (t *transStream) AddSink(sink stream.Sink) error {
-	if sink_, ok := sink.(*m3u8Sink); ok {
+	t.BaseTransStream.AddSink(sink)
+
 	if t.m3u8.Size() > 0 {
 		return sink.Input([]byte(t.m3u8.ToString()))
-		} else {
-			t.m3u8Sinks[sink.Id()] = sink_
-			return nil
-		}
 	}
 
-	return t.BaseTransStream.AddSink(sink)
+	t.m3u8Sinks[sink.Id()] = sink.(*M3U8Sink)
+	return nil
 }
 
 func (t *transStream) onTSWrite(data []byte) {
@@ -139,19 +138,17 @@ func (t *transStream) flushSegment(end bool) error {
 	duration := float32(t.muxer.Duration()) / 90000
 
 	t.m3u8.AddSegment(duration, t.context.url, t.context.segmentSeq, t.context.path)
-	if _, err := t.m3u8File.Seek(0, 0); err != nil {
-		return err
-	}
-	if err := t.m3u8File.Truncate(0); err != nil {
-		return err
-	}
-
 	m3u8Txt := t.m3u8.ToString()
 	if end {
 		m3u8Txt += "#EXT-X-ENDLIST"
 	}
+	t.m3u8StringFormat = m3u8Txt
 
-	if _, err := t.m3u8File.Write([]byte(m3u8Txt)); err != nil {
+	if _, err := t.m3u8File.Seek(0, 0); err != nil {
+		return err
+	} else if err := t.m3u8File.Truncate(0); err != nil {
+		return err
+	} else if _, err := t.m3u8File.Write([]byte(m3u8Txt)); err != nil {
 		return err
 	}
 
@@ -159,9 +156,10 @@ func (t *transStream) flushSegment(end bool) error {
 	//缓存完第二个切片, 才响应发送m3u8文件. 如果一个切片就发, 播放器缓存少会卡顿.
 	if len(t.m3u8Sinks) > 0 && t.m3u8.Size() > 1 {
 		for _, sink := range t.m3u8Sinks {
-			sink.Input([]byte(m3u8Txt))
+			sink.SendM3U8Data(&t.m3u8StringFormat)
 		}
-		t.m3u8Sinks = make(map[stream.SinkId]stream.Sink, 0)
+
+		t.m3u8Sinks = make(map[stream.SinkId]*M3U8Sink, 0)
 	}
 	return nil
 }
@@ -283,8 +281,7 @@ func NewTransStream(dir, m3u8Name, tsFormat, tsUrl string, segmentDuration, play
 	stream_.m3u8 = NewM3U8Writer(playlistLength)
 	stream_.m3u8File = file
 
-	//等待响应m3u8文件的sink
+	stream_.m3u8Sinks = make(map[stream.SinkId]*M3U8Sink, 24)
-	stream_.m3u8Sinks = make(map[stream.SinkId]stream.Sink, 24)
 	return stream_, nil
 }
 

hls/m3u8.go

@@ -131,7 +131,7 @@ func (m *m3u8Writer) ToString() string {
 			m.stringBuffer.WriteString("#EXTINF:")
 			m.stringBuffer.WriteString(strconv.FormatFloat(float64(segment.(Segment).duration), 'f', -1, 32))
 			m.stringBuffer.WriteString(",\r\n")
-			m.stringBuffer.WriteString(segment.(Segment).url)
+			m.stringBuffer.WriteString(segment.(Segment).url + "%s")
 			m.stringBuffer.WriteString("\r\n")
 		}
 	}