fix offer/answer handling

.vscode/launch.json

@@ -19,6 +19,7 @@
         "INPUT_DEVICE": "/dev/input/event5",
         "JINGLE_PATH": "${workspaceFolder}/audio",
         "SONOS_TARGET": "Living Room",
+        "VIDEO_SRC_CODEC": "h264"
       }
     },
     {

pkg/server/http.go

@@ -10,7 +10,6 @@ import (
 	"github.com/gin-gonic/gin"
 	"github.com/kaedwen/webrtc/pkg/common"
 	"github.com/kaedwen/webrtc/static"
-	"github.com/pion/webrtc/v3"
 	"go.uber.org/zap"
 	"nhooyr.io/websocket"
 	"nhooyr.io/websocket/wsjson"
@@ -18,8 +17,8 @@ import (
 
 type SignalingHandle struct {
 	Id   string
-	Recv chan webrtc.SessionDescription
+	Recv chan *IncomingSignalingMessage
-	Trcv chan webrtc.SessionDescription
+	Trcv chan *OutgoingSignalingMessage
 }
 
 type HttpServer struct {
@@ -32,8 +31,8 @@ type HttpServer struct {
 func NewSignalingHandle(id string) SignalingHandle {
 	return SignalingHandle{
 		Id:   id,
-		Recv: make(chan webrtc.SessionDescription, 10),
+		Recv: make(chan *IncomingSignalingMessage, 10),
-		Trcv: make(chan webrtc.SessionDescription, 10),
+		Trcv: make(chan *OutgoingSignalingMessage, 10),
 	}
 }
 
@@ -143,8 +142,8 @@ func (h *HttpServer) signalingHandler(c *gin.Context) {
 	go func() {
 		for {
 			// wait and read/parse message
-			var sdp webrtc.SessionDescription
+			var m IncomingSignalingMessage
-			err := wsjson.Read(ctx, conn, &sdp)
+			err := wsjson.Read(ctx, conn, &m)
 			if err != nil {
 				status := websocket.CloseStatus(err)
 				if status == websocket.StatusNormalClosure || status == websocket.StatusGoingAway {
@@ -161,18 +160,18 @@ func (h *HttpServer) signalingHandler(c *gin.Context) {
 				break
 			}
 
-			h.lg.Info("received message", zap.String("type", sdp.Type.String()))
+			h.lg.Info("received message", zap.String("type", m.Type))
 
 			// forward in channel
-			hndl.Recv <- sdp
+			hndl.Recv <- &m
 		}
 
 		cancel()
 	}()
 
 	// loop write
-	for sdp := range hndl.Trcv {
+	for m := range hndl.Trcv {
-		err := wsjson.Write(ctx, conn, sdp)
+		err := wsjson.Write(ctx, conn, m)
 		if err != nil {
 			status := websocket.CloseStatus(err)
 			if status == websocket.StatusNormalClosure || status == websocket.StatusGoingAway {
@@ -184,7 +183,7 @@ func (h *HttpServer) signalingHandler(c *gin.Context) {
 			break
 		}
 
-		h.lg.Info("tranceived message", zap.String("type", sdp.Type.String()))
+		h.lg.Info("tranceived message", zap.String("type", m.Type))
 	}
 
 }

pkg/server/model.go

@@ -0,0 +1,101 @@
+package server
+
+import (
+	"encoding/json"
+
+	"github.com/pion/webrtc/v3"
+)
+
+type SignalingMessageType = string
+
+const (
+	MessageTypeIceCandidate SignalingMessageType = "new-ice-candidate"
+	MessageTypeAnswer       SignalingMessageType = "answer"
+	MessageTypeOffer        SignalingMessageType = "offer"
+)
+
+// INCOMING
+
+type IncomingSignalingMessage struct {
+	Type SignalingMessageType `json:"type"`
+	Data json.RawMessage      `json:"data"`
+}
+
+type IceCandidateMessage struct {
+	*IncomingSignalingMessage
+	Candidate webrtc.ICECandidateInit
+}
+
+type OfferMessage struct {
+	*IncomingSignalingMessage
+	Offer webrtc.SessionDescription
+}
+
+type AnswerMessage struct {
+	*IncomingSignalingMessage
+	Answer webrtc.SessionDescription
+}
+
+func (m *IncomingSignalingMessage) IsIceCandidateMessage() bool {
+	return m.Type == MessageTypeIceCandidate
+}
+
+func (m *IncomingSignalingMessage) IsAnswerMessage() bool {
+	return m.Type == MessageTypeAnswer
+}
+
+func (m *IncomingSignalingMessage) IsOfferMessage() bool {
+	return m.Type == MessageTypeOffer
+}
+
+func (m *IncomingSignalingMessage) ToIceCandidateMessage() (*IceCandidateMessage, error) {
+	nm := IceCandidateMessage{
+		IncomingSignalingMessage: m,
+	}
+
+	return &nm, json.Unmarshal(m.Data, &nm.Candidate)
+}
+
+func (m *IncomingSignalingMessage) ToAnswerMessage() (*AnswerMessage, error) {
+	nm := AnswerMessage{
+		IncomingSignalingMessage: m,
+	}
+
+	return &nm, json.Unmarshal(m.Data, &nm.Answer)
+}
+
+func (m *IncomingSignalingMessage) ToOfferMessage() (*OfferMessage, error) {
+	nm := OfferMessage{
+		IncomingSignalingMessage: m,
+	}
+
+	return &nm, json.Unmarshal(m.Data, &nm.Offer)
+}
+
+// OUTGOING
+
+type OutgoingSignalingMessage struct {
+	Type SignalingMessageType `json:"type"`
+	Data any                  `json:"data"`
+}
+
+func NewIceCandidateMessage(candidate webrtc.ICECandidate) *OutgoingSignalingMessage {
+	return &OutgoingSignalingMessage{
+		Type: MessageTypeIceCandidate,
+		Data: candidate,
+	}
+}
+
+func NewAnswerMessage(answer *webrtc.SessionDescription) *OutgoingSignalingMessage {
+	return &OutgoingSignalingMessage{
+		Type: MessageTypeAnswer,
+		Data: answer,
+	}
+}
+
+func NewOfferMessage(offer *webrtc.SessionDescription) *OutgoingSignalingMessage {
+	return &OutgoingSignalingMessage{
+		Type: MessageTypeOffer,
+		Data: offer,
+	}
+}

pkg/streamer/gst-src.go

@@ -47,7 +47,7 @@ func CreateAudioPipelineSrc(dst StreamElement) (*SrcPipeline, error) {
 	sink := elems[len(elems)-1]
 
 	for name, value := range dst.Properties {
-		sink.SetProperty(name, value)
+		sink.Set(name, value)
 	}
 
 	// Add the elements to the pipeline and link them

pkg/webrtc/webrtc.go

@@ -221,6 +221,13 @@ func (wh *WebrtcHandler) createPeerHandle(rctx context.Context, sh *server.Signa
 	// create a context for this handle
 	hctx, hcancel := context.WithCancel(rctx)
 
+	// sent the candidate out when where is one
+	peerConnection.OnICECandidate(func(i *webrtc.ICECandidate) {
+		if i != nil {
+			sh.Trcv <- server.NewIceCandidateMessage(*i)
+		}
+	})
+
 	// Set a handler for when a new remote track starts, this handler creates a gstreamer pipeline
 	// for the given codec
 	peerConnection.OnTrack(func(track *webrtc.TrackRemote, receiver *webrtc.RTPReceiver) {
@@ -263,11 +270,9 @@ func (wh *WebrtcHandler) createPeerHandle(rctx context.Context, sh *server.Signa
 		}()
 
 		pipeline, err := streamer.CreateAudioPipelineSrc(streamer.StreamElement{
-			Kind: cfg.Sink,
+			Kind:       cfg.Sink,
-			Properties: map[string]interface{}{
+			Properties: properties,
-				"device": cfg.Device,
+			Queue:      cfg.Queue,
-			},
-			Queue: cfg.Queue,
 		})
 		if err != nil {
 			wh.lg.Error("failed to create src pipeline", zap.Error(err))
@@ -358,7 +363,7 @@ func (wh *WebrtcHandler) createPeerHandle(rctx context.Context, sh *server.Signa
 		}
 
 		// Create channel that is blocked until ICE Gathering is complete
-		gatherComplete := webrtc.GatheringCompletePromise(peerConnection)
+		//gatherComplete := webrtc.GatheringCompletePromise(peerConnection)
 
 		// Sets the LocalDescription, and starts our UDP listeners
 		err = peerConnection.SetLocalDescription(answer)
@@ -366,23 +371,23 @@ func (wh *WebrtcHandler) createPeerHandle(rctx context.Context, sh *server.Signa
 			return err
 		}
 
-		<-gatherComplete
+		//<-gatherComplete
 
 		// Send the answer
-		sh.Trcv <- *peerConnection.LocalDescription()
+		sh.Trcv <- server.NewAnswerMessage(peerConnection.LocalDescription())
 
 		return nil
 	}
 
-	hndl := PeerHandle{
+	hndl := PeerHandle{audioTrack, videoTrack}
-		audioTrack: audioTrack,
+
-		videoTrack: videoTrack,
+	// add handle to list
-	}
+	wh.peerHandles[sh.Id] = &hndl
 
 	go func() {
 		for {
 			select {
-			case offer, ok := <-sh.Recv:
+			case m, ok := <-sh.Recv:
 				// return when channel is closed
 				if !ok {
 					wh.mu.Lock()
@@ -399,9 +404,35 @@ func (wh *WebrtcHandler) createPeerHandle(rctx context.Context, sh *server.Signa
 					return
 				}
 
-				err := onOfferReceived(offer)
+				switch true {
-				if err != nil {
+				case m.IsIceCandidateMessage():
-					wh.lg.Error("failed to handle offer", zap.Error(err))
+					pm, err := m.ToIceCandidateMessage()
+					if err != nil {
+						wh.lg.Error("failed to parse candidate", zap.Error(err))
+						continue
+					}
+
+					wh.lg.Info("received new ice candidate", zap.Any("candidate", pm.Candidate))
+
+					err = peerConnection.AddICECandidate(pm.Candidate)
+					if err != nil {
+						wh.lg.Error("failed to add ice candidate", zap.Error(err))
+					}
+				case m.IsAnswerMessage():
+					wh.lg.Info("received answer", zap.Any("data", m.Data))
+				case m.IsOfferMessage():
+					pm, err := m.ToOfferMessage()
+					if err != nil {
+						wh.lg.Error("failed to parse offer", zap.Error(err))
+						continue
+					}
+
+					wh.lg.Info("received offer", zap.Any("data", pm.Offer))
+
+					err = onOfferReceived(pm.Offer)
+					if err != nil {
+						wh.lg.Error("failed to handle offer", zap.Error(err))
+					}
 				}
 			case <-hctx.Done():
 				return
@@ -409,8 +440,5 @@ func (wh *WebrtcHandler) createPeerHandle(rctx context.Context, sh *server.Signa
 		}
 	}()
 
-	// add handle to list
-	wh.peerHandles[sh.Id] = &hndl
-
 	return nil
 }

static/dist/index.html

@@ -1,13 +1,13 @@
 <!doctype html>
-<html lang="en" data-critters-container>
+<html lang="en">
 <head>
   <meta charset="utf-8">
   <title>Webrtc</title>
   <base href="/">
   <meta name="viewport" content="width=device-width, initial-scale=1">
   <link rel="icon" type="image/x-icon" href="favicon.ico">
-<style>html,body{width:100%;height:100%;margin:0}body{background-color:#2e2e2e}</style><link rel="stylesheet" href="styles.89648d27bd81920d.css" media="print" onload="this.media='all'"><noscript><link rel="stylesheet" href="styles.89648d27bd81920d.css"></noscript></head>
+<link rel="stylesheet" href="styles.css"></head>
 <body>
   <app-root></app-root>
-<script src="runtime.c87b550f0b12139c.js" type="module"></script><script src="polyfills.c1402358c38e8522.js" type="module"></script><script src="main.74d7f4c379afaac2.js" type="module"></script></body>
+<script src="runtime.js" type="module"></script><script src="polyfills.js" type="module"></script><script src="vendor.js" type="module"></script><script src="main.js" type="module"></script></body>
 </html>

static/src/app/app.component.ts

@@ -2,6 +2,7 @@ import { Component, ComponentRef, HostListener, OnInit, ViewChild, ViewContainer
 import { VideoComponent } from './components/video/video.component';
 import { AudioComponent } from './components/audio/audio.component';
 import { SignalingService } from './services/signaling.service';
+import { IsAnswer, IsIceCandidate, IsOffer } from './model';
 
 @Component({
   selector: 'app-root',
@@ -9,7 +10,7 @@ import { SignalingService } from './services/signaling.service';
   styleUrls: ['./app.component.scss']
 })
 export class AppComponent implements OnInit {
-  @ViewChild("vc", {static: true, read: ViewContainerRef }) 
+  @ViewChild("vc", { static: true, read: ViewContainerRef })
   public vcr!: ViewContainerRef;
 
   private audioList: ComponentRef<AudioComponent>[] = [];
@@ -22,7 +23,7 @@ export class AppComponent implements OnInit {
     if (!this.selfAudioRunning) {
       await this.startAudio()
     }
-  
+
     for (const el of this.audioList) {
       el.instance.toggleMute();
     }
@@ -44,14 +45,14 @@ export class AppComponent implements OnInit {
     this.pc = new RTCPeerConnection();
 
     // Once remote track media arrives, show it in remote element.
-    this.pc.ontrack = (event) => {
+    this.pc.ontrack = (e) => {
-      const [remoteStream] = event.streams;
+      const [remoteStream] = e.streams;
-      if (event.track.kind === 'video') {
+      if (e.track.kind === 'video') {
         const video = this.vcr.createComponent(VideoComponent, {});
         video.instance.setStream(remoteStream);
         video.instance.play();
         console.log(video);
-      } else if (event.track.kind === 'audio') {
+      } else if (e.track.kind === 'audio') {
         const audio = this.vcr.createComponent(AudioComponent, {});
         audio.instance.setStream(remoteStream);
         console.log(audio);
@@ -61,52 +62,77 @@ export class AppComponent implements OnInit {
     }
 
     this.pc.oniceconnectionstatechange = (e) => {
-      console.log(e);
+      console.log('ICE: state change', e);
     };
 
+    this.pc.onicegatheringstatechange = (e) => {
+      switch (this.pc.iceGatheringState) {
+        case "new":
+          console.log('ICE: gathering is either just starting or has been reset', e);
+          break;
+        case "gathering":
+          console.log('ICE: gathering has begun or is ongoing', e);
+          break;
+        case "complete":
+          console.log('ICE: gathering has ended', e);
+          break;
+      }
+    }
+
     // Send any ice candidates to the other peer.
-    this.pc.onicecandidate = ({candidate}) => {
+    this.pc.onicecandidate = ({ candidate }) => {
-      console.log(candidate);
+      console.log('ICE: candidate', candidate);
-      if (candidate == null) {
+      if (candidate !== null) {
-        signaling.next(this.pc.localDescription)
+        signaling.SendCandidate(candidate);
+      } else {
+        /* there are no more candidates coming during this negotiation */
       }
     };
 
     // Offer to receive 1 audio, and 1 video track
-    this.pc.addTransceiver('audio', {direction: 'sendrecv'})
+    this.pc.addTransceiver('audio', { direction: 'sendrecv' })
-    this.pc.addTransceiver('video', {direction: 'recvonly'})
+    this.pc.addTransceiver('video', { direction: 'recvonly' })
-    this.pc.createOffer().then(d => this.pc.setLocalDescription(d)).catch((e) => {
+
-      console.error(e);
-    });
-    
     // Let the "negotiationneeded" event trigger offer generation.
-    this.pc.onnegotiationneeded = async (e) => {
+    this.pc.onnegotiationneeded = (e) => {
-      try {
+      this.pc.createOffer()
-        await this.pc.setLocalDescription();
+        .then(async (d) => {
-        signaling.next(this.pc.localDescription);
+          await this.pc.setLocalDescription(d)
-      } catch (err) {
+          return this.pc.localDescription!
-        console.error(err);
+        })
-      }
+        .then((d) => {
+          console.log('SDP: sending offer', d);
+          signaling.SendOffer(d)
+        })
+        .catch((e) => {
+          console.error(e);
+        });
     };
 
-    this.signaling.subscribe(async (x) => {
+    this.signaling.subscribe((m) => {
-      console.log(x);
+      switch (true) {
-      if (x.type === 'answer') {
+        case IsIceCandidate(m):
-        console.log('Received answer');
+          console.log('ICE: received new candidate', m.data);
+          break;
+        case IsOffer(m):
+          console.log('SDP: received offer', m.data);
+          break;
+        case IsAnswer(m):
+          console.log('SDP: received answer', m.data);
 
-        try {
+          this.pc.setRemoteDescription(m.data).catch((e) => {
-          // the answer
+            console.error(e);
-          await this.pc.setRemoteDescription(x);
+          });
-        } catch(e) {
-          console.error(e);
-        }
 
+          break;
+        default:
+          console.log('WARN: received unknown data', m);
       }
     });
   }
 
   public async ngOnInit(): Promise<void> {
-    
+
   }
 
 }

static/src/app/model.ts

@@ -0,0 +1,33 @@
+
+export interface SignalingMessage {
+  type: 'new-ice-candidate' | 'offer' | 'answer';
+  data: any;
+}
+
+export interface AnswerMessage extends SignalingMessage {
+  data: RTCSessionDescriptionInit;
+}
+
+export interface OfferMessage extends SignalingMessage {
+  data: RTCSessionDescriptionInit;
+}
+
+export interface IceCandidateMessage extends SignalingMessage {
+  data: RTCIceCandidate;
+}
+
+export const IsSignalingMessage = (d: any): d is SignalingMessage => {
+  return !!d && typeof(d.type) === 'string';
+}
+
+export const IsIceCandidate = (d: any): d is IceCandidateMessage => {
+  return IsSignalingMessage(d) && d.type === 'new-ice-candidate';
+}
+
+export const IsAnswer = (d: any): d is AnswerMessage => {
+  return IsSignalingMessage(d) && d.type === 'answer';
+}
+
+export const IsOffer = (d: any): d is OfferMessage => {
+  return IsSignalingMessage(d) && d.type === 'offer';
+}

static/src/app/services/signaling.service.ts

@@ -1,11 +1,12 @@
 import { WebSocketSubject, WebSocketSubjectConfig } from 'rxjs/webSocket';
+import { SignalingMessage } from '../model';
 import { Injectable } from '@angular/core';
 import { v4 } from 'uuid';
 
 @Injectable({
   providedIn: 'root'
 })
-export class SignalingService extends WebSocketSubject<any> {
+export class SignalingService extends WebSocketSubject<SignalingMessage> {
   constructor() {
     const config: WebSocketSubjectConfig<any> = {
       url: `${location.origin.replace("http", "ws")}/signaling/${v4()}`,
@@ -13,4 +14,18 @@ export class SignalingService extends WebSocketSubject<any> {
 
     super(config);
   }
+
+  public SendCandidate(candidate: RTCIceCandidate) {
+    this.next({
+      type: 'new-ice-candidate',
+      data: candidate,
+    });
+  }
+
+  public SendOffer(offer: RTCSessionDescriptionInit) {
+    this.next({
+      type: 'offer',
+      data: offer,
+    });
+  }
 }