archive/strukturag_nextcloud-spreed-signaling
0
0
Fork 0
mirror of https://github.com/strukturag/nextcloud-spreed-signaling.git synced 2025-03-30 09:23:35 +00:00
Code Issues Projects Releases Wiki Activity

Notify remote to stop publishing when last local subscriber is closed.

Browse source
This commit is contained in:
Joachim Bauch 2024-11-07 14:33:02 +01:00
parent b7c40d1cc0
commit c12addb2a8
No known key found for this signature in database
GPG key ID: 77C1D22D53E15F02
9 changed files with 175 additions and 10 deletions

3
mcu_common.go
View file

@ -166,6 +166,7 @@ type RemotePublisherController interface {
PublisherId() string
StartPublishing(ctx context.Context, publisher McuRemotePublisherProperties) error
StopPublishing(ctx context.Context, publisher McuRemotePublisherProperties) error
GetStreams(ctx context.Context) ([]PublisherStream, error)
}
@ -214,7 +215,7 @@ type McuPublisher interface {
GetStreams(ctx context.Context) ([]PublisherStream, error)
PublishRemote(ctx context.Context, remoteId string, hostname string, port int, rtcpPort int) error
UnpublishRemote(ctx context.Context, remoteId string) error
UnpublishRemote(ctx context.Context, remoteId string, hostname string, port int, rtcpPort int) error
}
type McuSubscriber interface {

2
mcu_janus.go
View file

@ -785,6 +785,8 @@ func (m *mcuJanus) getOrCreateRemotePublisher(ctx context.Context, controller Re
settings: settings,
},
controller: controller,
port: int(port),
rtcpPort: int(rtcp_port),
}

10
mcu_janus_publisher.go
View file

