From b6fc9f559bcdcbc79848f4193174e85d570041d5 Mon Sep 17 00:00:00 2001 From: Joachim Bauch Date: Mon, 11 Nov 2024 10:17:25 +0100 Subject: [PATCH] Add more cases when to stop remote publishing. --- proxy/proxy_server.go | 12 + proxy/proxy_server_test.go | 485 +++++++++++++++++++++++++++++++++++++ proxy/proxy_session.go | 81 +++++++ 3 files changed, 578 insertions(+) diff --git a/proxy/proxy_server.go b/proxy/proxy_server.go index 0e6634dc..b256d099 100644 --- a/proxy/proxy_server.go +++ b/proxy/proxy_server.go @@ -1163,6 +1163,7 @@ func (s *ProxyServer) processCommand(ctx context.Context, client *ProxyClient, s } } + session.AddRemotePublisher(publisher, cmd.Hostname, cmd.Port, cmd.RtcpPort) response := &signaling.ProxyServerMessage{ Id: message.Id, Type: "command", @@ -1193,6 +1194,8 @@ func (s *ProxyServer) processCommand(ctx context.Context, client *ProxyClient, s return } + session.RemoveRemotePublisher(publisher, cmd.Hostname, cmd.Port, cmd.RtcpPort) + response := &signaling.ProxyServerMessage{ Id: message.Id, Type: "command", @@ -1599,3 +1602,12 @@ func (s *ProxyServer) getRemoteConnection(url string) (*RemoteConnection, error) s.remoteConnections[url] = conn return conn, nil } + +func (s *ProxyServer) PublisherDeleted(publisher signaling.McuPublisher) { + s.sessionsLock.RLock() + defer s.sessionsLock.RUnlock() + + for _, session := range s.sessions { + session.OnPublisherDeleted(publisher) + } +} diff --git a/proxy/proxy_server_test.go b/proxy/proxy_server_test.go index cb6f42f0..1cb82247 100644 --- a/proxy/proxy_server_test.go +++ b/proxy/proxy_server_test.go @@ -835,3 +835,488 @@ func TestProxyRemoteSubscriber(t *testing.T) { } } } + +func TestProxyCloseRemoteOnSessionClose(t *testing.T) { + signaling.CatchLogForTest(t) + assert := assert.New(t) + require := require.New(t) + proxy, key, server := newProxyServerForTest(t) + + mcu := NewRemoteSubscriberTestMCU(t) + proxy.mcu = mcu + // Unused but must be set so remote subscribing works + proxy.tokenId = "token" + proxy.tokenKey = key + proxy.remoteHostname = "test-hostname" + + ctx, cancel := context.WithTimeout(context.Background(), testTimeout) + defer cancel() + + client := NewProxyTestClient(ctx, t, server.URL) + defer client.CloseWithBye() + + require.NoError(client.SendHello(key)) + + if hello, err := client.RunUntilHello(ctx); assert.NoError(err) { + assert.NotEmpty(hello.Hello.SessionId, "%+v", hello) + } + + _, err := client.RunUntilLoad(ctx, 0) + assert.NoError(err) + + publisherId := "the-publisher-id" + claims := &signaling.TokenClaims{ + RegisteredClaims: jwt.RegisteredClaims{ + IssuedAt: jwt.NewNumericDate(time.Now().Add(-maxTokenAge / 2)), + Issuer: TokenIdForTest, + Subject: publisherId, + }, + } + token := jwt.NewWithClaims(jwt.SigningMethodRS256, claims) + tokenString, err := token.SignedString(key) + require.NoError(err) + + require.NoError(client.WriteJSON(&signaling.ProxyClientMessage{ + Id: "2345", + Type: "command", + Command: &signaling.CommandProxyClientMessage{ + Type: "create-subscriber", + StreamType: signaling.StreamTypeVideo, + PublisherId: publisherId, + RemoteUrl: "https://remote-hostname", + RemoteToken: tokenString, + }, + })) + + if message, err := client.RunUntilMessage(ctx); assert.NoError(err) { + assert.Equal("2345", message.Id) + if err := checkMessageType(message, "command"); assert.NoError(err) { + require.NotEmpty(message.Command.Id) + } + } + + // Closing the session will cause any active remote publishers stop be stopped. + client.CloseWithBye() + + if assert.NotNil(mcu.publisher) && assert.NotNil(mcu.subscriber) { + select { + case <-mcu.subscriber.closed.Done(): + case <-ctx.Done(): + assert.Fail("subscriber was not closed") + } + select { + case <-mcu.publisher.closed.Done(): + case <-ctx.Done(): + assert.Fail("publisher was not closed") + } + } +} + +type UnpublishRemoteTestMCU struct { + TestMCU + + publisher *UnpublishRemoteTestPublisher +} + +func NewUnpublishRemoteTestMCU(t *testing.T) *UnpublishRemoteTestMCU { + return &UnpublishRemoteTestMCU{ + TestMCU: TestMCU{ + t: t, + }, + } +} + +type UnpublishRemoteTestPublisher struct { + TestMCUPublisher + + t *testing.T + + remoteId string + remoteData *remotePublisherData +} + +func (m *UnpublishRemoteTestMCU) NewPublisher(ctx context.Context, listener signaling.McuListener, id string, sid string, streamType signaling.StreamType, settings signaling.NewPublisherSettings, initiator signaling.McuInitiator) (signaling.McuPublisher, error) { + publisher := &UnpublishRemoteTestPublisher{ + TestMCUPublisher: TestMCUPublisher{ + id: id, + sid: sid, + streamType: streamType, + }, + + t: m.t, + } + m.publisher = publisher + return publisher, nil +} + +func (p *UnpublishRemoteTestPublisher) PublishRemote(ctx context.Context, remoteId string, hostname string, port int, rtcpPort int) error { + if assert.Empty(p.t, p.remoteId) { + p.remoteId = remoteId + p.remoteData = &remotePublisherData{ + hostname: hostname, + port: port, + rtcpPort: rtcpPort, + } + } + return nil +} + +func (p *UnpublishRemoteTestPublisher) UnpublishRemote(ctx context.Context, remoteId string, hostname string, port int, rtcpPort int) error { + assert.Equal(p.t, remoteId, p.remoteId) + if assert.NotNil(p.t, p.remoteData) && + assert.Equal(p.t, p.remoteData.hostname, hostname) && + assert.EqualValues(p.t, p.remoteData.port, port) && + assert.EqualValues(p.t, p.remoteData.rtcpPort, rtcpPort) { + p.remoteId = "" + p.remoteData = nil + } + return nil +} + +func TestProxyUnpublishRemote(t *testing.T) { + signaling.CatchLogForTest(t) + assert := assert.New(t) + require := require.New(t) + proxy, key, server := newProxyServerForTest(t) + + mcu := NewUnpublishRemoteTestMCU(t) + proxy.mcu = mcu + + ctx, cancel := context.WithTimeout(context.Background(), testTimeout) + defer cancel() + + client1 := NewProxyTestClient(ctx, t, server.URL) + defer client1.CloseWithBye() + + require.NoError(client1.SendHello(key)) + + if hello, err := client1.RunUntilHello(ctx); assert.NoError(err) { + assert.NotEmpty(hello.Hello.SessionId, "%+v", hello) + } + + _, err := client1.RunUntilLoad(ctx, 0) + assert.NoError(err) + + publisherId := "the-publisher-id" + require.NoError(client1.WriteJSON(&signaling.ProxyClientMessage{ + Id: "2345", + Type: "command", + Command: &signaling.CommandProxyClientMessage{ + Type: "create-publisher", + PublisherId: publisherId, + Sid: "1234-abcd", + StreamType: signaling.StreamTypeVideo, + PublisherSettings: &signaling.NewPublisherSettings{ + Bitrate: 1234567, + MediaTypes: signaling.MediaTypeAudio | signaling.MediaTypeVideo, + }, + }, + })) + + var clientId string + if message, err := client1.RunUntilMessage(ctx); assert.NoError(err) { + assert.Equal("2345", message.Id) + if err := checkMessageType(message, "command"); assert.NoError(err) { + require.NotEmpty(message.Command.Id) + clientId = message.Command.Id + } + } + + client2 := NewProxyTestClient(ctx, t, server.URL) + defer client2.CloseWithBye() + + require.NoError(client2.SendHello(key)) + + hello2, err := client2.RunUntilHello(ctx) + if assert.NoError(err) { + assert.NotEmpty(hello2.Hello.SessionId, "%+v", hello2) + } + + _, err = client2.RunUntilLoad(ctx, 0) + assert.NoError(err) + + require.NoError(client2.WriteJSON(&signaling.ProxyClientMessage{ + Id: "3456", + Type: "command", + Command: &signaling.CommandProxyClientMessage{ + Type: "publish-remote", + StreamType: signaling.StreamTypeVideo, + ClientId: clientId, + Hostname: "remote-host", + Port: 10001, + RtcpPort: 10002, + }, + })) + + if message, err := client2.RunUntilMessage(ctx); assert.NoError(err) { + assert.Equal("3456", message.Id) + if err := checkMessageType(message, "command"); assert.NoError(err) { + require.NotEmpty(message.Command.Id) + } + } + + if assert.NotNil(mcu.publisher) { + assert.Equal(hello2.Hello.SessionId, mcu.publisher.remoteId) + if assert.NotNil(mcu.publisher.remoteData) { + assert.Equal("remote-host", mcu.publisher.remoteData.hostname) + assert.EqualValues(10001, mcu.publisher.remoteData.port) + assert.EqualValues(10002, mcu.publisher.remoteData.rtcpPort) + } + } + + require.NoError(client2.WriteJSON(&signaling.ProxyClientMessage{ + Id: "4567", + Type: "command", + Command: &signaling.CommandProxyClientMessage{ + Type: "unpublish-remote", + StreamType: signaling.StreamTypeVideo, + ClientId: clientId, + Hostname: "remote-host", + Port: 10001, + RtcpPort: 10002, + }, + })) + + if message, err := client2.RunUntilMessage(ctx); assert.NoError(err) { + assert.Equal("4567", message.Id) + if err := checkMessageType(message, "command"); assert.NoError(err) { + require.NotEmpty(message.Command.Id) + } + } + + if assert.NotNil(mcu.publisher) { + assert.Empty(mcu.publisher.remoteId) + assert.Nil(mcu.publisher.remoteData) + } +} + +func TestProxyUnpublishRemotePublisherClosed(t *testing.T) { + signaling.CatchLogForTest(t) + assert := assert.New(t) + require := require.New(t) + proxy, key, server := newProxyServerForTest(t) + + mcu := NewUnpublishRemoteTestMCU(t) + proxy.mcu = mcu + + ctx, cancel := context.WithTimeout(context.Background(), testTimeout) + defer cancel() + + client1 := NewProxyTestClient(ctx, t, server.URL) + defer client1.CloseWithBye() + + require.NoError(client1.SendHello(key)) + + if hello, err := client1.RunUntilHello(ctx); assert.NoError(err) { + assert.NotEmpty(hello.Hello.SessionId, "%+v", hello) + } + + _, err := client1.RunUntilLoad(ctx, 0) + assert.NoError(err) + + publisherId := "the-publisher-id" + require.NoError(client1.WriteJSON(&signaling.ProxyClientMessage{ + Id: "2345", + Type: "command", + Command: &signaling.CommandProxyClientMessage{ + Type: "create-publisher", + PublisherId: publisherId, + Sid: "1234-abcd", + StreamType: signaling.StreamTypeVideo, + PublisherSettings: &signaling.NewPublisherSettings{ + Bitrate: 1234567, + MediaTypes: signaling.MediaTypeAudio | signaling.MediaTypeVideo, + }, + }, + })) + + var clientId string + if message, err := client1.RunUntilMessage(ctx); assert.NoError(err) { + assert.Equal("2345", message.Id) + if err := checkMessageType(message, "command"); assert.NoError(err) { + require.NotEmpty(message.Command.Id) + clientId = message.Command.Id + } + } + + client2 := NewProxyTestClient(ctx, t, server.URL) + defer client2.CloseWithBye() + + require.NoError(client2.SendHello(key)) + + hello2, err := client2.RunUntilHello(ctx) + if assert.NoError(err) { + assert.NotEmpty(hello2.Hello.SessionId, "%+v", hello2) + } + + _, err = client2.RunUntilLoad(ctx, 0) + assert.NoError(err) + + require.NoError(client2.WriteJSON(&signaling.ProxyClientMessage{ + Id: "3456", + Type: "command", + Command: &signaling.CommandProxyClientMessage{ + Type: "publish-remote", + StreamType: signaling.StreamTypeVideo, + ClientId: clientId, + Hostname: "remote-host", + Port: 10001, + RtcpPort: 10002, + }, + })) + + if message, err := client2.RunUntilMessage(ctx); assert.NoError(err) { + assert.Equal("3456", message.Id) + if err := checkMessageType(message, "command"); assert.NoError(err) { + require.NotEmpty(message.Command.Id) + } + } + + if assert.NotNil(mcu.publisher) { + assert.Equal(hello2.Hello.SessionId, mcu.publisher.remoteId) + if assert.NotNil(mcu.publisher.remoteData) { + assert.Equal("remote-host", mcu.publisher.remoteData.hostname) + assert.EqualValues(10001, mcu.publisher.remoteData.port) + assert.EqualValues(10002, mcu.publisher.remoteData.rtcpPort) + } + } + + require.NoError(client1.WriteJSON(&signaling.ProxyClientMessage{ + Id: "4567", + Type: "command", + Command: &signaling.CommandProxyClientMessage{ + Type: "delete-publisher", + ClientId: clientId, + }, + })) + + if message, err := client1.RunUntilMessage(ctx); assert.NoError(err) { + assert.Equal("4567", message.Id) + if err := checkMessageType(message, "command"); assert.NoError(err) { + require.NotEmpty(message.Command.Id) + } + } + + // Remote publishing was not stopped explicitly... + if assert.NotNil(mcu.publisher) { + assert.Equal(hello2.Hello.SessionId, mcu.publisher.remoteId) + if assert.NotNil(mcu.publisher.remoteData) { + assert.Equal("remote-host", mcu.publisher.remoteData.hostname) + assert.EqualValues(10001, mcu.publisher.remoteData.port) + assert.EqualValues(10002, mcu.publisher.remoteData.rtcpPort) + } + } + + // ...but the session no longer contains information on the remote publisher. + if data, err := proxy.cookie.DecodePublic(hello2.Hello.SessionId); assert.NoError(err) { + session := proxy.GetSession(data.Sid) + if assert.NotNil(session) { + session.remotePublishersLock.Lock() + defer session.remotePublishersLock.Unlock() + assert.Empty(session.remotePublishers) + } + } + + mcu.publisher.remoteId = "" + mcu.publisher.remoteData = nil +} + +func TestProxyUnpublishRemoteOnSessionClose(t *testing.T) { + signaling.CatchLogForTest(t) + assert := assert.New(t) + require := require.New(t) + proxy, key, server := newProxyServerForTest(t) + + mcu := NewUnpublishRemoteTestMCU(t) + proxy.mcu = mcu + + ctx, cancel := context.WithTimeout(context.Background(), testTimeout) + defer cancel() + + client1 := NewProxyTestClient(ctx, t, server.URL) + defer client1.CloseWithBye() + + require.NoError(client1.SendHello(key)) + + if hello, err := client1.RunUntilHello(ctx); assert.NoError(err) { + assert.NotEmpty(hello.Hello.SessionId, "%+v", hello) + } + + _, err := client1.RunUntilLoad(ctx, 0) + assert.NoError(err) + + publisherId := "the-publisher-id" + require.NoError(client1.WriteJSON(&signaling.ProxyClientMessage{ + Id: "2345", + Type: "command", + Command: &signaling.CommandProxyClientMessage{ + Type: "create-publisher", + PublisherId: publisherId, + Sid: "1234-abcd", + StreamType: signaling.StreamTypeVideo, + PublisherSettings: &signaling.NewPublisherSettings{ + Bitrate: 1234567, + MediaTypes: signaling.MediaTypeAudio | signaling.MediaTypeVideo, + }, + }, + })) + + var clientId string + if message, err := client1.RunUntilMessage(ctx); assert.NoError(err) { + assert.Equal("2345", message.Id) + if err := checkMessageType(message, "command"); assert.NoError(err) { + require.NotEmpty(message.Command.Id) + clientId = message.Command.Id + } + } + + client2 := NewProxyTestClient(ctx, t, server.URL) + defer client2.CloseWithBye() + + require.NoError(client2.SendHello(key)) + + hello2, err := client2.RunUntilHello(ctx) + if assert.NoError(err) { + assert.NotEmpty(hello2.Hello.SessionId, "%+v", hello2) + } + + _, err = client2.RunUntilLoad(ctx, 0) + assert.NoError(err) + + require.NoError(client2.WriteJSON(&signaling.ProxyClientMessage{ + Id: "3456", + Type: "command", + Command: &signaling.CommandProxyClientMessage{ + Type: "publish-remote", + StreamType: signaling.StreamTypeVideo, + ClientId: clientId, + Hostname: "remote-host", + Port: 10001, + RtcpPort: 10002, + }, + })) + + if message, err := client2.RunUntilMessage(ctx); assert.NoError(err) { + assert.Equal("3456", message.Id) + if err := checkMessageType(message, "command"); assert.NoError(err) { + require.NotEmpty(message.Command.Id) + } + } + + if assert.NotNil(mcu.publisher) { + assert.Equal(hello2.Hello.SessionId, mcu.publisher.remoteId) + if assert.NotNil(mcu.publisher.remoteData) { + assert.Equal("remote-host", mcu.publisher.remoteData.hostname) + assert.EqualValues(10001, mcu.publisher.remoteData.port) + assert.EqualValues(10002, mcu.publisher.remoteData.rtcpPort) + } + } + + // Closing the session will cause any active remote publishers stop be stopped. + client2.CloseWithBye() + + if assert.NotNil(mcu.publisher) { + assert.Empty(mcu.publisher.remoteId) + assert.Nil(mcu.publisher.remoteData) + } +} diff --git a/proxy/proxy_session.go b/proxy/proxy_session.go index ed9ac260..6225ae7e 100644 --- a/proxy/proxy_session.go +++ b/proxy/proxy_session.go @@ -23,6 +23,7 @@ package main import ( "context" + "fmt" "log" "sync" "sync/atomic" @@ -36,6 +37,12 @@ const ( sessionExpirationTime = time.Minute ) +type remotePublisherData struct { + hostname string + port int + rtcpPort int +} + type ProxySession struct { proxy *ProxyServer id string @@ -55,6 +62,9 @@ type ProxySession struct { subscribersLock sync.Mutex subscribers map[string]signaling.McuSubscriber subscriberIds map[signaling.McuSubscriber]string + + remotePublishersLock sync.Mutex + remotePublishers map[signaling.McuPublisher]map[string]*remotePublisherData } func NewProxySession(proxy *ProxyServer, sid uint64, id string) *ProxySession { @@ -121,6 +131,7 @@ func (s *ProxySession) Close() { s.closeFunc() s.clearPublishers() s.clearSubscribers() + s.clearRemotePublishers() s.proxy.DeleteSession(s.Sid()) } @@ -287,6 +298,8 @@ func (s *ProxySession) DeletePublisher(publisher signaling.McuPublisher) string delete(s.publishers, id) delete(s.publisherIds, publisher) + delete(s.remotePublishers, publisher) + go s.proxy.PublisherDeleted(publisher) return id } @@ -329,6 +342,20 @@ func (s *ProxySession) clearPublishers() { clear(s.publisherIds) } +func (s *ProxySession) clearRemotePublishers() { + s.remotePublishersLock.Lock() + defer s.remotePublishersLock.Unlock() + + go func(remotePublishers map[signaling.McuPublisher]map[string]*remotePublisherData) { + for publisher, entries := range remotePublishers { + for _, data := range entries { + publisher.UnpublishRemote(context.Background(), s.PublicId(), data.hostname, data.port, data.rtcpPort) + } + } + }(s.remotePublishers) + s.remotePublishers = nil +} + func (s *ProxySession) clearSubscribers() { s.publishersLock.Lock() defer s.publishersLock.Unlock() @@ -349,4 +376,58 @@ func (s *ProxySession) clearSubscribers() { func (s *ProxySession) NotifyDisconnected() { s.clearPublishers() s.clearSubscribers() + s.clearRemotePublishers() +} + +func (s *ProxySession) AddRemotePublisher(publisher signaling.McuPublisher, hostname string, port int, rtcpPort int) bool { + s.remotePublishersLock.Lock() + defer s.remotePublishersLock.Unlock() + + remote, found := s.remotePublishers[publisher] + if !found { + remote = make(map[string]*remotePublisherData) + if s.remotePublishers == nil { + s.remotePublishers = make(map[signaling.McuPublisher]map[string]*remotePublisherData) + } + s.remotePublishers[publisher] = remote + } + + key := fmt.Sprintf("%s:%d%d", hostname, port, rtcpPort) + if _, found := remote[key]; found { + return false + } + + data := &remotePublisherData{ + hostname: hostname, + port: port, + rtcpPort: rtcpPort, + } + remote[key] = data + return true +} + +func (s *ProxySession) RemoveRemotePublisher(publisher signaling.McuPublisher, hostname string, port int, rtcpPort int) { + s.remotePublishersLock.Lock() + defer s.remotePublishersLock.Unlock() + + remote, found := s.remotePublishers[publisher] + if !found { + return + } + + key := fmt.Sprintf("%s:%d%d", hostname, port, rtcpPort) + delete(remote, key) + if len(remote) == 0 { + delete(s.remotePublishers, publisher) + if len(s.remotePublishers) == 0 { + s.remotePublishers = nil + } + } +} + +func (s *ProxySession) OnPublisherDeleted(publisher signaling.McuPublisher) { + s.remotePublishersLock.Lock() + defer s.remotePublishersLock.Unlock() + + delete(s.remotePublishers, publisher) }