streaming

Media streaming and broadcast systems in Go
Log | Files | Refs | README | LICENSE

session_test.go (3071B)

      1 package rtp
      2 
      3 import (
      4 	"errors"
      5 	"fmt"
      6 	"net"
      7 	"testing"
      8 	"time"
      9 )
     10 
     11 // fakePlayer is a basic RTP packet receiver which discards packet
     12 // payloads. It verifies the stream of packets by inspecting the packet header,
     13 // thought it requires packets are received in order.
     14 type fakePlayer struct {
     15 	conn       net.PacketConn
     16 	decoded    chan *Packet
     17 	clock      int
     18 	typ        PayloadType
     19 	syncSource uint32
     20 }
     21 
     22 func (fp *fakePlayer) render(ch chan error) {
     23 	var prev Packet
     24 	for p := range fp.decoded {
     25 		if prev.Payload == nil {
     26 			fp.typ = p.Header.Type
     27 			fp.syncSource = p.Header.SyncSource
     28 			prev = *p
     29 			continue
     30 		}
     31 		if p.Header.Version != VersionRFC3550 {
     32 			ch <- fmt.Errorf("bad version %d, want %d", p.Header.Version, VersionRFC3550)
     33 		}
     34 		if p.Header.Type != fp.typ {
     35 			ch <- fmt.Errorf("unexpected payload type %d, want %d", p.Header.Type, fp.typ)
     36 		}
     37 		if p.Header.Sequence != prev.Header.Sequence+1 {
     38 			ch <- fmt.Errorf("bad packet sequence: previous %d, current %d", prev.Header.Sequence, p.Header.Sequence)
     39 		}
     40 		// TODO(otl): check timestamp is expected based on fp.clock.
     41 		if p.Header.SyncSource != fp.syncSource {
     42 			ch <- fmt.Errorf("unexpected sync source %d, want %d", p.Header.SyncSource, fp.syncSource)
     43 		}
     44 		// TODO(otl): check payload is expected? non-nil?
     45 		prev = *p
     46 	}
     47 }
     48 
     49 func (fp *fakePlayer) receive(ch chan error) {
     50 	go fp.render(ch)
     51 	buf := make([]byte, 1492)
     52 	for {
     53 		if err := fp.conn.SetDeadline(time.Now().Add(2 * time.Second)); err != nil {
     54 			ch <- err
     55 		}
     56 		n, _, err := fp.conn.ReadFrom(buf)
     57 		if errors.Is(err, net.ErrClosed) {
     58 			break
     59 		} else if err != nil {
     60 			ch <- err
     61 			continue
     62 		}
     63 		var p Packet
     64 		if err := Unmarshal(buf[:n], &p); err != nil {
     65 			ch <- fmt.Errorf("unmarshal packet: %w", err)
     66 			continue
     67 		}
     68 		fp.decoded <- &p
     69 	}
     70 	close(fp.decoded)
     71 }
     72 
     73 func (fp *fakePlayer) stop() error {
     74 	return fp.conn.Close()
     75 }
     76 
     77 // textPackets returns a channel that sends count Packets through ch every dur...
     78 func textPackets(dur time.Duration, count int) chan Packet {
     79 	ch := make(chan Packet)
     80 	go func() {
     81 		typ := DynamicPayloadType()
     82 		ticker := time.NewTicker(dur)
     83 		var i int
     84 		for t := range ticker.C {
     85 			ch <- Packet{
     86 				Header{Type: typ},
     87 				[]byte(t.Format(time.RFC3339Nano)),
     88 			}
     89 			i++
     90 			if i == count {
     91 				ticker.Stop()
     92 				close(ch)
     93 				break
     94 			}
     95 		}
     96 	}()
     97 	return ch
     98 }
     99 
    100 func TestSession(t *testing.T) {
    101 	ln, err := net.ListenPacket("udp", "[::1]:0")
    102 	if err != nil {
    103 		t.Fatal(err)
    104 	}
    105 	player := fakePlayer{
    106 		conn:    ln,
    107 		decoded: make(chan *Packet),
    108 		clock:   ClockText,
    109 	}
    110 
    111 	errs := make(chan error)
    112 	go player.receive(errs)
    113 
    114 	session, err := Dial("udp", ln.LocalAddr().String())
    115 	if err != nil {
    116 		t.Fatal(err)
    117 	}
    118 	session.Clock = ClockText
    119 	pchan := textPackets(40*time.Millisecond, 25)
    120 	for {
    121 		select {
    122 		case err := <-errs:
    123 			t.Error(err)
    124 		case p, ok := <-pchan:
    125 			if !ok {
    126 				if err := player.stop(); err != nil {
    127 					t.Errorf("stop fake player: %v", err)
    128 				}
    129 				return
    130 			}
    131 			if err := session.Transmit(&p); err != nil {
    132 				t.Errorf("transmit: %v", err)
    133 				continue
    134 			}
    135 		}
    136 	}
    137 }