Skip to content

Commit b4bd8ed

Browse files
fancycodefancycode
committed
Add tests for setting bandwidth.
1 parent d45fc47 commit b4bd8ed

3 files changed

Lines changed: 293 additions & 3 deletions

File tree

‎mcu_janus_test.go‎

Lines changed: 87 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@ import (
2525
"context"
2626
"encoding/json"
2727
"maps"
28+
"net/http/httptest"
2829
"strings"
2930
"sync"
3031
"sync/atomic"
@@ -2102,3 +2103,89 @@ func Test_JanusSubscriberUpdateOffer(t *testing.T) {
21022103
// Test MCU will trigger an updated offer.
21032104
client2.RunUntilOffer(ctx, MockSdpOfferAudioOnly)
21042105
}
2106+
2107+
func Test_JanusSetBandwidth(t *testing.T) {
2108+
t.Parallel()
2109+
require := require.New(t)
2110+
assert := assert.New(t)
2111+
2112+
var counter atomic.Int32
2113+
mcu, gateway := newMcuJanusForTesting(t)
2114+
gateway.registerHandlers(map[string]TestJanusHandler{
2115+
"configure": func(room *TestJanusRoom, body, jsep api.StringMap) (any, *janus.ErrorMsg) {
2116+
assert.EqualValues(1, room.id)
2117+
switch counter.Add(1) {
2118+
case 1:
2119+
return &janus.EventMsg{
2120+
Jsep: api.StringMap{
2121+
"type": "answer",
2122+
"sdp": MockSdpAnswerAudioAndVideo,
2123+
},
2124+
}, nil
2125+
case 2:
2126+
// When the room updates the bandwidth per publisher, it will notify
2127+
// Janus about it.
2128+
assert.EqualValues(500000, body["bitrate"], "got %+v", body)
2129+
return &janus.EventMsg{}, nil
2130+
default:
2131+
assert.Fail("too many configure requests", "received body=%+v, jsep=%+v", body, jsep)
2132+
return &janus.ErrorMsg{
2133+
Err: janus.ErrorData{
2134+
Code: JANUS_ERROR_UNKNOWN,
2135+
Reason: "too many configure requests",
2136+
},
2137+
}, nil
2138+
}
2139+
},
2140+
})
2141+
2142+
hub, _, _, server := CreateHubForTestWithConfig(t, func(s *httptest.Server) (*goconf.ConfigFile, error) {
2143+
config, err := getTestConfig(s)
2144+
if err != nil {
2145+
return nil, err
2146+
}
2147+
2148+
config.AddOption("backend", "maxstreambitrate", "700000")
2149+
config.AddOption("backend", "maxscreenbitrate", "800000")
2150+
2151+
config.AddOption("backend", "bitrateperroom", "1000000")
2152+
config.AddOption("backend", "minpublisherbitrate", "10000")
2153+
config.AddOption("backend", "maxpublisherbitrate", "500000")
2154+
return config, err
2155+
})
2156+
hub.SetMcu(mcu)
2157+
2158+
ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
2159+
defer cancel()
2160+
2161+
client, hello := NewTestClientWithHello(ctx, t, server, hub, testDefaultUserId+"1")
2162+
2163+
// Join room by id.
2164+
roomId := "test-room"
2165+
roomMsg := MustSucceed2(t, client.JoinRoom, ctx, roomId)
2166+
require.Equal(roomId, roomMsg.Room.RoomId)
2167+
client.RunUntilJoined(ctx, hello.Hello)
2168+
2169+
require.NoError(client.SendMessage(MessageClientMessageRecipient{
2170+
Type: "session",
2171+
SessionId: hello.Hello.SessionId,
2172+
}, MessageClientMessageData{
2173+
Type: "offer",
2174+
RoomType: "video",
2175+
Payload: api.StringMap{
2176+
"sdp": MockSdpOfferAudioAndVideo,
2177+
},
2178+
}))
2179+
2180+
client.RunUntilAnswer(ctx, MockSdpAnswerAudioAndVideo)
2181+
2182+
pub, err := mcu.getPublisher(ctx, hello.Hello.SessionId, StreamTypeVideo)
2183+
require.NoError(err)
2184+
pub.UpdateBandwidth("video", api.BandwidthFromBits(2000), api.BandwidthFromBits(100000))
2185+
2186+
room := hub.getRoom(roomId)
2187+
require.NotNil(room)
2188+
room.updateBandwidth().Wait()
2189+
2190+
assert.EqualValues(2, counter.Load())
2191+
}

‎mcu_proxy_test.go‎

Lines changed: 159 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,7 @@ import (
4949
"github.com/stretchr/testify/require"
5050
"go.etcd.io/etcd/server/v3/embed"
5151

52+
"github.com/strukturag/nextcloud-spreed-signaling/api"
5253
"github.com/strukturag/nextcloud-spreed-signaling/log"
5354
)
5455

@@ -198,7 +199,8 @@ func Test_sortConnectionsForCountryWithOverride(t *testing.T) {
198199
type proxyServerClientHandler func(msg *ProxyClientMessage) (*ProxyServerMessage, error)
199200

200201
type testProxyServerPublisher struct {
201-
id PublicSessionId
202+
id PublicSessionId
203+
bandwidth api.AtomicBandwidth
202204
}
203205

204206
type testProxyServerSubscriber struct {
@@ -298,6 +300,8 @@ func (c *testProxyServerClient) processRegularMessage(msg *ProxyClientMessage) (
298300
switch msg.Type {
299301
case "command":
300302
handler = c.processCommandMessage
303+
case "payload":
304+
handler = c.processPayloadMessage
301305
}
302306

303307
if handler == nil {
@@ -431,6 +435,21 @@ func (c *testProxyServerClient) processCommandMessage(msg *ProxyClientMessage) (
431435
}
432436
c.server.updateLoad(-1)
433437
}
438+
case "update-bandwidth":
439+
pub := c.server.getPublisher(PublicSessionId(msg.Command.ClientId))
440+
if pub == nil {
441+
response = msg.NewWrappedErrorServerMessage(fmt.Errorf("publisher %s not found", msg.Command.ClientId))
442+
return response, nil
443+
}
444+
445+
pub.bandwidth.Store(msg.Command.Bandwidth)
446+
response = &ProxyServerMessage{
447+
Id: msg.Id,
448+
Type: "command",
449+
Command: &CommandProxyServerMessage{
450+
Id: string(pub.id),
451+
},
452+
}
434453
}
435454
if response == nil {
436455
response = msg.NewWrappedErrorServerMessage(fmt.Errorf("command \"%s\" is not implemented", msg.Command.Type))
@@ -439,6 +458,36 @@ func (c *testProxyServerClient) processCommandMessage(msg *ProxyClientMessage) (
439458
return response, nil
440459
}
441460

461+
func (c *testProxyServerClient) processPayloadMessage(msg *ProxyClientMessage) (*ProxyServerMessage, error) {
462+
var response *ProxyServerMessage
463+
switch msg.Payload.Type {
464+
case "offer":
465+
pub := c.server.getPublisher(PublicSessionId(msg.Payload.ClientId))
466+
if pub == nil {
467+
response = msg.NewWrappedErrorServerMessage(fmt.Errorf("no such publisher: %s", msg.Payload.ClientId))
468+
return response, nil
469+
}
470+
471+
assert.Equal(c.t, MockSdpOfferAudioAndVideo, msg.Payload.Payload["sdp"])
472+
response = &ProxyServerMessage{
473+
Id: msg.Id,
474+
Type: "payload",
475+
Payload: &PayloadProxyServerMessage{
476+
ClientId: string(pub.id),
477+
Type: "answer",
478+
Payload: api.StringMap{
479+
"type": "answer",
480+
"sdp": MockSdpAnswerAudioAndVideo,
481+
},
482+
},
483+
}
484+
default:
485+
response = msg.NewWrappedErrorServerMessage(fmt.Errorf("payload type \"%s\" is not implemented", msg.Payload.Type))
486+
}
487+
488+
return response, nil
489+
}
490+
442491
func (c *testProxyServerClient) close() {
443492
c.mu.Lock()
444493
defer c.mu.Unlock()
@@ -2598,3 +2647,112 @@ func Test_ProxyResumeFail(t *testing.T) {
25982647
assert.NotEqual(sessionId, connections[0].SessionId())
25992648
}
26002649
}
2650+
2651+
func Test_ProxySetBandwidth(t *testing.T) {
2652+
t.Parallel()
2653+
require := require.New(t)
2654+
assert := assert.New(t)
2655+
server := NewProxyServerForTest(t, "DE")
2656+
mcu, _ := newMcuProxyForTestWithOptions(t, proxyTestOptions{
2657+
servers: []*TestProxyServerHandler{server},
2658+
}, 0, nil)
2659+
2660+
hub, _, _, hubserver := CreateHubForTestWithConfig(t, func(s *httptest.Server) (*goconf.ConfigFile, error) {
2661+
config, err := getTestConfig(s)
2662+
if err != nil {
2663+
return nil, err
2664+
}
2665+
2666+
config.AddOption("backend", "maxstreambitrate", "700000")
2667+
config.AddOption("backend", "maxscreenbitrate", "800000")
2668+
2669+
config.AddOption("backend", "bitrateperroom", "1000000")
2670+
config.AddOption("backend", "minpublisherbitrate", "10000")
2671+
config.AddOption("backend", "maxpublisherbitrate", "500000")
2672+
return config, err
2673+
})
2674+
hub.SetMcu(mcu)
2675+
2676+
ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
2677+
defer cancel()
2678+
2679+
client, hello := NewTestClientWithHello(ctx, t, hubserver, hub, testDefaultUserId+"1")
2680+
2681+
// Join room by id.
2682+
roomId := "test-room"
2683+
roomMsg := MustSucceed2(t, client.JoinRoom, ctx, roomId)
2684+
require.Equal(roomId, roomMsg.Room.RoomId)
2685+
client.RunUntilJoined(ctx, hello.Hello)
2686+
2687+
require.NoError(client.SendMessage(MessageClientMessageRecipient{
2688+
Type: "session",
2689+
SessionId: hello.Hello.SessionId,
2690+
}, MessageClientMessageData{
2691+
Type: "offer",
2692+
RoomType: "video",
2693+
Payload: api.StringMap{
2694+
"sdp": MockSdpOfferAudioAndVideo,
2695+
},
2696+
}))
2697+
2698+
client.RunUntilAnswer(ctx, MockSdpAnswerAudioAndVideo)
2699+
2700+
pub := mcu.getPublisherConnection(hello.Hello.SessionId, StreamTypeVideo)
2701+
require.NotNil(pub)
2702+
2703+
var publisherId string
2704+
var publisher *mcuProxyPublisher
2705+
pub.publishersLock.RLock()
2706+
if assert.Len(pub.publishers, 1) {
2707+
for id, mcuPub := range pub.publishers {
2708+
publisherId = id
2709+
publisher = mcuPub
2710+
break
2711+
}
2712+
}
2713+
pub.publishersLock.RUnlock()
2714+
require.NotEmpty(publisherId)
2715+
require.NotNil(publisher)
2716+
2717+
proxyclient := server.GetSingleClient()
2718+
proxyclient.sendMessage(&ProxyServerMessage{
2719+
Type: "event",
2720+
Event: &EventProxyServerMessage{
2721+
Type: "update-load",
2722+
Load: 1,
2723+
ClientBandwidths: map[string]EventProxyServerBandwidth{
2724+
publisherId: {
2725+
Sent: api.BandwidthFromBits(0),
2726+
Received: api.BandwidthFromBits(100000),
2727+
},
2728+
},
2729+
},
2730+
})
2731+
2732+
for publisher.Bandwidth() == nil {
2733+
if !assert.NoError(ctx.Err()) {
2734+
break
2735+
}
2736+
2737+
time.Sleep(time.Millisecond)
2738+
}
2739+
if bw := publisher.Bandwidth(); assert.NotNil(bw) {
2740+
assert.EqualValues(0, bw.Sent)
2741+
assert.EqualValues(100000, bw.Received)
2742+
}
2743+
2744+
room := hub.getRoom(roomId)
2745+
require.NotNil(room)
2746+
room.updateBandwidth().Wait()
2747+
2748+
proxyPub := server.getPublisher(PublicSessionId(publisherId))
2749+
require.NotNil(proxyPub)
2750+
for proxyPub.bandwidth.Load() == 0 {
2751+
if !assert.NoError(ctx.Err()) {
2752+
break
2753+
}
2754+
2755+
time.Sleep(time.Millisecond)
2756+
}
2757+
assert.EqualValues(500000, proxyPub.bandwidth.Load())
2758+
}

‎proxy/proxy_server_test.go‎

Lines changed: 47 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -452,20 +452,30 @@ func (p *TestMCUPublisher) UnpublishRemote(ctx context.Context, remoteId signali
452452

453453
type PublisherTestMCU struct {
454454
TestMCU
455+
456+
publisher atomic.Pointer[TestPublisherWithBandwidth]
455457
}
456458

457459
type TestPublisherWithBandwidth struct {
458460
TestMCUPublisher
459461

460-
bandwidth *signaling.McuClientBandwidthInfo
462+
t *testing.T
463+
bandwidth *signaling.McuClientBandwidthInfo
464+
bandwidthSet atomic.Bool
461465
}
462466

463467
func (p *TestPublisherWithBandwidth) Bandwidth() *signaling.McuClientBandwidthInfo {
464468
return p.bandwidth
465469
}
466470

467471
func (p *TestPublisherWithBandwidth) SetBandwidth(ctx context.Context, bandwidth api.Bandwidth) error {
468-
return errors.New("not implemented")
472+
assert.EqualValues(p.t, 20000, bandwidth)
473+
p.bandwidthSet.Store(true)
474+
return nil
475+
}
476+
477+
func (m *PublisherTestMCU) GetPublisher() *TestPublisherWithBandwidth {
478+
return m.publisher.Load()
469479
}
470480

471481
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) {
@@ -476,11 +486,16 @@ func (m *PublisherTestMCU) NewPublisher(ctx context.Context, listener signaling.
476486
streamType: streamType,
477487
},
478488

489+
t: m.t,
479490
bandwidth: &signaling.McuClientBandwidthInfo{
480491
Sent: api.BandwidthFromBytes(1000),
481492
Received: api.BandwidthFromBytes(2000),
482493
},
483494
}
495+
if !m.publisher.CompareAndSwap(nil, publisher) {
496+
return nil, errors.New("only one publisher supported")
497+
}
498+
484499
return publisher, nil
485500
}
486501

@@ -528,12 +543,17 @@ func TestProxyPublisherBandwidth(t *testing.T) {
528543
},
529544
}))
530545