@ -380,8 +380,8 @@ func (p *mcuJanusPublisher) GetStreams(ctx context.Context) ([]PublisherStream,
return streams, nil
}
func getPublisherRemoteId(id string, remoteId string) string {
return fmt.Sprintf("%s@%s", id, remoteId)
func getPublisherRemoteId(id string, remoteId string, hostname string, port int, rtcpPort int) string {
return fmt.Sprintf("%s-%s@%s:%d:%d", id, remoteId, hostname, port, rtcpPort)
}
func (p *mcuJanusPublisher) PublishRemote(ctx context.Context, remoteId string, hostname string, port int, rtcpPort int) error {
@ -389,7 +389,7 @@ func (p *mcuJanusPublisher) PublishRemote(ctx context.Context, remoteId string,
"request": "publish_remotely",
"room": p.roomId,
"publisher_id": streamTypeUserIds[p.streamType],
"remote_id": getPublisherRemoteId(p.id, remoteId),
"remote_id": getPublisherRemoteId(p.id, remoteId, hostname, port, rtcpPort),
"host": hostname,
"port": port,
"rtcp_port": rtcpPort,
@ -421,12 +421,12 @@ func (p *mcuJanusPublisher) PublishRemote(ctx context.Context, remoteId string,
return nil
}
func (p *mcuJanusPublisher) UnpublishRemote(ctx context.Context, remoteId string) error {
func (p *mcuJanusPublisher) UnpublishRemote(ctx context.Context, remoteId string, hostname string, port int, rtcpPort int) error {
msg := map[string]interface{}{
"request": "unpublish_remotely",
"room": p.roomId,
"publisher_id": streamTypeUserIds[p.streamType],
"remote_id": getPublisherRemoteId(p.id, remoteId),
"remote_id": getPublisherRemoteId(p.id, remoteId, hostname, port, rtcpPort),
}
response, err := p.handle.Request(ctx, msg)
if err != nil {

6
mcu_janus_remote_publisher.go
View file

@ -34,6 +34,8 @@ type mcuJanusRemotePublisher struct {
ref atomic.Int64
controller RemotePublisherController
port int
rtcpPort int
}
@ -116,6 +118,10 @@ func (p *mcuJanusRemotePublisher) Close(ctx context.Context) {
return
}
if err := p.controller.StopPublishing(ctx, p); err != nil {
log.Printf("Error stopping remote publisher %s in room %d: %s", p.id, p.roomId, err)
}
p.mu.Lock()
if handle := p.handle; handle != nil {
response, err := p.handle.Request(ctx, map[string]interface{}{

2
mcu_proxy.go
View file

@ -227,7 +227,7 @@ func (p *mcuProxyPublisher) PublishRemote(ctx context.Context, remoteId string,
return errors.New("remote publishing not supported for proxy publishers")
}
func (p *mcuProxyPublisher) UnpublishRemote(ctx context.Context, remoteId string) error {
func (p *mcuProxyPublisher) UnpublishRemote(ctx context.Context, remoteId string, hostname string, port int, rtcpPort int) error {
return errors.New("remote publishing not supported for proxy publishers")
}

104
mcu_proxy_test.go
View file

@ -1502,6 +1502,110 @@ func Test_ProxyRemotePublisher(t *testing.T) {
defer sub.Close(context.Background())
}
func Test_ProxyMultipleRemotePublisher(t *testing.T) {
CatchLogForTest(t)
t.Parallel()
etcd := NewEtcdForTest(t)
grpcServer1, addr1 := NewGrpcServerForTest(t)
grpcServer2, addr2 := NewGrpcServerForTest(t)
grpcServer3, addr3 := NewGrpcServerForTest(t)
hub1 := &mockGrpcServerHub{}
hub2 := &mockGrpcServerHub{}
hub3 := &mockGrpcServerHub{}
grpcServer1.hub = hub1
grpcServer2.hub = hub2
grpcServer3.hub = hub3
SetEtcdValue(etcd, "/grpctargets/one", []byte("{\"address\":\""+addr1+"\"}"))
SetEtcdValue(etcd, "/grpctargets/two", []byte("{\"address\":\""+addr2+"\"}"))
SetEtcdValue(etcd, "/grpctargets/three", []byte("{\"address\":\""+addr3+"\"}"))
server1 := NewProxyServerForTest(t, "DE")
server2 := NewProxyServerForTest(t, "US")
server3 := NewProxyServerForTest(t, "US")
mcu1 := newMcuProxyForTestWithOptions(t, proxyTestOptions{
etcd: etcd,
servers: []*TestProxyServerHandler{
server1,
server2,
server3,
},
})
mcu2 := newMcuProxyForTestWithOptions(t, proxyTestOptions{
etcd: etcd,
servers: []*TestProxyServerHandler{
server1,
server2,
server3,
},
})
mcu3 := newMcuProxyForTestWithOptions(t, proxyTestOptions{
etcd: etcd,
servers: []*TestProxyServerHandler{
server1,
server2,
server3,
},
})
ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
defer cancel()
pubId := "the-publisher"
pubSid := "1234567890"
pubListener := &MockMcuListener{
publicId: pubId + "-public",
}
pubInitiator := &MockMcuInitiator{
country: "DE",
}
session1 := &ClientSession{
publicId: pubId,
publishers: make(map[StreamType]McuPublisher),
}
hub1.addSession(session1)
defer hub1.removeSession(session1)
pub, err := mcu1.NewPublisher(ctx, pubListener, pubId, pubSid, StreamTypeVideo, NewPublisherSettings{
MediaTypes: MediaTypeVideo | MediaTypeAudio,
}, pubInitiator)
require.NoError(t, err)
defer pub.Close(context.Background())
session1.mu.Lock()
session1.publishers[StreamTypeVideo] = pub
session1.publisherWaiters.Wakeup()
session1.mu.Unlock()
sub1Listener := &MockMcuListener{
publicId: "subscriber-public-1",
}
sub1Initiator := &MockMcuInitiator{
country: "US",
}
sub1, err := mcu2.NewSubscriber(ctx, sub1Listener, pubId, StreamTypeVideo, sub1Initiator)
require.NoError(t, err)
defer sub1.Close(context.Background())
sub2Listener := &MockMcuListener{
publicId: "subscriber-public-2",
}
sub2Initiator := &MockMcuInitiator{
country: "US",
}
sub2, err := mcu3.NewSubscriber(ctx, sub2Listener, pubId, StreamTypeVideo, sub2Initiator)
require.NoError(t, err)
defer sub2.Close(context.Background())
}
func Test_ProxyRemotePublisherWait(t *testing.T) {
CatchLogForTest(t)
t.Parallel()

2
mcu_test.go
View file

@ -229,7 +229,7 @@ func (p *TestMCUPublisher) PublishRemote(ctx context.Context, remoteId string, h
return errors.New("remote publishing not supported")
}
func (p *TestMCUPublisher) UnpublishRemote(ctx context.Context, remoteId string) error {
func (p *TestMCUPublisher) UnpublishRemote(ctx context.Context, remoteId string, hostname string, port int, rtcpPort int) error {
return errors.New("remote publishing not supported")
}

54
proxy/proxy_server.go
View file

@ -856,6 +856,28 @@ func (p *proxyRemotePublisher) StartPublishing(ctx context.Context, publisher si
return nil
}
func (p *proxyRemotePublisher) StopPublishing(ctx context.Context, publisher signaling.McuRemotePublisherProperties) error {
conn, err := p.proxy.getRemoteConnection(p.remoteUrl)
if err != nil {
return err
}
if _, err := conn.RequestMessage(ctx, &signaling.ProxyClientMessage{
Type: "command",
Command: &signaling.CommandProxyClientMessage{
Type: "unpublish-remote",
ClientId: p.publisherId,
Hostname: p.proxy.remoteHostname,
Port: publisher.Port(),
RtcpPort: publisher.RtcpPort(),
},
}); err != nil {
return err
}
return nil
}
func (p *proxyRemotePublisher) GetStreams(ctx context.Context) ([]signaling.PublisherStream, error) {
conn, err := p.proxy.getRemoteConnection(p.remoteUrl)
if err != nil {
@ -1125,7 +1147,7 @@ func (s *ProxyServer) processCommand(ctx context.Context, client *ProxyClient, s
ctx2, cancel = context.WithTimeout(ctx, s.mcuTimeout)
defer cancel()
if err := publisher.UnpublishRemote(ctx2, session.PublicId()); err != nil {
if err := publisher.UnpublishRemote(ctx2, session.PublicId(), cmd.Hostname, cmd.Port, cmd.RtcpPort); err != nil {
log.Printf("Error unpublishing old %s %s to remote %s (port=%d, rtcpPort=%d): %s", publisher.StreamType(), cmd.ClientId, cmd.Hostname, cmd.Port, cmd.RtcpPort, err)
session.sendMessage(message.NewWrappedErrorServerMessage(err))
return
@ -1141,6 +1163,36 @@ func (s *ProxyServer) processCommand(ctx context.Context, client *ProxyClient, s
}
}
response := &signaling.ProxyServerMessage{
Id: message.Id,
Type: "command",
Command: &signaling.CommandProxyServerMessage{
Id: cmd.ClientId,
},
}
session.sendMessage(response)
case "unpublish-remote":
client := s.GetClient(cmd.ClientId)
if client == nil {
session.sendMessage(message.NewErrorServerMessage(UnknownClient))
return
}
publisher, ok := client.(signaling.McuPublisher)
if !ok {
session.sendMessage(message.NewErrorServerMessage(UnknownClient))
return
}
ctx2, cancel := context.WithTimeout(ctx, s.mcuTimeout)
defer cancel()
if err := publisher.UnpublishRemote(ctx2, session.PublicId(), cmd.Hostname, cmd.Port, cmd.RtcpPort); err != nil {
log.Printf("Error unpublishing %s %s from remote %s: %s", publisher.StreamType(), cmd.ClientId, cmd.Hostname, err)
session.sendMessage(message.NewWrappedErrorServerMessage(err))
return
}
response := &signaling.ProxyServerMessage{
Id: message.Id,
Type: "command",

2
proxy/proxy_server_test.go
View file

@ -431,7 +431,7 @@ func (p *TestMCUPublisher) PublishRemote(ctx context.Context, remoteId string, h
return errors.New("not implemented")
}
func (p *TestMCUPublisher) UnpublishRemote(ctx context.Context, remoteId string) error {
func (p *TestMCUPublisher) UnpublishRemote(ctx context.Context, remoteId string, hostname string, port int, rtcpPort int) error {
return errors.New("not implemented")
}