Simulcast: Allow layer selection for publisher (#477)

* Simulcast: Allow layer selection for publisher

Changed ion-sfu channel messaging to be peer based and added aditional types

Addressing PR comments

Rename IonSfuMessage to ChannelAPIMessage. Use interface{} instead of string, fixes double marshalling issue

* Rebase on new interface changes

pkg/middlewares/datachannel/subscriberapi.go

@@ -3,23 +3,92 @@ package datachannel
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 
 	"github.com/pion/ion-sfu/pkg/sfu"
 	"github.com/pion/webrtc/v3"
 )
 
 const (
-	highValue   = "high"
-	mediumValue = "medium"
-	lowValue    = "low"
-	mutedValue  = "none"
+	highValue         = "high"
+	mediumValue       = "medium"
+	lowValue          = "low"
+	mutedValue        = "none"
+	ActiveLayerMethod = "activeLayer"
 )
 
 type setRemoteMedia struct {
-	StreamID  string `json:"streamId"`
-	Video     string `json:"video"`
-	Framerate string `json:"framerate"`
-	Audio     bool   `json:"audio"`
+	StreamID  string   `json:"streamId"`
+	Video     string   `json:"video"`
+	Framerate string   `json:"framerate"`
+	Audio     bool     `json:"audio"`
+	Layers    []string `json:"layers"`
+}
+
+type activeLayerMessage struct {
+	StreamID        string   `json:"streamId"`
+	ActiveLayer     string   `json:"activeLayer"`
+	AvailableLayers []string `json:"availableLayers"`
+}
+
+func layerStrToInt(layer string) (int, error) {
+	switch layer {
+	case highValue:
+		return 2, nil
+	case mediumValue:
+		return 1, nil
+	case lowValue:
+		return 0, nil
+	default:
+		// unknown value
+		return -1, fmt.Errorf("Unknown value")
+	}
+}
+
+func layerIntToStr(layer int) (string, error) {
+	switch layer {
+	case 0:
+		return lowValue, nil
+	case 1:
+		return mediumValue, nil
+	case 2:
+		return highValue, nil
+	default:
+		return "", fmt.Errorf("Unknown value: %d", layer)
+	}
+}
+
+func transformLayers(layers []string) ([]uint16, error) {
+	res := make([]uint16, len(layers))
+	for _, layer := range layers {
+		if l, err := layerStrToInt(layer); err == nil {
+			res = append(res, uint16(l))
+		} else {
+			return nil, fmt.Errorf("Unknown layer value: %v", layer)
+		}
+	}
+	return res, nil
+}
+
+func sendMessage(streamID string, peer sfu.Peer, layers []string, activeLayer int) {
+	al, _ := layerIntToStr(activeLayer)
+	payload := activeLayerMessage{
+		StreamID:        streamID,
+		ActiveLayer:     al,
+		AvailableLayers: layers,
+	}
+	msg := sfu.ChannelAPIMessage{
+		Method: ActiveLayerMethod,
+		Params: payload,
+	}
+	bytes, err := json.Marshal(msg)
+	if err != nil {
+		sfu.Logger.Error(err, "unable to marshal active layer message")
+	}
+
+	if err := peer.SendAPIChannelMessage(&bytes); err != nil {
+		sfu.Logger.Error(err, "unable to send ActiveLayerMessage to peer", "peer_id", peer.ID())
+	}
 }
 
 func SubscriberAPI(next sfu.MessageProcessor) sfu.MessageProcessor {
@@ -28,35 +97,59 @@ func SubscriberAPI(next sfu.MessageProcessor) sfu.MessageProcessor {
 		if err := json.Unmarshal(args.Message.Data, srm); err != nil {
 			return
 		}
-		downTracks := args.Peer.Subscriber().GetDownTracks(srm.StreamID)
-		for _, dt := range downTracks {
-			switch dt.Kind() {
-			case webrtc.RTPCodecTypeAudio:
-				dt.Mute(!srm.Audio)
-			case webrtc.RTPCodecTypeVideo:
-				switch srm.Video {
-				case highValue:
-					dt.Mute(false)
-					dt.SwitchSpatialLayer(2, true)
-				case mediumValue:
-					dt.Mute(false)
-					dt.SwitchSpatialLayer(1, true)
-				case lowValue:
-					dt.Mute(false)
-					dt.SwitchSpatialLayer(0, true)
-				case mutedValue:
-					dt.Mute(true)
-				}
-				switch srm.Framerate {
-				case highValue:
-					dt.SwitchTemporalLayer(2, true)
-				case mediumValue:
-					dt.SwitchTemporalLayer(1, true)
-				case lowValue:
-					dt.SwitchTemporalLayer(0, true)
+		// Publisher changing active layers
+		if srm.Layers != nil && len(srm.Layers) > 0 {
+			layers, err := transformLayers(srm.Layers)
+			if err != nil {
+				sfu.Logger.Error(err, "error reading layers")
+				next.Process(ctx, args)
+				return
+			}
+
+			session := args.Peer.Session()
+			peers := session.Peers()
+			for _, peer := range peers {
+				if peer.ID() != args.Peer.ID() {
+					downTracks := peer.Subscriber().GetDownTracks(srm.StreamID)
+					for _, dt := range downTracks {
+						if dt.Kind() == webrtc.RTPCodecTypeVideo {
+							newLayer, _ := dt.UptrackLayersChange(layers)
+							sendMessage(srm.StreamID, peer, srm.Layers, int(newLayer))
+						}
+					}
 				}
 			}
+		} else {
+			downTracks := args.Peer.Subscriber().GetDownTracks(srm.StreamID)
+			for _, dt := range downTracks {
+				switch dt.Kind() {
+				case webrtc.RTPCodecTypeAudio:
+					dt.Mute(!srm.Audio)
+				case webrtc.RTPCodecTypeVideo:
+					switch srm.Video {
+					case highValue:
+						dt.Mute(false)
+						dt.SwitchSpatialLayer(2, true)
+					case mediumValue:
+						dt.Mute(false)
+						dt.SwitchSpatialLayer(1, true)
+					case lowValue:
+						dt.Mute(false)
+						dt.SwitchSpatialLayer(0, true)
+					case mutedValue:
+						dt.Mute(true)
+					}
+					switch srm.Framerate {
+					case highValue:
+						dt.SwitchTemporalLayer(2, true)
+					case mediumValue:
+						dt.SwitchTemporalLayer(1, true)
+					case lowValue:
+						dt.SwitchTemporalLayer(0, true)
+					}
+				}
 
+			}
 		}
 		next.Process(ctx, args)
 	})

pkg/sfu/downtrack.go

@@ -1,6 +1,7 @@
 package sfu
 
 import (
+	"fmt"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -211,6 +212,40 @@ func (d *DownTrack) SwitchSpatialLayer(targetLayer int64, setAsMax bool) {
 	}
 }
 
+func (d *DownTrack) UptrackLayersChange(availableLayers []uint16) (int64, error) {
+	if d.trackType == SimulcastDownTrack {
+		currentLayer := uint16(atomic.LoadInt32(&d.spatialLayer))
+		maxLayer := uint16(atomic.LoadInt64(&d.maxSpatialLayer))
+
+		var maxFound uint16 = 0
+		layerFound := false
+		var minFound uint16 = 0
+		for _, target := range availableLayers {
+			if target <= maxLayer {
+				if target > maxFound {
+					maxFound = target
+					layerFound = true
+				}
+			} else {
+				if minFound > target {
+					minFound = target
+				}
+			}
+		}
+		var targetLayer uint16
+		if layerFound {
+			targetLayer = maxFound
+		} else {
+			targetLayer = minFound
+		}
+		if currentLayer != targetLayer {
+			d.SwitchSpatialLayer(int64(targetLayer), false)
+		}
+		return int64(targetLayer), nil
+	}
+	return -1, fmt.Errorf("Downtrack %s does not support simulcast", d.id)
+}
+
 func (d *DownTrack) SwitchTemporalLayer(targetLayer int64, setAsMax bool) {
 	if d.trackType == SimulcastDownTrack {
 		layer := atomic.LoadInt32(&d.temporalLayer)

pkg/sfu/peer.go

@@ -30,6 +30,7 @@ type Peer interface {
 	Publisher() *Publisher
 	Subscriber() *Subscriber
 	Close() error
+	SendAPIChannelMessage(msg *[]byte) error
 }
 
 // JoinConfig allow adding more control to the peers joining a SessionLocal.
@@ -48,6 +49,11 @@ type SessionProvider interface {
 	GetSession(sid string) (Session, WebRTCTransportConfig)
 }
 
+type ChannelAPIMessage struct {
+	Method string      `json:"method"`
+	Params interface{} `json:"params,omitempty"`
+}
+
 // PeerLocal represents a pair peer connection
 type PeerLocal struct {
 	sync.Mutex
@@ -241,6 +247,22 @@ func (p *PeerLocal) Trickle(candidate webrtc.ICECandidateInit, target int) error
 	return nil
 }
 
+func (p *PeerLocal) SendAPIChannelMessage(msg *[]byte) error {
+	if p.subscriber == nil {
+		return fmt.Errorf("No subscriber for this peer")
+	}
+	dc := p.subscriber.DataChannel(APIChannelLabel)
+
+	if dc == nil {
+		return fmt.Errorf("Data channel %s doesn't exist", APIChannelLabel)
+	}
+
+	if err := dc.SendText(string(*msg)); err != nil {
+		return fmt.Errorf("Failed to send message: %v", err)
+	}
+	return nil
+}
+
 // Close shuts down the peer connection and sends true to the done channel
 func (p *PeerLocal) Close() error {
 	p.Lock()

pkg/sfu/session.go

@@ -22,6 +22,7 @@ type Session interface {
 	GetDCMiddlewares() []*Datachannel
 	GetDataChannelLabels() []string
 	GetDataChannels(origin, label string) (dcs []*webrtc.DataChannel)
+	Peers() []Peer
 }
 
 type SessionLocal struct {
@@ -35,6 +36,10 @@ type SessionLocal struct {
 	onCloseHandler func()
 }
 
+const (
+	AudioLevelsMethod = "audioLevels"
+)
+
 // NewSession creates a new SessionLocal
 func NewSession(id string, dcs []*Datachannel, cfg WebRTCTransportConfig) Session {
 	s := &SessionLocal{
@@ -45,7 +50,6 @@ func NewSession(id string, dcs []*Datachannel, cfg WebRTCTransportConfig) Sessio
 	}
 	go s.audioLevelObserver(cfg.Router.AudioLevelInterval)
 	return s
-
 }
 
 // ID return SessionLocal id
@@ -251,7 +255,12 @@ func (s *SessionLocal) audioLevelObserver(audioLevelInterval int) {
 			continue
 		}
 
-		l, err := json.Marshal(&levels)
+		msg := ChannelAPIMessage{
+			Method: AudioLevelsMethod,
+			Params: levels,
+		}
+
+		l, err := json.Marshal(&msg)
 		if err != nil {
 			Logger.Error(err, "Marshaling audio levels err")
 			continue