mirror
/
av
mirror of https://bitbucket.org/ausocean/av.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Seems to be working to some degree. Need to improve PSI stuff by adding CC information

This commit is contained in:
Jack Richardson 2018-01-16 15:36:51 +10:30
parent 0cf51ee5b1
commit 85ae2189f3
7 changed files with 151 additions and 106 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
h264/h264Parser.go
View File

@ -58,7 +58,7 @@ func (p* H264Parser)Stop(){
func (p* H264Parser)Parse() { func (p* H264Parser)Parse() {
p.isParsing = true p.isParsing = true
outputBuffer := []byte{} outputBuffer := make([]byte, 0, 10000)
searchingForEnd := false searchingForEnd := false
p.InputByteChan = make(chan byte, 10000) p.InputByteChan = make(chan byte, 10000)
for p.isParsing { for p.isParsing {

9
mpegts/MpegTs.go
View File

@ -34,6 +34,11 @@ import (
_"fmt" _"fmt"
) )
const (
maxMpegTsSize = 188
mpegtsPayloadSize = 176
)
/* /*
The below data struct encapsulates the fields of an MPEG-TS packet. Below is The below data struct encapsulates the fields of an MPEG-TS packet. Below is
the formatting of an MPEG-TS packet for reference! the formatting of an MPEG-TS packet for reference!
@ -127,7 +132,7 @@ type MpegTsPacket struct {
// TODO: make payload private considering we now have FillPayload method // TODO: make payload private considering we now have FillPayload method
func (p *MpegTsPacket) FillPayload(channel chan byte){ func (p *MpegTsPacket) FillPayload(channel chan byte){
p.Payload = []byte{} p.Payload = make([]byte,0,mpegtsPayloadSize)
currentPktLength := 6 + int(tools.BoolToByte(p.PCRF))*6+int(tools.BoolToByte(p.OPCRF))*6+ currentPktLength := 6 + int(tools.BoolToByte(p.PCRF))*6+int(tools.BoolToByte(p.OPCRF))*6+
int(tools.BoolToByte(p.SPF))*1+int(tools.BoolToByte(p.TPDF))*1+len(p.TPD) int(tools.BoolToByte(p.SPF))*1+int(tools.BoolToByte(p.TPDF))*1+len(p.TPD)
for len(channel) > 0 && (currentPktLength+len(p.Payload)) < 188 { for len(channel) > 0 && (currentPktLength+len(p.Payload)) < 188 {
@ -145,6 +150,7 @@ func (p *MpegTsPacket) ToByteSlice() (output []byte, err error) {
} }
afl := 1+int(tools.BoolToByte(p.PCRF))*6+int(tools.BoolToByte(p.OPCRF))* afl := 1+int(tools.BoolToByte(p.PCRF))*6+int(tools.BoolToByte(p.OPCRF))*
6+int(tools.BoolToByte(p.SPF))*1+int(tools.BoolToByte(p.TPDF))*1+len(p.TPD)+len(stuffing) 6+int(tools.BoolToByte(p.SPF))*1+int(tools.BoolToByte(p.TPDF))*1+len(p.TPD)+len(stuffing)
output = make([]byte,0,maxMpegTsSize)
output = append(output, []byte{ output = append(output, []byte{
0x47, 0x47,
(tools.BoolToByte(p.TEI)<<7 | tools.BoolToByte(p.PUSI)<<6 | tools.BoolToByte(p.Priority)<<5 | (tools.BoolToByte(p.TEI)<<7 | tools.BoolToByte(p.PUSI)<<6 | tools.BoolToByte(p.Priority)<<5 |
@ -167,7 +173,6 @@ func (p *MpegTsPacket) ToByteSlice() (output []byte, err error) {
if p.TPDF { if p.TPDF {
output = append(output, append([]byte{p.TPDL}, p.TPD...)...) output = append(output, append([]byte{p.TPDL}, p.TPD...)...)
} }
output = append(output, append(p.Ext, append(stuffing, p.Payload...)...)...) output = append(output, append(p.Ext, append(stuffing, p.Payload...)...)...)
if len(output) != 188 { if len(output) != 188 {
err = errors.New("Length of MPEG-TS packet is not 188! Something is wrong!") err = errors.New("Length of MPEG-TS packet is not 188! Something is wrong!")

8
pes/Pes.go
View File

@ -29,6 +29,11 @@ package pes
import ( import (
"bitbucket.org/ausocean/av/tools" "bitbucket.org/ausocean/av/tools"
) )
const (
maxPesSize = 10000
)
/* /*
The below data struct encapsulates the fields of an PES packet. Below is The below data struct encapsulates the fields of an PES packet. Below is
the formatting of a PES packet for reference! the formatting of a PES packet for reference!
@ -68,6 +73,7 @@ the formatting of a PES packet for reference!
| - | ... | | - | ... |
---------------------------------------------------------------------------- ----------------------------------------------------------------------------
*/ */
// TODO: add DSMTM, ACI, CRC, Ext fields
type PESPacket struct { type PESPacket struct {
StreamID byte // Type of stream StreamID byte // Type of stream
Length uint16 // Pes packet length in bytes after this field Length uint16 // Pes packet length in bytes after this field
@ -88,12 +94,12 @@ type PESPacket struct {
DTS uint64 // Decoding timestamp DTS uint64 // Decoding timestamp
ESCR uint64 // Elementary stream clock reference ESCR uint64 // Elementary stream clock reference
ESR uint32 // Elementary stream rate reference ESR uint32 // Elementary stream rate reference
// TODO: add DSMTM, ACI, CRC, Ext fields
Stuff []byte // Stuffing bytes Stuff []byte // Stuffing bytes
Data []byte // Pes packet data Data []byte // Pes packet data
} }
func (p *PESPacket) ToByteSlice() (output []byte) { func (p *PESPacket) ToByteSlice() (output []byte) {
output = make([]byte, 0, maxPesSize)
output = append(output, []byte{ output = append(output, []byte{
0x00, 0x00, 0x01, 0x00, 0x00, 0x01,
p.StreamID, p.StreamID,

145
revid/RevidInstance.go
View File

@ -35,6 +35,7 @@ import (
"crypto/md5" "crypto/md5"
"encoding/hex" "encoding/hex"
"fmt" "fmt"
"io"
"io/ioutil" "io/ioutil"
"log" "log"
"net" "net"
@ -43,7 +44,6 @@ import (
"os/exec" "os/exec"
"strconv" "strconv"
"time" "time"
"io"
"bitbucket.org/ausocean/av/h264" "bitbucket.org/ausocean/av/h264"
"bitbucket.org/ausocean/av/tsgenerator" "bitbucket.org/ausocean/av/tsgenerator"
@ -62,7 +62,7 @@ const (
rtpPackets = 7 // # of RTP packets per ethernet frame (7 is the max) rtpPackets = 7 // # of RTP packets per ethernet frame (7 is the max)
rtpHeaderSize = 12 rtpHeaderSize = 12
rtpSSRC = 1 // any value will do rtpSSRC = 1 // any value will do
bufferSize = 1000 / clipDuration bufferSize = 100 / clipDuration
httpTimeOut = 5 // s httpTimeOut = 5 // s
motionThreshold = "0.0025" motionThreshold = "0.0025"
qscale = "3" qscale = "3"
@ -73,19 +73,27 @@ const (
) )
const ( const (
raspivid = 0 Raspivid = 0
rtp = 1 Rtp = 1
h264Codec = 2 H264Codec = 2
file = 4 File = 4
httpOut = 5 HttpOut = 5
) )
var cmd *exec.Cmd
var inputReader *bufio.Reader
type Config struct { type Config struct {
Input uint8 Input uint8
InputCmd string InputCmd string
Output uint8 Output uint8
OutputFileName string OutputFileName string
InputFileName string InputFileName string
Height string
Width string
Bitrate string
FrameRate string
HttpAddress string
} }
type RevidInst interface { type RevidInst interface {
@ -107,6 +115,8 @@ type revidInst struct {
Error *log.Logger Error *log.Logger
outputFile *os.File outputFile *os.File
inputFile *os.File inputFile *os.File
generator tsgenerator.TsGenerator
h264Parser h264.H264Parser
} }
func NewRevidInstance(config Config) (r *revidInst, err error) { func NewRevidInstance(config Config) (r *revidInst, err error) {
@ -118,19 +128,25 @@ func NewRevidInstance(config Config) (r *revidInst, err error) {
r.dumpPCRBase = 0 r.dumpPCRBase = 0
r.ChangeState(config) r.ChangeState(config)
switch r.config.Output { switch r.config.Output {
case file: case File:
r.outputFile, err = os.Create(r.config.OutputFileName) r.outputFile, err = os.Create(r.config.OutputFileName)
if err != nil { if err != nil {
return nil, err return nil, err
} }
} }
switch r.config.Input { switch r.config.Input {
case file: case File:
r.inputFile, err = os.Open(r.config.InputFileName) r.inputFile, err = os.Open(r.config.InputFileName)
if err != nil { if err != nil {
return nil, err return nil, err
} }
} }
r.generator = tsgenerator.NewTsGenerator(framesPerSec)
r.h264Parser = h264.H264Parser{OutputChan: r.generator.GetNalInputChan()}
// TODO: Need to create constructor for parser otherwise I'm going to break
// something eventuallyl
go r.h264Parser.Parse()
go r.input()
return return
} }
@ -142,25 +158,11 @@ func (r *revidInst) ChangeState(newConfig Config) error {
func (r *revidInst) Start() { func (r *revidInst) Start() {
r.isRunning = true r.isRunning = true
go r.input() go r.generator.Generate()
go r.output()
}
func (r *revidInst) Stop() {
r.isRunning = false
}
func (r *revidInst) input() {
generator := tsgenerator.NewTsGenerator(framesPerSec)
go generator.Generate()
h264Parser := h264.H264Parser{OutputChan: generator.NalInputChan}
// TODO: Need to create constructor for parser otherwise I'm going to break
// something eventuallyl
go h264Parser.Parse()
var inputReader *bufio.Reader
switch r.config.Input { switch r.config.Input {
case raspivid: case Raspivid:
cmd := exec.Command("raspivid", "-o", "-", "-n", "-t", "0", "-b", "1000000", "-w","1280","-h","720") cmd = exec.Command("raspivid", "-o", "-", "-n", "-t", "0", "-b",
r.config.Bitrate, "-w", r.config.Width, "-h", r.config.Height, "-fps", r.config.FrameRate)
stdout, _ := cmd.StdoutPipe() stdout, _ := cmd.StdoutPipe()
err := cmd.Start() err := cmd.Start()
inputReader = bufio.NewReader(stdout) inputReader = bufio.NewReader(stdout)
@ -168,40 +170,30 @@ func (r *revidInst) input() {
r.Error.Println(err.Error()) r.Error.Println(err.Error())
return return
} }
case file: case File:
default: default:
r.Error.Println("Input not valid!") r.Error.Println("Input not valid!")
} }
clipSize := 0
packetCount := 0
now := time.Now()
prevTime := now
startPackets := [][]byte{
{71, 64, 17, 16, 0, 66, 240, 65, 0, 1, 193, 0, 0, 255, 1, 255, 0, 1, 252, 128, 48, 72, 46, 1, 6, 70, 70, 109, 112, 101, 103, 37, 115, 116, 114, 101, 97, 109, 101, 100, 32, 98, 121, 32, 116, 104, 101, 32, 71, 101, 111, 86, 105, 115, 105, 111, 110, 32, 82, 116, 115, 112, 32, 83, 101, 114, 118, 101, 114, 99, 176, 214, 195, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255},
{71, 64, 0, 16, 0, 0, 176, 13, 0, 1, 193, 0, 0, 0, 1, 240, 0, 42, 177, 4, 178, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255},
/*PMT*/ {71, 80, 0, 16,
/*Start of payload*/
0, 2, 176, 18, 0, 1, 193, 0, 0, 0xE1, 0x00, 0xF0, 0, 0x1B, 0xE1, 0, 0xF0, 0, 0x15, 0xBD, 0x4D, 0x56, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255},
}
donePSI := false
ii := 0
fmt.Println("reading")
var h264Data []byte var h264Data []byte
switch(r.config.Input){ switch r.config.Input {
case raspivid: case Raspivid:
go func() { go func() {
for { for r.isRunning {
h264Data = make([]byte, 2) h264Data = make([]byte, 1)
_, err := io.ReadFull(inputReader, h264Data) _, err := io.ReadFull(inputReader, h264Data)
if err == nil { if err != nil {
h264Parser.InputByteChan<-h264Data[0] if err.Error() == "EOF" {
h264Parser.InputByteChan<-h264Data[1] r.Error.Println("No data from camera!")
time.Sleep(5*time.Second)
} else {
r.Error.Println(err.Error())
}
} else {
r.h264Parser.InputByteChan <- h264Data[0]
} }
} }
}() }()
case file: case File:
stats, err := r.inputFile.Stat() stats, err := r.inputFile.Stat()
if err != nil { if err != nil {
panic("Could not get file stats!") panic("Could not get file stats!")
@ -212,32 +204,53 @@ func (r *revidInst) input() {
r.Error.Println(err.Error()) r.Error.Println(err.Error())
} }
for i := range h264Data { for i := range h264Data {
h264Parser.InputByteChan<-h264Data[i] r.h264Parser.InputByteChan <- h264Data[i]
} }
} }
for r.isRunning { go r.output()
}
func (r *revidInst) Stop() {
if r.isRunning {
r.isRunning = false
r.generator.Stop()
cmd.Process.Kill()
}
}
func (r *revidInst) input() {
clipSize := 0
packetCount := 0
now := time.Now()
prevTime := now
startPackets := [][]byte{
{71, 64, 17, 16, 0, 66, 240, 65, 0, 1, 193, 0, 0, 255, 1, 255, 0, 1, 252, 128, 48, 72, 46, 1, 6, 70, 70, 109, 112, 101, 103, 37, 115, 116, 114, 101, 97, 109, 101, 100, 32, 98, 121, 32, 116, 104, 101, 32, 71, 101, 111, 86, 105, 115, 105, 111, 110, 32, 82, 116, 115, 112, 32, 83, 101, 114, 118, 101, 114, 99, 176, 214, 195, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255},
{71, 64, 0, 16, 0, 0, 176, 13, 0, 1, 193, 0, 0, 0, 1, 240, 0, 42, 177, 4, 178, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255},
/*PMT*/ {71, 80, 0, 16,
/*Start of payload*/
0, 2, 176, 18, 0, 1, 193, 0, 0, 0xE1, 0x00, 0xF0, 0, 0x1B, 0xE1, 0, 0xF0, 0, 0x15, 0xBD, 0x4D, 0x56, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255},
}
fmt.Println("reading")
for {
if clip, err := r.ringBuffer.Get(); err != nil { if clip, err := r.ringBuffer.Get(); err != nil {
r.Error.Println(err.Error()) r.Error.Println(err.Error())
return return
} else { } else {
for { for {
tsPacket := <-(r.generator.GetTsOutputChan())
for ii:=0; ii < 3 && tsPacket.PUSI; ii++ {
upperBound := clipSize + mp2tPacketSize upperBound := clipSize + mp2tPacketSize
if ii < 3 && !donePSI {
packetByteSlice := startPackets[ii] packetByteSlice := startPackets[ii]
copy(clip[clipSize:upperBound], packetByteSlice) copy(clip[clipSize:upperBound], packetByteSlice)
ii++ packetCount++
} else { clipSize += mp2tPacketSize
donePSI = true
if err != nil {
fmt.Println(err)
} }
tsPacket := <-generator.TsChan
byteSlice, err := tsPacket.ToByteSlice() byteSlice, err := tsPacket.ToByteSlice()
if err != nil { if err != nil {
r.Error.Println(err.Error()) r.Error.Println(err.Error())
} }
upperBound := clipSize + mp2tPacketSize
copy(clip[clipSize:upperBound], byteSlice) copy(clip[clipSize:upperBound], byteSlice)
}
packetCount++ packetCount++
clipSize += mp2tPacketSize clipSize += mp2tPacketSize
// send if (1) our buffer is full or (2) 1 second has elapsed and we have % packetsPerFrame // send if (1) our buffer is full or (2) 1 second has elapsed and we have % packetsPerFrame
@ -262,8 +275,12 @@ func (r *revidInst) output() {
for r.isRunning { for r.isRunning {
if clip, err := r.ringBuffer.Read(); err == nil { if clip, err := r.ringBuffer.Read(); err == nil {
switch r.config.Output { switch r.config.Output {
case file: case File:
r.outputFile.Write(clip) r.outputFile.Write(clip)
case HttpOut:
for err := sendClipToHTTP(clip, r.config.HttpAddress); err != nil; {
err = sendClipToHTTP(clip, r.config.HttpAddress)
}
default: default:
r.Error.Println("No output?") r.Error.Println("No output?")
} }
@ -275,7 +292,7 @@ func (r *revidInst) output() {
} }
// sendClipToHTPP posts a video clip via HTTP, using a new TCP connection each time. // sendClipToHTPP posts a video clip via HTTP, using a new TCP connection each time.
func sendClipToHTTP(clip []byte, output string, _ net.Conn) error { func sendClipToHTTP(clip []byte, output string) error {
timeout := time.Duration(httpTimeOut * time.Second) timeout := time.Duration(httpTimeOut * time.Second)
client := http.Client{ client := http.Client{
Timeout: timeout, Timeout: timeout,

8
revid/revid_test.go
View File

@ -61,9 +61,13 @@ func TestFileInput(t *testing.T){
*/ */
func TestRaspividInput(t *testing.T){ func TestRaspividInput(t *testing.T){
config := Config{ config := Config{
Input: raspivid, Input: Raspivid,
Output: file, Output: File,
OutputFileName: "output/TestRaspividOutput.ts", OutputFileName: "output/TestRaspividOutput.ts",
Width: "1280",
Height: "720",
Bitrate: "1000000",
FrameRate: "25",
} }
revidInst, err := NewRevidInstance(config) revidInst, err := NewRevidInstance(config)
if err != nil { if err != nil {

10
ringbuffer/RingBuffer.go
View File

@ -95,14 +95,16 @@ func (rb *ringBuffer) Get() ([]byte, error) {
if !rb.IsWritable() { if !rb.IsWritable() {
return nil, errors.New("Buffer full!") return nil, errors.New("Buffer full!")
} }
if rb.currentlyWriting { var nextlast int
return nil, errors.New("Second call to Get! Call DoneWriting first!") if !rb.currentlyWriting {
}
rb.currentlyWriting = true rb.currentlyWriting = true
nextlast := rb.last + 1 nextlast = rb.last + 1
if nextlast == rb.size { if nextlast == rb.size {
nextlast = 0 nextlast = 0
} }
} else {
nextlast = rb.last
}
return rb.dataMemory[nextlast], nil return rb.dataMemory[nextlast], nil
} }

27
tsgenerator/TsGenerator.go
View File

@ -39,6 +39,9 @@ import (
type TsGenerator interface { type TsGenerator interface {
Generate() Generate()
GetNalInputChan() chan<- []byte
GetTsOutputChan() <-chan *mpegts.MpegTsPacket
Stop()
} }
type tsGenerator struct { type tsGenerator struct {
@ -55,6 +58,15 @@ type tsGenerator struct {
currentPcrTime float64 currentPcrTime float64
fps uint fps uint
isGenerating bool isGenerating bool
pesPktChan chan []byte
}
func (g *tsGenerator)GetNalInputChan() chan<- []byte {
return g.NalInputChan
}
func (g *tsGenerator)GetTsOutputChan() <-chan *mpegts.MpegTsPacket {
return g.TsChan
} }
func NewTsGenerator(fps uint) (g *tsGenerator) { func NewTsGenerator(fps uint) (g *tsGenerator) {
@ -72,6 +84,8 @@ func NewTsGenerator(fps uint) (g *tsGenerator) {
g.fps = fps g.fps = fps
g.currentPcrTime = .0 g.currentPcrTime = .0
g.currentPtsTime = .7 g.currentPtsTime = .7
g.pesPktChan = make(chan []byte, 1000)
g.payloadByteChan = make(chan byte, 100000)
return return
} }
@ -93,12 +107,9 @@ func (g *tsGenerator) Stop(){
func (g *tsGenerator) Generate() { func (g *tsGenerator) Generate() {
g.isGenerating = true g.isGenerating = true
pesPktChan := make(chan []byte, 1000)
payloadByteChan := make(chan byte, 100000)
var rtpBuffer [](*rtp.RtpPacket) var rtpBuffer [](*rtp.RtpPacket)
for g.isGenerating { for g.isGenerating {
select { select {
default:
case rtpPacket := <-g.inputChan: case rtpPacket := <-g.inputChan:
rtpBuffer = append(rtpBuffer, &rtpPacket) rtpBuffer = append(rtpBuffer, &rtpPacket)
if len(rtpBuffer) > 2 { if len(rtpBuffer) > 2 {
@ -182,13 +193,13 @@ func (g *tsGenerator) Generate() {
Data: nalUnit, Data: nalUnit,
HeaderLength: 5, HeaderLength: 5,
} }
pesPktChan <- pesPkt.ToByteSlice() g.pesPktChan <- pesPkt.ToByteSlice()
case pesPkt := <-pesPktChan: case pesPkt := <-g.pesPktChan:
for ii := range pesPkt { for ii := range pesPkt {
payloadByteChan <- pesPkt[ii] g.payloadByteChan <- pesPkt[ii]
} }
pusi := true pusi := true
for len(payloadByteChan) > 0 { for len(g.payloadByteChan) > 0 {
pkt := mpegts.MpegTsPacket{ pkt := mpegts.MpegTsPacket{
PUSI: pusi, PUSI: pusi,
PID: 256, PID: 256,
@ -197,7 +208,7 @@ func (g *tsGenerator) Generate() {
AFC: byte(3), AFC: byte(3),
PCRF: pusi, PCRF: pusi,
} }
pkt.FillPayload(payloadByteChan) pkt.FillPayload(g.payloadByteChan)
if pusi { if pusi {
pkt.PCR = g.genPcr() pkt.PCR = g.genPcr()
pusi = false pusi = false