546+
var publisherId string
531547
if message, err := client.RunUntilMessage(ctx); assert.NoError(err) {
532548
assert.Equal("2345", message.Id)
533549
if err := checkMessageType(message, "command"); assert.NoError(err) {
534550
assert.NotEmpty(message.Command.Id)
551+
publisherId = message.Command.Id
535552
}
536553
}
554+
require.NotEmpty(publisherId)
555+
publisher := mcu.GetPublisher()
556+
require.NotNil(publisher)
537557

538558
proxy.updateLoad()
539559

@@ -548,8 +568,33 @@ func TestProxyPublisherBandwidth(t *testing.T) {
548568
assert.InEpsilon(10, *bw.Outgoing, 0.0001)
549569
}
550570
}
571+
if assert.Len(message.Event.ClientBandwidths, 1) {
572+
if bw := message.Event.ClientBandwidths; assert.NotNil(bw[publisherId], "expected %s, got %+v", bw) {
573+
assert.EqualValues(8000, bw[publisherId].Sent)
574+
assert.EqualValues(16000, bw[publisherId].Received)
575+
}
576+
}
577+
}
578+
}
579+
580+
require.NoError(client.WriteJSON(&signaling.ProxyClientMessage{
581+
Id: "3456",
582+
Type: "command",
583+
Command: &signaling.CommandProxyClientMessage{
584+
Type: "update-bandwidth",
585+
ClientId: publisherId,
586+
Bandwidth: api.BandwidthFromBits(20000),
587+
},
588+
}))
589+
590+
if message, err := client.RunUntilMessage(ctx); assert.NoError(err) {
591+
assert.Equal("3456", message.Id)
592+
if err := checkMessageType(message, "command"); assert.NoError(err) {
593+
assert.Equal(publisherId, message.Command.Id)
551594
}
552595
}
596+
597+
assert.True(publisher.bandwidthSet.Load(), "should have set bandwidth")
553598
}
554599

555600
type HangingTestMCU struct {

0 commit comments

Comments
 (0)