rtp: started using mpegts encoder inside rtp encoder so that fragmentation i.e. smaller rtp packets is easier. Streams fine.

This commit is contained in:
saxon 2018-11-21 00:09:42 +10:30
parent 42097ddef7
commit 16614df9f5
4 changed files with 95 additions and 47 deletions

View File

@ -389,7 +389,6 @@ loop:
} }
} }
} }
r.destination.release() r.destination.release()
r.config.Logger.Log(smartlogger.Debug, pkg+"done reading that clip from ring buffer") r.config.Logger.Log(smartlogger.Debug, pkg+"done reading that clip from ring buffer")

View File

@ -29,7 +29,6 @@ LICENSE
package revid package revid
import ( import (
"fmt"
"io" "io"
"net" "net"
"os" "os"
@ -318,7 +317,6 @@ func (s *udpSender) load(c *ring.Chunk) error {
} }
func (s *udpSender) send() error { func (s *udpSender) send() error {
fmt.Println(len(s.chunk.Bytes()))
_, err := s.chunk.WriteTo(s.conn) _, err := s.chunk.WriteTo(s.conn)
return err return err
} }

View File

@ -163,6 +163,8 @@ type Encoder struct {
frameInterval time.Duration frameInterval time.Duration
ptsOffset time.Duration ptsOffset time.Duration
psiCount uint
continuity map[int]byte continuity map[int]byte
} }
@ -195,32 +197,35 @@ const (
// generate handles the incoming data and generates equivalent mpegts packets - // generate handles the incoming data and generates equivalent mpegts packets -
// sending them to the output channel // sending them to the output channel
func (e *Encoder) Encode(nalu []byte) error { func (e *Encoder) Encode(nalu []byte) error {
// Write PAT if e.psiCount == 0 {
patPkt := Packet{ // Write PAT
PUSI: true, patPkt := Packet{
PID: patPid, PUSI: true,
CC: e.ccFor(patPid), PID: patPid,
AFC: hasPayload, CC: e.ccFor(patPid),
Payload: patTable, AFC: hasPayload,
} Payload: patTable,
_, err := e.dst.Write(patPkt.Bytes()) }
if err != nil { _, err := e.dst.Write(patPkt.Bytes())
return err if err != nil {
} return err
}
// Write PMT. // Write PMT.
pmtPkt := Packet{ pmtPkt := Packet{
PUSI: true, PUSI: true,
PID: pmtPid, PID: pmtPid,
CC: e.ccFor(pmtPid), CC: e.ccFor(pmtPid),
AFC: hasPayload, AFC: hasPayload,
Payload: pmtTable, Payload: pmtTable,
}
_, err = e.dst.Write(pmtPkt.Bytes())
if err != nil {
return err
}
e.psiCount = 100
} }
_, err = e.dst.Write(pmtPkt.Bytes()) e.psiCount -= 1
if err != nil {
return err
}
// Prepare PES data. // Prepare PES data.
pesPkt := pes.Packet{ pesPkt := pes.Packet{
StreamID: streamID, StreamID: streamID,
@ -250,7 +255,6 @@ func (e *Encoder) Encode(nalu []byte) error {
pkt.PCR = e.pcr() pkt.PCR = e.pcr()
pusi = false pusi = false
} }
_, err := e.dst.Write(pkt.Bytes()) _, err := e.dst.Write(pkt.Bytes())
if err != nil { if err != nil {
return err return err

View File

@ -28,6 +28,7 @@ LICENSE
package rtp package rtp
import ( import (
"bitbucket.org/ausocean/av/stream/mts"
"io" "io"
"math/rand" "math/rand"
"time" "time"
@ -36,51 +37,97 @@ import (
const ( const (
yes = 1 yes = 1
no = 0 no = 0
defaultPktType = 1 defaultPktType = 33
timestampFreq = 90000 // Hz timestampFreq = 90000 // Hz
mtsSize = 188
bufferSize = 1000
) )
type Queue struct {
buf [bufferSize][mtsSize]byte
right uint
left uint
len uint
}
func (q *Queue) Write(frame []byte) (int, error) {
copy(q.buf[q.right][:], frame)
q.right += 1
q.len += 1
return 188, nil
}
func (q *Queue) Read() []byte {
q.left += 1
q.len -= 1
return q.buf[q.left-1][:]
}
func (q *Queue) Len() uint {
return q.len
}
func (q *Queue) Reset() {
q.left = 0
q.right = 0
q.len = 0
}
type Encoder struct { type Encoder struct {
dst io.Writer dst io.Writer
ssrc uint32 ssrc uint32
seqNo uint16 seqNo uint16
clock time.Duration clock time.Duration
frameInterval time.Duration frameInterval time.Duration
fps int
mtsEncoder *mts.Encoder
queue *Queue
} }
// NewEncoder returns a new Encoder type given an io.Writer - the destination // NewEncoder returns a new Encoder type given an io.Writer - the destination
// after encoding and the desired fps // after encoding and the desired fps
func NewEncoder(dst io.Writer, fps int) *Encoder { func NewEncoder(dst io.Writer, fps int) *Encoder {
q := &Queue{}
return &Encoder{ return &Encoder{
dst: dst, dst: dst,
ssrc: rand.Uint32(), ssrc: rand.Uint32(),
frameInterval: time.Duration(float64(time.Second) / float64(fps)), frameInterval: time.Duration(float64(time.Second) / float64(fps)),
fps: fps,
mtsEncoder: mts.NewEncoder(q, float64(fps)),
queue: q,
} }
} }
// Encode takes a nalu unit and encodes it into an rtp packet and // Encode takes a nalu unit and encodes it into an rtp packet and
// writes to the io.Writer given in NewEncoder // writes to the io.Writer given in NewEncoder
func (e *Encoder) Encode(nalu []byte) error { func (e *Encoder) Encode(nalu []byte) error {
pkt := Pkt{ e.mtsEncoder.Encode(nalu)
V: rtpVer, // version for e.queue.Len() > 0 {
P: no, // padding var payload []byte
X: no, // header extension for i := 0; i < 7 && e.queue.Len() > 0; i++ {
CC: no, // CSRC count payload = append(payload, e.queue.Read()...)
M: no, // NOTE: need to check if this works (decoders should ignore this) }
PT: defaultPktType, // NOTE: 1-23 according to rtp-h264 specs (don't think we need this) pkt := Pkt{
SN: e.nxtSeqNo(), // sequence number V: rtpVer, // version
TS: e.nxtTimestamp(), // timestamp P: no, // padding
SSRC: e.ssrc, // source identifier X: no, // header extension
Payload: nalu, CC: no, // CSRC count
Padding: no, M: no, // NOTE: need to check if this works (decoders should ignore this)
} PT: defaultPktType, // 33 for mpegts
SN: e.nxtSeqNo(), // sequence number
TS: e.nxtTimestamp(), // timestamp
SSRC: e.ssrc, // source identifier
Payload: payload,
Padding: no,
}
_, err := e.dst.Write(pkt.Bytes())
if err != nil {
return err
}
_, err := e.dst.Write(pkt.Bytes()) e.tick()
if err != nil {
return err
} }
e.queue.Reset()
e.tick()
return nil return nil
} }