commit c484859e32f2a8336a78c5d7966c0e9f2d2a4022
parent 947ac78cf7609137d996869dd37f1ebeada69ee1
Author: Oliver Lowe <o@olowe.co>
Date: Sat, 22 Jun 2024 15:14:41 +1000
rtp: implement RTP sessions and packet transmission
This is not meant to be performant or robust to any flaky networks.
Instead it lets Go's test runner generate and consume streams by
starting a session on the local system, transmit some packets to a
fake player, then check if the packets are formed ok. Now we finally
have a way to mess with our packets and headers and get fast feedback
without involving some other program like ffmpeg or VLC to do
streaming for us!
Diffstat:
| M | rtp/rtp.go | | | 5 | +++++ |
| A | rtp/session.go | | | 77 | +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ |
| A | rtp/session_test.go | | | 137 | +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ |
3 files changed, 219 insertions(+), 0 deletions(-)
diff --git a/rtp/rtp.go b/rtp/rtp.go
@@ -105,6 +105,11 @@ func (t PayloadType) String() string {
return "unknown"
}
+const (
+ ClockMP2T = 90000 // 90KHz
+ ClockText = 1000 // 1KHz
+)
+
type Extension struct {
Profile [2]byte
Data []byte
diff --git a/rtp/session.go b/rtp/session.go
@@ -0,0 +1,77 @@
+package rtp
+
+import (
+ "fmt"
+ "math/rand"
+ "net"
+ "time"
+)
+
+func Dial(network, addr string) (*Session, error) {
+ conn, err := net.Dial(network, addr)
+ if err != nil {
+ return nil, err
+ }
+ s := Session{conn: conn}
+ s.init()
+ return &s, nil
+}
+
+// Session represents a RTP session... TODO(otl)
+// When a Session is established with Dial(), ...
+type Session struct {
+ Type PayloadType
+ // Clock is the rate of the... in hertz.
+ // If zero, automatic detection attempted...
+ Clock int
+
+ conn net.Conn
+ sequence uint16
+ timestamp uint32
+ previous time.Time
+ syncSource uint32
+}
+
+func (s *Session) init() {
+ s.sequence = uint16(rand.Intn(0xffff)) // max uint16
+ s.syncSource = rand.Uint32()
+ s.timestamp = rand.Uint32()
+}
+
+// Transmit sends the encoded form of packet to the destination address in s.
+// The Session will manage
+func (s *Session) Transmit(packet *Packet) error {
+ if packet.Header.Version == 0 {
+ packet.Header.Version = VersionRFC3550
+ }
+
+ packet.Header.Sequence = s.sequence
+ s.sequence++
+
+ ticks := ticksSince(s.previous, s.Clock)
+ packet.Header.Timestamp = s.timestamp + ticks
+ s.previous = time.Now()
+ s.timestamp += ticks
+
+ if packet.Header.SyncSource == 0 {
+ packet.Header.SyncSource = s.syncSource
+ }
+
+ b, err := Marshal(packet)
+ if err != nil {
+ return fmt.Errorf("marshal packet: %w", err)
+ }
+ n, err := s.conn.Write(b)
+ if n != len(b) {
+ if err != nil {
+ return fmt.Errorf("short write %d bytes: %w", n, err)
+ }
+ return fmt.Errorf("short write (%d bytes)", n)
+ }
+ return err
+}
+
+func ticksSince(t time.Time, clockRate int) (ticks uint32) {
+ dur := int(time.Since(t)/time.Second) * clockRate
+ return uint32(dur)
+}
diff --git a/rtp/session_test.go b/rtp/session_test.go
@@ -0,0 +1,137 @@
+package rtp
+
+import (
+ "errors"
+ "fmt"
+ "net"
+ "testing"
+ "time"
+)
+
+// fakePlayer is a basic RTP packet receiver which discards packet
+// payloads. It verifies the stream of packets by inspecting the packet header,
+// thought it requires packets are received in order.
+type fakePlayer struct {
+ conn net.PacketConn
+ decoded chan *Packet
+ clock int
+ typ PayloadType
+ syncSource uint32
+}
+
+func (fp *fakePlayer) render(ch chan error) {
+ var prev Packet
+ for p := range fp.decoded {
+ if prev.Payload == nil {
+ fp.typ = p.Header.Type
+ fp.syncSource = p.Header.SyncSource
+ prev = *p
+ continue
+ }
+ if p.Header.Version != VersionRFC3550 {
+ ch <- fmt.Errorf("bad version %d, want %d", p.Header.Version, VersionRFC3550)
+ }
+ if p.Header.Type != fp.typ {
+ ch <- fmt.Errorf("unexpected payload type %d, want %d", p.Header.Type, fp.typ)
+ }
+ if p.Header.Sequence != prev.Header.Sequence+1 {
+ ch <- fmt.Errorf("bad packet sequence: previous %d, current %d", prev.Header.Sequence, p.Header.Sequence)
+ }
+ // TODO(otl): check timestamp is expected based on fp.clock.
+ if p.Header.SyncSource != fp.syncSource {
+ ch <- fmt.Errorf("unexpected sync source %d, want %d", p.Header.SyncSource, fp.syncSource)
+ }
+ // TODO(otl): check payload is expected? non-nil?
+ prev = *p
+ }
+}
+
+func (fp *fakePlayer) receive(ch chan error) {
+ go fp.render(ch)
+ buf := make([]byte, 1492)
+ for {
+ if err := fp.conn.SetDeadline(time.Now().Add(2 * time.Second)); err != nil {
+ ch <- err
+ }
+ n, _, err := fp.conn.ReadFrom(buf)
+ if errors.Is(err, net.ErrClosed) {
+ break
+ } else if err != nil {
+ ch <- err
+ continue
+ }
+ var p Packet
+ if err := Unmarshal(buf[:n], &p); err != nil {
+ ch <- fmt.Errorf("unmarshal packet: %w", err)
+ continue
+ }
+ fp.decoded <- &p
+ }
+ close(fp.decoded)
+}
+
+func (fp *fakePlayer) stop() error {
+ return fp.conn.Close()
+}
+
+// textPackets returns a channel that sends count Packets through ch every dur...
+func textPackets(dur time.Duration, count int) chan Packet {
+ ch := make(chan Packet)
+ go func() {
+ typ := DynamicPayloadType()
+ ticker := time.NewTicker(dur)
+ var i int
+ for t := range ticker.C {
+ ch <- Packet{
+ Header{Type: typ},
+ []byte(t.Format(time.RFC3339Nano)),
+ }
+ i++
+ if i == count {
+ ticker.Stop()
+ close(ch)
+ break
+ }
+ }
+ }()
+ return ch
+}
+
+func TestSession(t *testing.T) {
+ ln, err := net.ListenPacket("udp", "[::]:0")
+ if err != nil {
+ t.Fatal(err)
+ }
+ player := fakePlayer{
+ conn: ln,
+ decoded: make(chan *Packet),
+ clock: ClockText,
+ }
+
+ errs := make(chan error)
+ go player.receive(errs)
+
+ session, err := Dial("udp", ln.LocalAddr().String())
+ if err != nil {
+ t.Fatal(err)
+ }
+ session.Clock = ClockText
+ pchan := textPackets(40*time.Millisecond, 25)
+ for {
+ select {
+ case err := <-errs:
+ t.Error(err)
+ case p, ok := <-pchan:
+ if !ok {
+ if err := player.stop(); err != nil {
+ t.Errorf("stop fake player: %v", err)
+ }
+ return
+ }
+ if err := session.Transmit(&p); err != nil {
+ t.Errorf("transmit: %v", err)
+ continue
+ }
+ }
+ }
+}
