/* NAME RtpToTsConverter.go - provides utilities for the conversion of Rtp packets to equivalent MpegTs packets. DESCRIPTION See Readme.md AUTHOR Saxon Nelson-Milton LICENSE RtpToTsConverter.go is Copyright (C) 2017 the Australian Ocean Lab (AusOcean) It is free software: you can redistribute it and/or modify them under the terms of the GNU General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version. It is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details. You should have received a copy of the GNU General Public License along with revid in gpl.txt. If not, see [GNU licenses](http://www.gnu.org/licenses). */ package tsgenerator import ( _ "fmt" _"os" "../mpegts" "../pes" "../tools" "../rtp" ) type TsGenerator interface { Generate() } type tsGenerator struct { TsChan <-chan *mpegts.MpegTsPacket tsChan chan<- *mpegts.MpegTsPacket InputChan chan<- rtp.RtpPacket inputChan <-chan rtp.RtpPacket NalInputChan chan<- []byte nalInputChan <-chan []byte currentTsPacket *mpegts.MpegTsPacket payloadByteChan chan byte currentCC byte currentPtsTime float64 currentPcrTime float64 fps uint isGenerating bool } func NewTsGenerator(fps uint) (g *tsGenerator) { g = new(tsGenerator) tsChan := make(chan *mpegts.MpegTsPacket, 100) g.TsChan = tsChan g.tsChan = tsChan inputChan := make(chan rtp.RtpPacket, 100) g.InputChan = inputChan g.inputChan = inputChan nalInputChan := make(chan []byte, 10000) g.NalInputChan = nalInputChan g.nalInputChan = nalInputChan g.currentCC = 0 g.fps = fps g.currentPcrTime = .0 g.currentPtsTime = .7 return } func (g *tsGenerator) genPts()(pts uint64){ pts = uint64(g.currentPtsTime * float64(90000)) g.currentPtsTime += 1.0/float64(g.fps) return } func (g *tsGenerator) genPcr()(pcr uint64){ pcr = uint64(g.currentPcrTime * float64(90000)) g.currentPcrTime += 1.0/float64(g.fps) return } func (g *tsGenerator) Stop(){ g.isGenerating = false } func (g *tsGenerator) Generate() { g.isGenerating = true pesPktChan := make(chan []byte, 1000) payloadByteChan := make(chan byte, 100000) var rtpBuffer [](*rtp.RtpPacket) for g.isGenerating { select { default: case rtpPacket := <-g.inputChan: rtpBuffer = append(rtpBuffer, &rtpPacket) if len(rtpBuffer) > 2 { // if there's something weird going on with sequence numbers then // insertion sort if rtpPacket.SequenceNumber < rtpBuffer[len(rtpBuffer)-2].SequenceNumber { for i := 1; i < len(rtpBuffer); i++ { for j := i; j > 0 && rtpBuffer[j].SequenceNumber < rtpBuffer[j-1].SequenceNumber; j-- { temp := rtpBuffer[j] rtpBuffer[j] = rtpBuffer[j-1] rtpBuffer[j-1] = temp } } } } if len(rtpBuffer) > 200 { for tools.GetOctectType(rtpBuffer[0]) != 7 { rtpBuffer = rtpBuffer[1:] } sps := make([]byte, len(rtpBuffer[0].Payload)) copy(sps[:], rtpBuffer[0].Payload[:]) rtpBuffer = rtpBuffer[1:] pps := make([]byte, len(rtpBuffer[0].Payload)) copy(pps[:], rtpBuffer[0].Payload[:]) rtpBuffer = rtpBuffer[1:] sei := make([]byte, len(rtpBuffer[0].Payload)) copy(sei[:], rtpBuffer[0].Payload[:]) rtpBuffer = rtpBuffer[1:] for tools.GetOctectType(rtpBuffer[0]) != 7 { switch tools.GetOctectType(rtpBuffer[0]) { case 28: if tools.GetStartBit(rtpBuffer[0]) == 1 { var buffer []byte buffer = append(buffer, []byte{0x00, 0x00, 0x01}...) buffer = append(buffer, []byte{0x09, 0x10}...) buffer = append(buffer, []byte{0x00, 0x00, 0x01}...) buffer = append(buffer, sps...) buffer = append(buffer, []byte{0x00, 0x00, 0x01}...) buffer = append(buffer, pps...) buffer = append(buffer, []byte{0x00, 0x00, 0x01}...) buffer = append(buffer, sei...) buffer = append(buffer, []byte{0x00, 0x00, 0x01}...) buffer = append(buffer, rtpBuffer[0].Payload[0]&0xE0|rtpBuffer[0].Payload[1]&0x1F) buffer = append(buffer, rtpBuffer[0].Payload[2:]...) rtpBuffer = rtpBuffer[1:] for { buffer = append(buffer, rtpBuffer[0].Payload[2:]...) if tools.GetEndBit(rtpBuffer[0]) == 1 { rtpBuffer = rtpBuffer[1:] g.NalInputChan <- buffer break } rtpBuffer = rtpBuffer[1:] } } case 1: var buffer []byte buffer = append(buffer, []byte{0x00, 0x00, 0x01}...) buffer = append(buffer, []byte{0x09, 0x10}...) buffer = append(buffer, []byte{0x00, 0x00, 0x01}...) buffer = append(buffer, sps...) buffer = append(buffer, []byte{0x00, 0x00, 0x01}...) buffer = append(buffer, pps...) buffer = append(buffer, []byte{0x00, 0x00, 0x01}...) buffer = append(buffer, sei...) buffer = append(buffer, []byte{0x00, 0x00, 0x01}...) buffer = append(buffer, rtpBuffer[0].Payload[0]&0xE0|rtpBuffer[0].Payload[1]&0x1F) buffer = append(buffer, rtpBuffer[0].Payload[2:]...) rtpBuffer = rtpBuffer[1:] g.NalInputChan <- buffer default: } } } case nalUnit := <-g.nalInputChan: pesPkt := pes.PESPacket{ StreamID: 0xE0, PDI: byte(2), PTS: g.genPts(), Data: nalUnit, HeaderLength: 5, } pesPktChan <- pesPkt.ToByteSlice() case pesPkt := <-pesPktChan: for ii := range pesPkt { payloadByteChan <- pesPkt[ii] } pusi := true for len(payloadByteChan) > 0 { pkt := mpegts.MpegTsPacket{ PUSI: pusi, PID: 256, RAI: pusi, CC: g.currentCC, AFC: byte(3), PCRF: pusi, } pkt.FillPayload(payloadByteChan) if pusi { pkt.PCR = g.genPcr() pusi = false } if g.currentCC++; g.currentCC > 15 { g.currentCC = 0 } g.tsChan <- &pkt } } } }