WebRTC offers multiple quality

messaging/quality.go

@@ -10,6 +10,12 @@ import (
 // Quality holds a specific stream quality.
 // It makes packages able to subscribe to an incoming stream.
 type Quality struct {
+	// Type of the quality
+	Name string
+
+	// Source Stream
+	Stream *Stream
+
 	// Incoming data come from this channel
 	Broadcast chan<- []byte
 
@@ -27,8 +33,9 @@ type Quality struct {
 	WebRtcRemoteSdp chan webrtc.SessionDescription
 }
 
-func newQuality() (q *Quality) {
-	q = &Quality{}
+func newQuality(name string, stream *Stream) (q *Quality) {
+	q = &Quality{Name: name}
+	q.Stream = stream
 	broadcast := make(chan []byte, 1024)
 	q.Broadcast = broadcast
 	q.outputs = make(map[chan []byte]struct{})

messaging/stream.go

@@ -40,7 +40,7 @@ func (s *Stream) CreateQuality(name string) (quality *Quality, err error) {
 	}
 
 	s.lockQualities.Lock()
-	quality = newQuality()
+	quality = newQuality(name, s)
 	s.qualities[name] = quality
 	s.lockQualities.Unlock()
 	return quality, nil

stream/srt/handler.go

@@ -24,6 +24,17 @@ func handleStreamer(socket *srtgo.SrtSocket, streams *messaging.Streams, name st
 		socket.Close()
 		return
 	}
+
+	// Create sub-qualities
+	for _, qualityName := range []string{"audio", "480p", "360p", "240p"} {
+		_, err := stream.CreateQuality(qualityName)
+		if err != nil {
+			log.Printf("Error on quality creating: %s", err)
+			socket.Close()
+			return
+		}
+	}
+
 	log.Printf("New SRT streamer for stream '%s' quality 'source'", name)
 
 	// Read RTP packets forever and send them to the WebRTC Client

stream/webrtc/ingest.go

@@ -14,33 +14,61 @@ import (
 
 func ingest(name string, q *messaging.Quality) {
 	// Register to get stream
-	videoInput := make(chan []byte, 1024)
-	q.Register(videoInput)
+	input := make(chan []byte, 1024)
+	// FIXME Stream data should already be transcoded
+	source, _ := q.Stream.GetQuality("source")
+	source.Register(input)
 
-	// Open a UDP Listener for RTP Packets on port 5004
-	videoListener, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 5004})
-	if err != nil {
-		log.Printf("Faited to open UDP listener %s", err)
-		return
+	// FIXME Bad code
+	port := 5000
+	var tracks map[string][]*webrtc.Track
+	qualityName := ""
+	switch q.Name {
+	case "audio":
+		port = 5004
+		tracks = audioTracks
+		break
+	case "source":
+		port = 5005
+		tracks = videoTracks
+		qualityName = "@source"
+		break
+	case "480p":
+		port = 5006
+		tracks = videoTracks
+		qualityName = "@480p"
+		break
+	case "360p":
+		port = 5007
+		tracks = videoTracks
+		qualityName = "@360p"
+		break
+	case "240p":
+		port = 5008
+		tracks = videoTracks
+		qualityName = "@240p"
+		break
 	}
-	audioListener, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 5005})
+
+	// Open a UDP Listener for RTP Packets
+	listener, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: port})
 	if err != nil {
 		log.Printf("Faited to open UDP listener %s", err)
 		return
 	}
 
-	// Start ffmpag to convert videoInput to video and audio UDP
-	ffmpeg, err := startFFmpeg(videoInput)
+	// Start ffmpag to convert input to video and audio UDP
+	ffmpeg, err := startFFmpeg(q, input)
 	if err != nil {
 		log.Printf("Error while starting ffmpeg: %s", err)
 		return
 	}
 
-	// Receive video
+	// Receive stream
 	go func() {
 		inboundRTPPacket := make([]byte, 1500) // UDP MTU
 		for {
-			n, _, err := videoListener.ReadFromUDP(inboundRTPPacket)
+			n, _, err := listener.ReadFromUDP(inboundRTPPacket)
 			if err != nil {
 				log.Printf("Failed to read from UDP: %s", err)
 				break
@@ -51,49 +79,13 @@ func ingest(name string, q *messaging.Quality) {
 				continue
 			}
 
-			if videoTracks[name] == nil {
-				videoTracks[name] = make([]*webrtc.Track, 0)
-			}
-
-			// Write RTP srtPacket to all video tracks
+			// Write RTP srtPacket to all tracks
 			// Adapt payload and SSRC to match destination
-			for _, videoTrack := range videoTracks[name] {
-				packet.Header.PayloadType = videoTrack.PayloadType()
-				packet.Header.SSRC = videoTrack.SSRC()
-				if writeErr := videoTrack.WriteRTP(packet); writeErr != nil {
-					log.Printf("Failed to write to video track: %s", err)
-					continue
-				}
-			}
-		}
-	}()
-
-	// Receive audio
-	go func() {
-		inboundRTPPacket := make([]byte, 1500) // UDP MTU
-		for {
-			n, _, err := audioListener.ReadFromUDP(inboundRTPPacket)
-			if err != nil {
-				log.Printf("Failed to read from UDP: %s", err)
-				break
-			}
-			packet := &rtp.Packet{}
-			if err := packet.Unmarshal(inboundRTPPacket[:n]); err != nil {
-				log.Printf("Failed to unmarshal RTP srtPacket: %s", err)
-				continue
-			}
-
-			if audioTracks[name] == nil {
-				audioTracks[name] = make([]*webrtc.Track, 0)
-			}
-
-			// Write RTP srtPacket to all audio tracks
-			// Adapt payload and SSRC to match destination
-			for _, audioTrack := range audioTracks[name] {
-				packet.Header.PayloadType = audioTrack.PayloadType()
-				packet.Header.SSRC = audioTrack.SSRC()
-				if writeErr := audioTrack.WriteRTP(packet); writeErr != nil {
-					log.Printf("Failed to write to audio track: %s", err)
+			for _, track := range tracks[name+qualityName] {
+				packet.Header.PayloadType = track.PayloadType()
+				packet.Header.SSRC = track.SSRC()
+				if writeErr := track.WriteRTP(packet); writeErr != nil {
+					log.Printf("Failed to write to track: %s", writeErr)
 					continue
 				}
 			}
@@ -105,24 +97,47 @@ func ingest(name string, q *messaging.Quality) {
 		log.Printf("Faited to wait for ffmpeg: %s", err)
 	}
 
-	// Close UDP listeners
-	if err = videoListener.Close(); err != nil {
+	// Close UDP listener
+	if err = listener.Close(); err != nil {
 		log.Printf("Faited to close UDP listener: %s", err)
 	}
-	if err = audioListener.Close(); err != nil {
-		log.Printf("Faited to close UDP listener: %s", err)
-	}
-	q.Unregister(videoInput)
+	q.Unregister(input)
 }
 
-func startFFmpeg(in <-chan []byte) (ffmpeg *exec.Cmd, err error) {
-	ffmpegArgs := []string{"-hide_banner", "-loglevel", "error", "-i", "pipe:0",
-		"-an", "-vcodec", "libvpx", "-crf", "10", "-cpu-used", "5", "-b:v", "6000k", "-maxrate", "8000k", "-bufsize", "12000k", // TODO Change bitrate when changing quality
-		"-qmin", "10", "-qmax", "42", "-threads", "4", "-deadline", "1", "-error-resilient", "1",
-		"-auto-alt-ref", "1",
-		"-f", "rtp", "rtp://127.0.0.1:5004",
-		"-vn", "-acodec", "libopus", "-cpu-used", "5", "-deadline", "1", "-qmin", "10", "-qmax", "42", "-error-resilient", "1", "-auto-alt-ref", "1",
-		"-f", "rtp", "rtp://127.0.0.1:5005"}
+func startFFmpeg(q *messaging.Quality, in <-chan []byte) (ffmpeg *exec.Cmd, err error) {
+	// FIXME Use transcoders to downscale, then remux in RTP
+	ffmpegArgs := []string{"-hide_banner", "-loglevel", "error", "-i", "pipe:0"}
+	switch q.Name {
+	case "audio":
+		ffmpegArgs = append(ffmpegArgs, "-vn", "-c:a", "libopus", "-b:a", "160k",
+			"-f", "rtp", "rtp://127.0.0.1:5004")
+		break
+	case "source":
+		ffmpegArgs = append(ffmpegArgs, "-an", "-c:v", "copy",
+			"-f", "rtp", "rtp://127.0.0.1:5005")
+		break
+	case "480p":
+		ffmpegArgs = append(ffmpegArgs,
+			"-an", "-c:v", "libx264", "-b:v", "1200k", "-maxrate", "2000k", "-bufsize", "3000k",
+			"-preset", "ultrafast", "-profile", "main", "-tune", "zerolatency",
+			"-vf", "scale=854:480",
+			"-f", "rtp", "rtp://127.0.0.1:5006")
+		break
+	case "360p":
+		ffmpegArgs = append(ffmpegArgs,
+			"-an", "-c:v", "libx264", "-b:v", "800k", "-maxrate", "1200k", "-bufsize", "1500k",
+			"-preset", "ultrafast", "-profile", "main", "-tune", "zerolatency",
+			"-vf", "scale=480:360",
+			"-f", "rtp", "rtp://127.0.0.1:5007")
+		break
+	case "240p":
+		ffmpegArgs = append(ffmpegArgs,
+			"-an", "-c:v", "libx264", "-b:v", "500k", "-maxrate", "800k", "-bufsize", "1000k",
+			"-preset", "ultrafast", "-profile", "main", "-tune", "zerolatency",
+			"-vf", "scale=360:240",
+			"-f", "rtp", "rtp://127.0.0.1:5008")
+		break
+	}
 	ffmpeg = exec.Command("ffmpeg", ffmpegArgs...)
 
 	// Handle errors output

stream/webrtc/webrtc.go

@@ -40,7 +40,7 @@ func removeTrack(tracks []*webrtc.Track, track *webrtc.Track) []*webrtc.Track {
 
 // GetNumberConnectedSessions get the number of currently connected clients
 func GetNumberConnectedSessions(streamID string) int {
-	return len(videoTracks[streamID])
+	return len(audioTracks[streamID])
 }
 
 // newPeerHandler is called when server receive a new session description
@@ -75,7 +75,7 @@ func newPeerHandler(name string, localSdpChan chan webrtc.SessionDescription, re
 	}
 
 	// Create video track
-	codec, payloadType := getPayloadType(mediaEngine, webrtc.RTPCodecTypeVideo, "VP8")
+	codec, payloadType := getPayloadType(mediaEngine, webrtc.RTPCodecTypeVideo, "H264")
 	videoTrack, err := webrtc.NewTrack(payloadType, rand.Uint32(), "video", "pion", codec)
 	if err != nil {
 		log.Println("Failed to create new video track", err)
@@ -117,21 +117,20 @@ func newPeerHandler(name string, localSdpChan chan webrtc.SessionDescription, re
 		quality = split[1]
 	}
 	log.Printf("New WebRTC session for stream %s, quality %s", streamID, quality)
-	// TODO Consider the quality
 
 	// Set the handler for ICE connection state
 	// This will notify you when the peer has connected/disconnected
 	peerConnection.OnICEConnectionStateChange(func(connectionState webrtc.ICEConnectionState) {
 		log.Printf("Connection State has changed %s \n", connectionState.String())
-		if videoTracks[streamID] == nil {
-			videoTracks[streamID] = make([]*webrtc.Track, 0, 1)
+		if videoTracks[streamID+"@"+quality] == nil {
+			videoTracks[streamID+"@"+quality] = make([]*webrtc.Track, 0, 1)
 		}
 		if audioTracks[streamID] == nil {
 			audioTracks[streamID] = make([]*webrtc.Track, 0, 1)
 		}
 		if connectionState == webrtc.ICEConnectionStateConnected {
 			// Register tracks
-			videoTracks[streamID] = append(videoTracks[streamID], videoTrack)
+			videoTracks[streamID+"@"+quality] = append(videoTracks[streamID+"@"+quality], videoTrack)
 			audioTracks[streamID] = append(audioTracks[streamID], audioTrack)
 			monitoring.WebRTCConnectedSessions.Inc()
 		} else if connectionState == webrtc.ICEConnectionStateDisconnected {
@@ -205,16 +204,17 @@ func Serve(streams *messaging.Streams, cfg *Options) {
 
 		// Get specific quality
 		// FIXME: make it possible to forward other qualities
-		qualityName := "source"
-		quality, err := stream.GetQuality(qualityName)
-		if err != nil {
-			log.Printf("Failed to get quality '%s'", qualityName)
-		}
+		for _, qualityName := range []string{"source", "audio", "480p", "360p", "240p"} {
+			quality, err := stream.GetQuality(qualityName)
+			if err != nil {
+				log.Printf("Failed to get quality '%s'", qualityName)
+			}
 
-		// Start forwarding
-		log.Printf("Starting webrtc for '%s' quality '%s'", name, qualityName)
-		go ingest(name, quality)
-		go listenSdp(name, quality.WebRtcLocalSdp, quality.WebRtcRemoteSdp, cfg)
+			// Start forwarding
+			log.Printf("Starting webrtc for '%s' quality '%s'", name, qualityName)
+			go ingest(name, quality)
+			go listenSdp(name, quality.WebRtcLocalSdp, quality.WebRtcRemoteSdp, cfg)
+		}
 	}
 }
 

stream/webrtc/webrtc_test.go

@@ -26,7 +26,7 @@ func TestServe(t *testing.T) {
 	peerConnection, _ := api.NewPeerConnection(webrtc.Configuration{})
 
 	// Create video track
-	codec, payloadType := getPayloadType(mediaEngine, webrtc.RTPCodecTypeVideo, "VP8")
+	codec, payloadType := getPayloadType(mediaEngine, webrtc.RTPCodecTypeVideo, "H264")
 	videoTrack, err := webrtc.NewTrack(payloadType, rand.Uint32(), "video", "pion", codec)
 	if err != nil {
 		t.Error("Failed to create new video track", err)

web/static/js/modules/webrtc.js

@@ -3,10 +3,12 @@
  */
 export class GsWebRTC {
     /**
-     * @param {list} stunServers 
-     * @param {HTMLElement} connectionIndicator 
+     * @param {list} stunServers STUN servers
+     * @param {HTMLElement} viewer Video HTML element
+     * @param {HTMLElement} connectionIndicator Connection indicator element
      */
-    constructor(stunServers, connectionIndicator) {
+    constructor(stunServers, viewer, connectionIndicator) {
+        this.viewer = viewer;
         this.connectionIndicator = connectionIndicator;
         this.pc = new RTCPeerConnection({
             iceServers: [{ urls: stunServers }]
@@ -26,7 +28,7 @@ export class GsWebRTC {
      * If connection closed or failed, try to reconnect.
      */
     _onConnectionStateChange() {
-        console.log("ICE connection state changed to " + this.pc.iceConnectionState);
+        console.log("[WebRTC] ICE connection state changed to " + this.pc.iceConnectionState);
         switch (this.pc.iceConnectionState) {
         case "disconnected":
             this.connectionIndicator.style.fill = "#dc3545";
@@ -39,7 +41,7 @@ export class GsWebRTC {
             break;
         case "closed":
         case "failed":
-            console.log("Connection closed, restarting...");
+            console.log("[WebRTC] Connection closed, restarting...");
             /*peerConnection.close();
                     peerConnection = null;
                     setTimeout(startPeerConnection, 1000);*/
@@ -52,10 +54,9 @@ export class GsWebRTC {
      * @param {Event} event 
      */
     _onTrack(event) {
-        console.log(`New ${event.track.kind} track`);
+        console.log(`[WebRTC] New ${event.track.kind} track`);
         if (event.track.kind === "video") {
-            const viewer = document.getElementById("viewer");
-            viewer.srcObject = event.streams[0];
+            this.viewer.srcObject = event.streams[0];
         }
     }
 
@@ -66,7 +67,7 @@ export class GsWebRTC {
     createOffer() {
         this.pc.createOffer().then(offer => {
             this.pc.setLocalDescription(offer);
-            console.log("WebRTC offer created");
+            console.log("[WebRTC] WebRTC offer created");
         }).catch(console.log);
     }
 
@@ -81,7 +82,7 @@ export class GsWebRTC {
         this.pc.onicecandidate = event => {
             if (event.candidate === null) {
                 // Send offer to server
-                console.log("Sending session description to server");
+                console.log("[WebRTC] Sending session description to server");
                 sendFunction(this.pc.localDescription);
             }
         };

web/static/js/modules/websocket.js

@@ -5,44 +5,42 @@ export class GsWebSocket {
     constructor() {
         const protocol = (window.location.protocol === "https:") ? "wss://" : "ws://";
         this.url = protocol + window.location.host + "/_ws/";
+
+        // Open WebSocket
+        this._open();
+
+        // Configure events
+        this.socket.addEventListener("open", () => {
+            console.log("[WebSocket] Connection established");
+        });
+        this.socket.addEventListener("close", () => {
+            console.log("[WebSocket] Connection closed, retrying connection in 1s...");
+            setTimeout(() => this._open(), 1000);
+        });
+        this.socket.addEventListener("error", () => {
+            console.log("[WebSocket] Connection errored, retrying connection in 1s...");
+            setTimeout(() => this._open(), 1000);
+        });
     }
 
     _open() {
+        console.log(`[WebSocket] Connecting to ${this.url}...`);
         this.socket = new WebSocket(this.url);
     }
 
     /**
-     * Open websocket.
-     * @param {Function} openCallback Function called when connection is established. 
-     * @param {Function} closeCallback Function called when connection is lost. 
-     */
-    open() {
-        this._open();
-        this.socket.addEventListener("open", () => {
-            console.log("WebSocket opened");
-        });
-        this.socket.addEventListener("close", () => {
-            console.log("WebSocket closed, retrying connection in 1s...");
-            setTimeout(() => this._open(), 1000);
-        });
-        this.socket.addEventListener("error", () => {
-            console.log("WebSocket errored, retrying connection in 1s...");
-            setTimeout(() => this._open(), 1000);
-        });
-    }
-
-    /**
-     * Exchange WebRTC session description with server.
+     * Send local WebRTC session description to remote.
      * @param {SessionDescription} localDescription WebRTC local SDP
      * @param {string} stream Name of the stream
      * @param {string} quality Requested quality 
      */
-    sendDescription(localDescription, stream, quality) {
+    sendLocalDescription(localDescription, stream, quality) {
         if (this.socket.readyState !== 1) {
-            console.log("Waiting for WebSocket to send data...");
-            setTimeout(() => this.sendDescription(localDescription, stream, quality), 100);
+            console.log("[WebSocket] Waiting for connection to send data...");
+            setTimeout(() => this.sendLocalDescription(localDescription, stream, quality), 100);
             return;
         }
+        console.log(`[WebSocket] Sending WebRTC local session description for stream ${stream} quality ${quality}`);
         this.socket.send(JSON.stringify({
             "webRtcSdp": localDescription,
             "stream": stream,
@@ -51,12 +49,12 @@ export class GsWebSocket {
     }
 
     /**
-     * Set callback function on new session description.
+     * Set callback function on new remote session description.
      * @param {Function} callback Function called when data is received
      */
-    onDescription(callback) {
+    onRemoteDescription(callback) {
         this.socket.addEventListener("message", (event) => {
-            console.log("Message from server ", event.data);
+            console.log("[WebSocket] Received WebRTC remote session description");
             const sdp = new RTCSessionDescription(JSON.parse(event.data));
             callback(sdp);
         });

web/static/js/viewer.js

@@ -10,28 +10,28 @@ import { GsWebRTC } from "./modules/webrtc.js";
  * @param {Number} viewersCounterRefreshPeriod 
  */
 export function initViewerPage(stream, stunServers, viewersCounterRefreshPeriod) {
+    // Viewer element
+    const viewer = document.getElementById("viewer");
+
     // Default quality
-    let quality = "source";
+    let quality = "240p";
 
-    // Create WebSocket
-    const s = new GsWebSocket();
-    s.open();
-
-    // Create WebRTC
-    const c = new GsWebRTC(
+    // Create WebSocket and WebRTC
+    const websocket = new GsWebSocket();
+    const webrtc = new GsWebRTC(
         stunServers,
+        viewer,
         document.getElementById("connectionIndicator"),
     );
-    c.createOffer();
-    c.onICECandidate(localDescription => {
-        s.sendDescription(localDescription, stream, quality);
+    webrtc.createOffer();
+    webrtc.onICECandidate(localDescription => {
+        websocket.sendLocalDescription(localDescription, stream, quality);
     });
-    s.onDescription(sdp => {
-        c.setRemoteDescription(sdp);
+    websocket.onRemoteDescription(sdp => {
+        webrtc.setRemoteDescription(sdp);
     });
 
     // Register keyboard events
-    const viewer = document.getElementById("viewer");
     window.addEventListener("keydown", (event) => {
         switch (event.key) {
         case "f":
@@ -81,7 +81,7 @@ export function initViewerPage(stream, stunServers, viewersCounterRefreshPeriod)
         quality = event.target.value;
         console.log(`Stream quality changed to ${quality}`);
 
-        // Restart the connection with a new quality
-        // FIXME
+        // Restart WebRTC negociation
+        webrtc.createOffer();
     });
 }

web/template/player.html

@@ -8,10 +8,10 @@
     <div class="controls">
       <span class="control-quality">
         <select id="quality">
-          <option value="">Source</option>
-          <option value="@720p">720p</option>
-          <option value="@480p">480p</option>
-          <option value="@240p">240p</option>
+          <option value="240p">Source</option>
+          <option value="480p">480p</option>
+          <option value="360p">360p</option>
+          <option value="240p">240p</option>
         </select>  
       </span>
       <code class="control-srt-link">srt://{{.Cfg.Hostname}}:{{.Cfg.SRTServerPort}}?streamid={{.Path}}</code>

web/websocket_handler.go

@@ -36,21 +36,21 @@ func websocketHandler(w http.ResponseWriter, r *http.Request) {
 		err = conn.ReadJSON(c)
 		if err != nil {
 			log.Printf("Failed to receive client description: %s", err)
-			return
+			continue
 		}
 
 		// Get requested stream
 		stream, err := streams.Get(c.Stream)
 		if err != nil {
 			log.Printf("Stream not found: %s", c.Stream)
-			return
+			continue
 		}
 
 		// Get requested quality
 		q, err := stream.GetQuality(c.Quality)
 		if err != nil {
 			log.Printf("Quality not found: %s", c.Quality)
-			return
+			continue
 		}
 
 		// Exchange session descriptions with WebRTC stream server
@@ -61,7 +61,7 @@ func websocketHandler(w http.ResponseWriter, r *http.Request) {
 		// Send new local description
 		if err := conn.WriteJSON(localDescription); err != nil {
 			log.Println(err)
-			return
+			continue
 		}
 	}
 }