Add tests for setting bandwidth.

Author: fancycode

mcu_janus_test.go

@@ -25,6 +25,7 @@ import (
 	"context"
 	"encoding/json"
 	"maps"
+	"net/http/httptest"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -2017,3 +2018,88 @@ func Test_JanusSubscriberUpdateOffer(t *testing.T) {
 	// Test MCU will trigger an updated offer.
 	client2.RunUntilOffer(ctx, MockSdpOfferAudioOnly)
 }
+
+func Test_JanusSetBandwidth(t *testing.T) {
+	require := require.New(t)
+	assert := assert.New(t)
+
+	var counter atomic.Int32
+	mcu, gateway := newMcuJanusForTesting(t)
+	gateway.registerHandlers(map[string]TestJanusHandler{
+		"configure": func(room *TestJanusRoom, body, jsep api.StringMap) (any, *janus.ErrorMsg) {
+			assert.EqualValues(1, room.id)
+			switch counter.Add(1) {
+			case 1:
+				return &janus.EventMsg{
+					Jsep: api.StringMap{
+						"type": "answer",
+						"sdp":  MockSdpAnswerAudioAndVideo,
+					},
+				}, nil
+			case 2:
+				// When the room updates the bandwidth per publisher, it will notify
+				// Janus about it.
+				assert.EqualValues(500000, body["bitrate"], "got %+v", body)
+				return &janus.EventMsg{}, nil
+			default:
+				assert.Fail("too many configure requests", "received body=%+v, jsep=%+v", body, jsep)
+				return &janus.ErrorMsg{
+					Err: janus.ErrorData{
+						Code:   JANUS_ERROR_UNKNOWN,
+						Reason: "too many configure requests",
+					},
+				}, nil
+			}
+		},
+	})
+
+	hub, _, _, server := CreateHubForTestWithConfig(t, func(s *httptest.Server) (*goconf.ConfigFile, error) {
+		config, err := getTestConfig(s)
+		if err != nil {
+			return nil, err
+		}
+
+		config.AddOption("backend", "maxstreambitrate", "700000")
+		config.AddOption("backend", "maxscreenbitrate", "800000")
+
+		config.AddOption("backend", "bitrateperroom", "1000000")
+		config.AddOption("backend", "minpublisherbitrate", "10000")
+		config.AddOption("backend", "maxpublisherbitrate", "500000")
+		return config, err
+	})
+	hub.SetMcu(mcu)
+
+	ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
+	defer cancel()
+
+	client, hello := NewTestClientWithHello(ctx, t, server, hub, testDefaultUserId+"1")
+
+	// Join room by id.
+	roomId := "test-room"
+	roomMsg := MustSucceed2(t, client.JoinRoom, ctx, roomId)
+	require.Equal(roomId, roomMsg.Room.RoomId)
+	client.RunUntilJoined(ctx, hello.Hello)
+
+	require.NoError(client.SendMessage(MessageClientMessageRecipient{
+		Type:      "session",
+		SessionId: hello.Hello.SessionId,
+	}, MessageClientMessageData{
+		Type:     "offer",
+		RoomType: "video",
+		Payload: api.StringMap{
+			"sdp": MockSdpOfferAudioAndVideo,
+		},
+	}))
+
+	client.RunUntilAnswer(ctx, MockSdpAnswerAudioAndVideo)
+
+	pub, err := mcu.getPublisher(ctx, hello.Hello.SessionId, StreamTypeVideo)
+	require.NoError(err)
+	pub.UpdateBandwidth("video", api.BandwidthFromBits(2000), api.BandwidthFromBits(100000))
+
+	room := hub.getRoom(roomId)
+	require.NotNil(room)
+	room.updateBandwidth().Wait()
+
+	assert.EqualValues(2, counter.Load())
+}

mcu_proxy_test.go

@@ -47,6 +47,8 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.etcd.io/etcd/server/v3/embed"
+
+	"github.com/strukturag/nextcloud-spreed-signaling/api"
 )
 
 const (
@@ -190,7 +192,8 @@ func Test_sortConnectionsForCountryWithOverride(t *testing.T) {
 type proxyServerClientHandler func(msg *ProxyClientMessage) (*ProxyServerMessage, error)
 
 type testProxyServerPublisher struct {
-	id PublicSessionId
+	id        PublicSessionId
+	bandwidth api.AtomicBandwidth
 }
 
 type testProxyServerSubscriber struct {
@@ -290,6 +293,8 @@ func (c *testProxyServerClient) processRegularMessage(msg *ProxyClientMessage) (
 	switch msg.Type {
 	case "command":
 		handler = c.processCommandMessage
+	case "payload":
+		handler = c.processPayloadMessage
 	}
 
 	if handler == nil {
@@ -423,6 +428,21 @@ func (c *testProxyServerClient) processCommandMessage(msg *ProxyClientMessage) (
 			}
 			c.server.updateLoad(-1)
 		}
+	case "update-bandwidth":
+		pub := c.server.getPublisher(PublicSessionId(msg.Command.ClientId))
+		if pub == nil {
+			response = msg.NewWrappedErrorServerMessage(fmt.Errorf("publisher %s not found", msg.Command.ClientId))
+			return response, nil
+		}
+
+		pub.bandwidth.Store(msg.Command.Bandwidth)
+		response = &ProxyServerMessage{
+			Id:   msg.Id,
+			Type: "command",
+			Command: &CommandProxyServerMessage{
+				Id: string(pub.id),
+			},
+		}
 	}
 	if response == nil {
 		response = msg.NewWrappedErrorServerMessage(fmt.Errorf("command \"%s\" is not implemented", msg.Command.Type))
@@ -431,6 +451,36 @@ func (c *testProxyServerClient) processCommandMessage(msg *ProxyClientMessage) (
 	return response, nil
 }
 
+func (c *testProxyServerClient) processPayloadMessage(msg *ProxyClientMessage) (*ProxyServerMessage, error) {
+	var response *ProxyServerMessage
+	switch msg.Payload.Type {
+	case "offer":
+		pub := c.server.getPublisher(PublicSessionId(msg.Payload.ClientId))
+		if pub == nil {
+			response = msg.NewWrappedErrorServerMessage(fmt.Errorf("no such publisher: %s", msg.Payload.ClientId))
+			return response, nil
+		}
+
+		assert.Equal(c.t, MockSdpOfferAudioAndVideo, msg.Payload.Payload["sdp"])
+		response = &ProxyServerMessage{
+			Id:   msg.Id,
+			Type: "payload",
+			Payload: &PayloadProxyServerMessage{
+				ClientId: string(pub.id),
+				Type:     "answer",
+				Payload: api.StringMap{
+					"type": "answer",
+					"sdp":  MockSdpAnswerAudioAndVideo,
+				},
+			},
+		}
+	default:
+		response = msg.NewWrappedErrorServerMessage(fmt.Errorf("payload type \"%s\" is not implemented", msg.Payload.Type))
+	}
+
+	return response, nil
+}
+
 func (c *testProxyServerClient) close() {
 	c.mu.Lock()
 	defer c.mu.Unlock()
@@ -2588,3 +2638,112 @@ func Test_ProxyResumeFail(t *testing.T) {
 		assert.NotEqual(sessionId, connections[0].SessionId())
 	}
 }
+
+func Test_ProxySetBandwidth(t *testing.T) {
+	t.Parallel()
+	require := require.New(t)
+	assert := assert.New(t)
+	server := NewProxyServerForTest(t, "DE")
+	mcu, _ := newMcuProxyForTestWithOptions(t, proxyTestOptions{
+		servers: []*TestProxyServerHandler{server},
+	}, 0)
+
+	hub, _, _, hubserver := CreateHubForTestWithConfig(t, func(s *httptest.Server) (*goconf.ConfigFile, error) {
+		config, err := getTestConfig(s)
+		if err != nil {
+			return nil, err
+		}
+
+		config.AddOption("backend", "maxstreambitrate", "700000")
+		config.AddOption("backend", "maxscreenbitrate", "800000")
+
+		config.AddOption("backend", "bitrateperroom", "1000000")
+		config.AddOption("backend", "minpublisherbitrate", "10000")
+		config.AddOption("backend", "maxpublisherbitrate", "500000")
+		return config, err
+	})
+	hub.SetMcu(mcu)
+
+	ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
+	defer cancel()
+
+	client, hello := NewTestClientWithHello(ctx, t, hubserver, hub, testDefaultUserId+"1")
+
+	// Join room by id.
+	roomId := "test-room"
+	roomMsg := MustSucceed2(t, client.JoinRoom, ctx, roomId)
+	require.Equal(roomId, roomMsg.Room.RoomId)
+	client.RunUntilJoined(ctx, hello.Hello)
+
+	require.NoError(client.SendMessage(MessageClientMessageRecipient{
+		Type:      "session",
+		SessionId: hello.Hello.SessionId,
+	}, MessageClientMessageData{
+		Type:     "offer",
+		RoomType: "video",
+		Payload: api.StringMap{
+			"sdp": MockSdpOfferAudioAndVideo,
+		},
+	}))
+
+	client.RunUntilAnswer(ctx, MockSdpAnswerAudioAndVideo)
+
+	pub := mcu.getPublisherConnection(hello.Hello.SessionId, StreamTypeVideo)
+	require.NotNil(pub)
+
+	var publisherId string
+	var publisher *mcuProxyPublisher
+	pub.publishersLock.RLock()
+	if assert.Len(pub.publishers, 1) {
+		for id, mcuPub := range pub.publishers {
+			publisherId = id
+			publisher = mcuPub
+			break
+		}
+	}
+	pub.publishersLock.RUnlock()
+	require.NotEmpty(publisherId)
+	require.NotNil(publisher)
+
+	proxyclient := server.GetSingleClient()
+	proxyclient.sendMessage(&ProxyServerMessage{
+		Type: "event",
+		Event: &EventProxyServerMessage{
+			Type: "update-load",
+			Load: 1,
+			ClientBandwidths: map[string]EventProxyServerBandwidth{
+				publisherId: {
+					Sent:     api.BandwidthFromBits(0),
+					Received: api.BandwidthFromBits(100000),
+				},
+			},
+		},
+	})
+
+	for publisher.Bandwidth() == nil {
+		if !assert.NoError(ctx.Err()) {
+			break
+		}
+
+		time.Sleep(time.Millisecond)
+	}
+	if bw := publisher.Bandwidth(); assert.NotNil(bw) {
+		assert.EqualValues(0, bw.Sent)
+		assert.EqualValues(100000, bw.Received)
+	}
+
+	room := hub.getRoom(roomId)
+	require.NotNil(room)
+	room.updateBandwidth().Wait()
+
+	proxyPub := server.getPublisher(PublicSessionId(publisherId))
+	require.NotNil(proxyPub)
+	for proxyPub.bandwidth.Load() == 0 {
+		if !assert.NoError(ctx.Err()) {
+			break
+		}
+
+		time.Sleep(time.Millisecond)
+	}
+	assert.EqualValues(500000, proxyPub.bandwidth.Load())
+}

proxy/proxy_server_test.go

@@ -443,20 +443,30 @@ func (p *TestMCUPublisher) UnpublishRemote(ctx context.Context, remoteId signali
 
 type PublisherTestMCU struct {
 	TestMCU
+
+	publisher atomic.Pointer[TestPublisherWithBandwidth]
 }
 
 type TestPublisherWithBandwidth struct {
 	TestMCUPublisher
 
-	bandwidth *signaling.McuClientBandwidthInfo
+	t            *testing.T
+	bandwidth    *signaling.McuClientBandwidthInfo
+	bandwidthSet atomic.Bool
 }
 
 func (p *TestPublisherWithBandwidth) Bandwidth() *signaling.McuClientBandwidthInfo {
 	return p.bandwidth
 }
 
 func (p *TestPublisherWithBandwidth) SetBandwidth(ctx context.Context, bandwidth api.Bandwidth) error {
-	return errors.New("not implemented")
+	assert.EqualValues(p.t, 20000, bandwidth)
+	p.bandwidthSet.Store(true)
+	return nil
+}
+
+func (m *PublisherTestMCU) GetPublisher() *TestPublisherWithBandwidth {
+	return m.publisher.Load()
 }
 
 func (m *PublisherTestMCU) NewPublisher(ctx context.Context, listener signaling.McuListener, id signaling.PublicSessionId, sid string, streamType signaling.StreamType, settings signaling.NewPublisherSettings, initiator signaling.McuInitiator) (signaling.McuPublisher, error) {
@@ -467,11 +477,16 @@ func (m *PublisherTestMCU) NewPublisher(ctx context.Context, listener signaling.
 			streamType: streamType,
 		},
 
+		t: m.t,
 		bandwidth: &signaling.McuClientBandwidthInfo{
 			Sent:     api.BandwidthFromBytes(1000),
 			Received: api.BandwidthFromBytes(2000),
 		},
 	}
+	if !m.publisher.CompareAndSwap(nil, publisher) {
+		return nil, errors.New("only one publisher supported")
+	}
+
 	return publisher, nil
 }
 
@@ -518,12 +533,17 @@ func TestProxyPublisherBandwidth(t *testing.T) {
 		},
 	}))
 
+	var publisherId string
 	if message, err := client.RunUntilMessage(ctx); assert.NoError(err) {
 		assert.Equal("2345", message.Id)
 		if err := checkMessageType(message, "command"); assert.NoError(err) {
 			assert.NotEmpty(message.Command.Id)
+			publisherId = message.Command.Id
 		}
 	}
+	require.NotEmpty(publisherId)
+	publisher := mcu.GetPublisher()
+	require.NotNil(publisher)
 
 	proxy.updateLoad()
 
@@ -538,8 +558,33 @@ func TestProxyPublisherBandwidth(t *testing.T) {
 					assert.EqualValues(10, *bw.Outgoing)
 				}
 			}
+			if assert.Len(message.Event.ClientBandwidths, 1) {
+				if bw := message.Event.ClientBandwidths; assert.NotNil(bw[publisherId], "expected %s, got %+v", bw) {
+					assert.EqualValues(8000, bw[publisherId].Sent)
+					assert.EqualValues(16000, bw[publisherId].Received)
+				}
+			}
+		}
+	}
+
+	require.NoError(client.WriteJSON(&signaling.ProxyClientMessage{
+		Id:   "3456",
+		Type: "command",
+		Command: &signaling.CommandProxyClientMessage{
+			Type:      "update-bandwidth",
+			ClientId:  publisherId,
+			Bandwidth: api.BandwidthFromBits(20000),
+		},
+	}))
+
+	if message, err := client.RunUntilMessage(ctx); assert.NoError(err) {
+		assert.Equal("3456", message.Id)
+		if err := checkMessageType(message, "command"); assert.NoError(err) {
+			assert.Equal(publisherId, message.Command.Id)
 		}
 	}
+
+	assert.True(publisher.bandwidthSet.Load(), "should have set bandwidth")
 }
 
 type HangingTestMCU struct {