mirror of https://github.com/gwuhaolin/livego.git
halwu(吴浩麟)
9 years ago
41 changed files with 921 additions and 961 deletions
@ -0,0 +1,264 @@
@@ -0,0 +1,264 @@
|
||||
package hls |
||||
|
||||
import ( |
||||
"fmt" |
||||
"time" |
||||
"bytes" |
||||
"log" |
||||
"errors" |
||||
"github.com/gwuhaolin/livego/parser" |
||||
"github.com/gwuhaolin/livego/av" |
||||
"github.com/gwuhaolin/livego/container/flv" |
||||
"github.com/gwuhaolin/livego/container/ts" |
||||
) |
||||
|
||||
const ( |
||||
videoHZ = 90000 |
||||
aacSampleLen = 1024 |
||||
maxQueueNum = 512 |
||||
|
||||
h264_default_hz uint64 = 90 |
||||
) |
||||
|
||||
type Source struct { |
||||
av.RWBaser |
||||
seq int |
||||
info av.Info |
||||
bwriter *bytes.Buffer |
||||
btswriter *bytes.Buffer |
||||
demuxer *flv.Demuxer |
||||
muxer *ts.Muxer |
||||
pts, dts uint64 |
||||
stat *status |
||||
align *align |
||||
cache *audioCache |
||||
tsCache *TSCacheItem |
||||
tsparser *parser.CodecParser |
||||
closed bool |
||||
packetQueue chan av.Packet |
||||
} |
||||
|
||||
func NewSource(info av.Info) *Source { |
||||
info.Inter = true |
||||
s := &Source{ |
||||
info: info, |
||||
align: &align{}, |
||||
stat: newStatus(), |
||||
RWBaser: av.NewRWBaser(time.Second * 10), |
||||
cache: newAudioCache(), |
||||
demuxer: flv.NewDemuxer(), |
||||
muxer: ts.NewMuxer(), |
||||
tsCache: NewTSCacheItem(info.Key), |
||||
tsparser: parser.NewCodecParser(), |
||||
bwriter: bytes.NewBuffer(make([]byte, 100*1024)), |
||||
packetQueue: make(chan av.Packet, maxQueueNum), |
||||
} |
||||
go func() { |
||||
err := s.SendPacket() |
||||
if err != nil { |
||||
log.Println("send packet error: ", err) |
||||
s.closed = true |
||||
} |
||||
}() |
||||
return s |
||||
} |
||||
|
||||
func (source *Source) GetCacheInc() *TSCacheItem { |
||||
return source.tsCache |
||||
} |
||||
|
||||
func (source *Source) DropPacket(pktQue chan av.Packet, info av.Info) { |
||||
log.Printf("[%v] packet queue max!!!", info) |
||||
for i := 0; i < maxQueueNum-84; i++ { |
||||
tmpPkt, ok := <-pktQue |
||||
// try to don't drop audio
|
||||
if ok && tmpPkt.IsAudio { |
||||
if len(pktQue) > maxQueueNum-2 { |
||||
<-pktQue |
||||
} else { |
||||
pktQue <- tmpPkt |
||||
} |
||||
} |
||||
|
||||
if ok && tmpPkt.IsVideo { |
||||
videoPkt, ok := tmpPkt.Header.(av.VideoPacketHeader) |
||||
// dont't drop sps config and dont't drop key frame
|
||||
if ok && (videoPkt.IsSeq() || videoPkt.IsKeyFrame()) { |
||||
pktQue <- tmpPkt |
||||
} |
||||
if len(pktQue) > maxQueueNum-10 { |
||||
<-pktQue |
||||
} |
||||
} |
||||
|
||||
} |
||||
log.Println("packet queue len: ", len(pktQue)) |
||||
} |
||||
|
||||
func (source *Source) Write(p av.Packet) error { |
||||
source.SetPreTime() |
||||
if len(source.packetQueue) >= maxQueueNum-24 { |
||||
source.DropPacket(source.packetQueue, source.info) |
||||
} else { |
||||
source.packetQueue <- p |
||||
} |
||||
return nil |
||||
} |
||||
|
||||
func (source *Source) SendPacket() error { |
||||
defer func() { |
||||
log.Printf("[%v] hls sender stop", source.info) |
||||
if r := recover(); r != nil { |
||||
log.Println("hls SendPacket panic: ", r) |
||||
} |
||||
}() |
||||
log.Printf("[%v] hls sender start", source.info) |
||||
for { |
||||
if source.closed { |
||||
return errors.New("closed") |
||||
} |
||||
|
||||
p, ok := <-source.packetQueue |
||||
if ok { |
||||
if p.IsMetadata { |
||||
continue |
||||
} |
||||
|
||||
err := source.demuxer.Demux(&p) |
||||
if err == flv.ErrAvcEndSEQ { |
||||
log.Println(err) |
||||
continue |
||||
} else { |
||||
if err != nil { |
||||
log.Println(err) |
||||
return err |
||||
} |
||||
} |
||||
compositionTime, isSeq, err := source.parse(&p) |
||||
if err != nil { |
||||
log.Println(err) |
||||
} |
||||
if err != nil || isSeq { |
||||
continue |
||||
} |
||||
if source.btswriter != nil { |
||||
source.stat.update(p.IsVideo, p.TimeStamp) |
||||
source.calcPtsDts(p.IsVideo, p.TimeStamp, uint32(compositionTime)) |
||||
source.tsMux(&p) |
||||
} |
||||
} else { |
||||
return errors.New("closed") |
||||
} |
||||
} |
||||
} |
||||
|
||||
func (source *Source) Info() (ret av.Info) { |
||||
return source.info |
||||
} |
||||
|
||||
func (source *Source) cleanup() { |
||||
close(source.packetQueue) |
||||
source.bwriter = nil |
||||
source.btswriter = nil |
||||
source.cache = nil |
||||
source.tsCache = nil |
||||
} |
||||
|
||||
func (source *Source) Close(err error) { |
||||
log.Println("hls source closed: ", source.info) |
||||
if !source.closed { |
||||
source.cleanup() |
||||
} |
||||
source.closed = true |
||||
} |
||||
|
||||
func (source *Source) cut() { |
||||
newf := true |
||||
if source.btswriter == nil { |
||||
source.btswriter = bytes.NewBuffer(nil) |
||||
} else if source.btswriter != nil && source.stat.durationMs() >= duration { |
||||
source.flushAudio() |
||||
|
||||
source.seq++ |
||||
filename := fmt.Sprintf("/%s/%d.ts", source.info.Key, time.Now().Unix()) |
||||
item := NewTSItem(filename, int(source.stat.durationMs()), source.seq, source.btswriter.Bytes()) |
||||
source.tsCache.SetItem(filename, item) |
||||
|
||||
source.btswriter.Reset() |
||||
source.stat.resetAndNew() |
||||
} else { |
||||
newf = false |
||||
} |
||||
if newf { |
||||
source.btswriter.Write(source.muxer.PAT()) |
||||
source.btswriter.Write(source.muxer.PMT(av.SOUND_AAC, true)) |
||||
} |
||||
} |
||||
|
||||
func (source *Source) parse(p *av.Packet) (int32, bool, error) { |
||||
var compositionTime int32 |
||||
var ah av.AudioPacketHeader |
||||
var vh av.VideoPacketHeader |
||||
if p.IsVideo { |
||||
vh = p.Header.(av.VideoPacketHeader) |
||||
if vh.CodecID() != av.VIDEO_H264 { |
||||
return compositionTime, false, ErrNoSupportVideoCodec |
||||
} |
||||
compositionTime = vh.CompositionTime() |
||||
if vh.IsKeyFrame() && vh.IsSeq() { |
||||
return compositionTime, true, source.tsparser.Parse(p, source.bwriter) |
||||
} |
||||
} else { |
||||
ah = p.Header.(av.AudioPacketHeader) |
||||
if ah.SoundFormat() != av.SOUND_AAC { |
||||
return compositionTime, false, ErrNoSupportAudioCodec |
||||
} |
||||
if ah.AACPacketType() == av.AAC_SEQHDR { |
||||
return compositionTime, true, source.tsparser.Parse(p, source.bwriter) |
||||
} |
||||
} |
||||
source.bwriter.Reset() |
||||
if err := source.tsparser.Parse(p, source.bwriter); err != nil { |
||||
return compositionTime, false, err |
||||
} |
||||
p.Data = source.bwriter.Bytes() |
||||
|
||||
if p.IsVideo && vh.IsKeyFrame() { |
||||
source.cut() |
||||
} |
||||
return compositionTime, false, nil |
||||
} |
||||
|
||||
func (source *Source) calcPtsDts(isVideo bool, ts, compositionTs uint32) { |
||||
source.dts = uint64(ts) * h264_default_hz |
||||
if isVideo { |
||||
source.pts = source.dts + uint64(compositionTs)*h264_default_hz |
||||
} else { |
||||
sampleRate, _ := source.tsparser.SampleRate() |
||||
source.align.align(&source.dts, uint32(videoHZ*aacSampleLen/sampleRate)) |
||||
source.pts = source.dts |
||||
} |
||||
} |
||||
func (source *Source) flushAudio() error { |
||||
return source.muxAudio(1) |
||||
} |
||||
|
||||
func (source *Source) muxAudio(limit byte) error { |
||||
if source.cache.CacheNum() < limit { |
||||
return nil |
||||
} |
||||
var p av.Packet |
||||
_, pts, buf := source.cache.GetFrame() |
||||
p.Data = buf |
||||
p.TimeStamp = uint32(pts / h264_default_hz) |
||||
return source.muxer.Mux(&p, source.btswriter) |
||||
} |
||||
|
||||
func (source *Source) tsMux(p *av.Packet) error { |
||||
if p.IsVideo { |
||||
return source.muxer.Mux(p, source.btswriter) |
||||
} else { |
||||
source.cache.Cache(p.Data, source.pts) |
||||
return source.muxAudio(cache_max_frames) |
||||
} |
||||
} |
||||
Loading…
Reference in new issue