mirror of
https://github.com/strukturag/nextcloud-spreed-signaling
synced 2024-05-08 08:36:32 +02:00
Move remote classes to separate files and add event handlers.
This commit is contained in:
Joachim Bauch
parent
db4527544d
commit
ceb87bb728
120
mcu_janus.go
120
mcu_janus.go
|
|
@ -1033,69 +1033,6 @@ func (p *mcuJanusPublisher) PublishRemote(ctx context.Context, hostname string,
|
|||
return nil
|
||||
}
|
||||
|
||||
type mcuJanusRemotePublisher struct {
|
||||
mcuJanusClient
|
||||
|
||||
ref atomic.Int64
|
||||
publisher string
|
||||
port int
|
||||
rtcpPort int
|
||||
}
|
||||
|
||||
func (p *mcuJanusRemotePublisher) addRef() int64 {
|
||||
return p.ref.Add(1)
|
||||
}
|
||||
|
||||
func (p *mcuJanusRemotePublisher) release() bool {
|
||||
return p.ref.Add(-1) == 0
|
||||
}
|
||||
|
||||
func (p *mcuJanusRemotePublisher) Port() int {
|
||||
return p.port
|
||||
}
|
||||
|
||||
func (p *mcuJanusRemotePublisher) RtcpPort() int {
|
||||
return p.rtcpPort
|
||||
}
|
||||
|
||||
func (p *mcuJanusRemotePublisher) Close(ctx context.Context) {
|
||||
if !p.release() {
|
||||
return
|
||||
}
|
||||
|
||||
p.mu.Lock()
|
||||
if handle := p.handle; handle != nil {
|
||||
response, err := p.handle.Request(ctx, map[string]interface{}{
|
||||
"request": "remove_remote_publisher",
|
||||
"room": p.roomId,
|
||||
"id": streamTypeUserIds[p.streamType],
|
||||
})
|
||||
if err != nil {
|
||||
log.Printf("Error removing remote publisher %d in room %d: %s", p.id, p.roomId, err)
|
||||
} else {
|
||||
log.Printf("Removed remote publisher: %+v", response)
|
||||
}
|
||||
if p.roomId != 0 {
|
||||
destroy_msg := map[string]interface{}{
|
||||
"request": "destroy",
|
||||
"room": p.roomId,
|
||||
}
|
||||
if _, err := handle.Request(ctx, destroy_msg); err != nil {
|
||||
log.Printf("Error destroying room %d: %s", p.roomId, err)
|
||||
} else {
|
||||
log.Printf("Room %d destroyed", p.roomId)
|
||||
}
|
||||
p.mcu.mu.Lock()
|
||||
delete(p.mcu.remotePublishers, getStreamId(p.publisher, p.streamType))
|
||||
p.mcu.mu.Unlock()
|
||||
p.roomId = 0
|
||||
}
|
||||
}
|
||||
|
||||
p.closeClient(ctx)
|
||||
p.mu.Unlock()
|
||||
}
|
||||
|
||||
type mcuJanusSubscriber struct {
|
||||
mcuJanusClient
|
||||
|
||||
|
|
@ -1191,20 +1128,6 @@ func (m *mcuJanus) NewSubscriber(ctx context.Context, listener McuListener, publ
|
|||
return client, nil
|
||||
}
|
||||
|
||||
type mcuJanusRemoteSubscriber struct {
|
||||
mcuJanusSubscriber
|
||||
|
||||
remote atomic.Pointer[mcuJanusRemotePublisher]
|
||||
}
|
||||
|
||||
func (s *mcuJanusRemoteSubscriber) Close(ctx context.Context) {
|
||||
s.mcuJanusSubscriber.Close(ctx)
|
||||
|
||||
if remote := s.remote.Swap(nil); remote != nil {
|
||||
remote.Close(context.Background())
|
||||
}
|
||||
}
|
||||
|
||||
func (m *mcuJanus) getOrCreateRemotePublisher(ctx context.Context, controller RemotePublisherController, streamType StreamType, bitrate int) (*mcuJanusRemotePublisher, error) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
|
|
@ -1269,26 +1192,35 @@ func (m *mcuJanus) getOrCreateRemotePublisher(ctx context.Context, controller Re
|
|||
rtcp_port := getPluginIntValue(response.PluginData, pluginVideoRoom, "rtcp_port")
|
||||
|
||||
pub = &mcuJanusRemotePublisher{
|
||||
mcuJanusClient: mcuJanusClient{
|
||||
mcu: m,
|
||||
mcuJanusPublisher: mcuJanusPublisher{
|
||||
mcuJanusClient: mcuJanusClient{
|
||||
mcu: m,
|
||||
|
||||
id: id,
|
||||
session: response.Session,
|
||||
roomId: roomId,
|
||||
sid: strconv.FormatUint(handle.Id, 10),
|
||||
streamType: streamType,
|
||||
maxBitrate: bitrate,
|
||||
id: id,
|
||||
session: response.Session,
|
||||
roomId: roomId,
|
||||
sid: strconv.FormatUint(handle.Id, 10),
|
||||
streamType: streamType,
|
||||
maxBitrate: bitrate,
|
||||
|
||||
handle: handle,
|
||||
handleId: handle.Id,
|
||||
closeChan: make(chan struct{}, 1),
|
||||
deferred: make(chan func(), 64),
|
||||
handle: handle,
|
||||
handleId: handle.Id,
|
||||
closeChan: make(chan struct{}, 1),
|
||||
deferred: make(chan func(), 64),
|
||||
},
|
||||
|
||||
id: controller.PublisherId(),
|
||||
},
|
||||
|
||||
publisher: controller.PublisherId(),
|
||||
port: int(port),
|
||||
rtcpPort: int(rtcp_port),
|
||||
port: int(port),
|
||||
rtcpPort: int(rtcp_port),
|
||||
}
|
||||
pub.mcuJanusClient.handleEvent = pub.handleEvent
|
||||
pub.mcuJanusClient.handleHangup = pub.handleHangup
|
||||
pub.mcuJanusClient.handleDetached = pub.handleDetached
|
||||
pub.mcuJanusClient.handleConnected = pub.handleConnected
|
||||
pub.mcuJanusClient.handleSlowLink = pub.handleSlowLink
|
||||
pub.mcuJanusClient.handleMedia = pub.handleMedia
|
||||
|
||||
if err := controller.StartPublishing(ctx, pub); err != nil {
|
||||
go pub.Close(context.Background())
|
||||
|
|
@ -1334,7 +1266,7 @@ func (m *mcuJanus) NewRemoteSubscriber(ctx context.Context, listener McuListener
|
|||
return nil, err
|
||||
}
|
||||
|
||||
log.Printf("Attached subscriber to room %d of publisher %s in plugin %s in session %d as %d", pub.roomId, pub.publisher, pluginVideoRoom, session.Id, handle.Id)
|
||||
log.Printf("Attached subscriber to room %d of publisher %s in plugin %s in session %d as %d", pub.roomId, pub.id, pluginVideoRoom, session.Id, handle.Id)
|
||||
|
||||
client := &mcuJanusRemoteSubscriber{
|
||||
mcuJanusSubscriber: mcuJanusSubscriber{
|
||||
|
|
@ -1353,7 +1285,7 @@ func (m *mcuJanus) NewRemoteSubscriber(ctx context.Context, listener McuListener
|
|||
closeChan: make(chan struct{}, 1),
|
||||
deferred: make(chan func(), 64),
|
||||
},
|
||||
publisher: pub.publisher,
|
||||
publisher: pub.id,
|
||||
},
|
||||
}
|
||||
client.remote.Store(pub)
|
||||
|
|
|
|||
150
mcu_janus_remote_publisher.go
Normal file
150
mcu_janus_remote_publisher.go
Normal file
|
|
@ -0,0 +1,150 @@
|
|||
/**
|
||||
* Standalone signaling server for the Nextcloud Spreed app.
|
||||
* Copyright (C) 2024 struktur AG
|
||||
*
|
||||
* @author Joachim Bauch <bauch@struktur.de>
|
||||
*
|
||||
* @license GNU AGPL version 3 or any later version
|
||||
*
|
||||
* This program is free software: you can redistribute it and/or modify
|
||||
* it under the terms of the GNU Affero General Public License as published by
|
||||
* the Free Software Foundation, either version 3 of the License, or
|
||||
* (at your option) any later version.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU Affero General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU Affero General Public License
|
||||
* along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
*/
|
||||
package signaling
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log"
|
||||
"sync/atomic"
|
||||
|
||||
"github.com/notedit/janus-go"
|
||||
)
|
||||
|
||||
type mcuJanusRemotePublisher struct {
|
||||
mcuJanusPublisher
|
||||
|
||||
ref atomic.Int64
|
||||
|
||||
port int
|
||||
rtcpPort int
|
||||
}
|
||||
|
||||
func (p *mcuJanusRemotePublisher) addRef() int64 {
|
||||
return p.ref.Add(1)
|
||||
}
|
||||
|
||||
func (p *mcuJanusRemotePublisher) release() bool {
|
||||
return p.ref.Add(-1) == 0
|
||||
}
|
||||
|
||||
func (p *mcuJanusRemotePublisher) Port() int {
|
||||
return p.port
|
||||
}
|
||||
|
||||
func (p *mcuJanusRemotePublisher) RtcpPort() int {
|
||||
return p.rtcpPort
|
||||
}
|
||||
|
||||
func (p *mcuJanusRemotePublisher) handleEvent(event *janus.EventMsg) {
|
||||
if videoroom := getPluginStringValue(event.Plugindata, pluginVideoRoom, "videoroom"); videoroom != "" {
|
||||
ctx := context.TODO()
|
||||
switch videoroom {
|
||||
case "destroyed":
|
||||
log.Printf("Remote publisher %d: associated room has been destroyed, closing", p.handleId)
|
||||
go p.Close(ctx)
|
||||
case "slow_link":
|
||||
// Ignore, processed through "handleSlowLink" in the general events.
|
||||
default:
|
||||
log.Printf("Unsupported videoroom remote publisher event in %d: %+v", p.handleId, event)
|
||||
}
|
||||
} else {
|
||||
log.Printf("Unsupported remote publisher event in %d: %+v", p.handleId, event)
|
||||
}
|
||||
}
|
||||
|
||||
func (p *mcuJanusRemotePublisher) handleHangup(event *janus.HangupMsg) {
|
||||
log.Printf("Remote publisher %d received hangup (%s), closing", p.handleId, event.Reason)
|
||||
go p.Close(context.Background())
|
||||
}
|
||||
|
||||
func (p *mcuJanusRemotePublisher) handleDetached(event *janus.DetachedMsg) {
|
||||
log.Printf("Remote publisher %d received detached, closing", p.handleId)
|
||||
go p.Close(context.Background())
|
||||
}
|
||||
|
||||
func (p *mcuJanusRemotePublisher) handleConnected(event *janus.WebRTCUpMsg) {
|
||||
log.Printf("Remote publisher %d received connected", p.handleId)
|
||||
p.mcu.publisherConnected.Notify(getStreamId(p.id, p.streamType))
|
||||
}
|
||||
|
||||
func (p *mcuJanusRemotePublisher) handleSlowLink(event *janus.SlowLinkMsg) {
|
||||
if event.Uplink {
|
||||
log.Printf("Remote publisher %s (%d) is reporting %d lost packets on the uplink (Janus -> client)", p.listener.PublicId(), p.handleId, event.Lost)
|
||||
} else {
|
||||
log.Printf("Remote publisher %s (%d) is reporting %d lost packets on the downlink (client -> Janus)", p.listener.PublicId(), p.handleId, event.Lost)
|
||||
}
|
||||
}
|
||||
|
||||
func (p *mcuJanusRemotePublisher) NotifyReconnected() {
|
||||
ctx := context.TODO()
|
||||
handle, session, roomId, _, err := p.mcu.getOrCreatePublisherHandle(ctx, p.id, p.streamType, p.bitrate)
|
||||
if err != nil {
|
||||
log.Printf("Could not reconnect remote publisher %s: %s", p.id, err)
|
||||
// TODO(jojo): Retry
|
||||
return
|
||||
}
|
||||
|
||||
p.handle = handle
|
||||
p.handleId = handle.Id
|
||||
p.session = session
|
||||
p.roomId = roomId
|
||||
|
||||
log.Printf("Remote publisher %s reconnected on handle %d", p.id, p.handleId)
|
||||
}
|
||||
|
||||
func (p *mcuJanusRemotePublisher) Close(ctx context.Context) {
|
||||
if !p.release() {
|
||||
return
|
||||
}
|
||||
|
||||
p.mu.Lock()
|
||||
if handle := p.handle; handle != nil {
|
||||
response, err := p.handle.Request(ctx, map[string]interface{}{
|
||||
"request": "remove_remote_publisher",
|
||||
"room": p.roomId,
|
||||
"id": streamTypeUserIds[p.streamType],
|
||||
})
|
||||
if err != nil {
|
||||
log.Printf("Error removing remote publisher %s in room %d: %s", p.id, p.roomId, err)
|
||||
} else {
|
||||
log.Printf("Removed remote publisher: %+v", response)
|
||||
}
|
||||
if p.roomId != 0 {
|
||||
destroy_msg := map[string]interface{}{
|
||||
"request": "destroy",
|
||||
"room": p.roomId,
|
||||
}
|
||||
if _, err := handle.Request(ctx, destroy_msg); err != nil {
|
||||
log.Printf("Error destroying room %d: %s", p.roomId, err)
|
||||
} else {
|
||||
log.Printf("Room %d destroyed", p.roomId)
|
||||
}
|
||||
p.mcu.mu.Lock()
|
||||
delete(p.mcu.remotePublishers, getStreamId(p.id, p.streamType))
|
||||
p.mcu.mu.Unlock()
|
||||
p.roomId = 0
|
||||
}
|
||||
}
|
||||
|
||||
p.closeClient(ctx)
|
||||
p.mu.Unlock()
|
||||
}
|
||||
115
mcu_janus_remote_subscriber.go
Normal file
115
mcu_janus_remote_subscriber.go
Normal file
|
|
@ -0,0 +1,115 @@
|
|||
/**
|
||||
* Standalone signaling server for the Nextcloud Spreed app.
|
||||
* Copyright (C) 2024 struktur AG
|
||||
*
|
||||
* @author Joachim Bauch <bauch@struktur.de>
|
||||
*
|
||||
* @license GNU AGPL version 3 or any later version
|
||||
*
|
||||
* This program is free software: you can redistribute it and/or modify
|
||||
* it under the terms of the GNU Affero General Public License as published by
|
||||
* the Free Software Foundation, either version 3 of the License, or
|
||||
* (at your option) any later version.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU Affero General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU Affero General Public License
|
||||
* along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
*/
|
||||
package signaling
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log"
|
||||
"strconv"
|
||||
"sync/atomic"
|
||||
|
||||
"github.com/notedit/janus-go"
|
||||
)
|
||||
|
||||
type mcuJanusRemoteSubscriber struct {
|
||||
mcuJanusSubscriber
|
||||
|
||||
remote atomic.Pointer[mcuJanusRemotePublisher]
|
||||
}
|
||||
|
||||
func (p *mcuJanusRemoteSubscriber) handleEvent(event *janus.EventMsg) {
|
||||
if videoroom := getPluginStringValue(event.Plugindata, pluginVideoRoom, "videoroom"); videoroom != "" {
|
||||
ctx := context.TODO()
|
||||
switch videoroom {
|
||||
case "destroyed":
|
||||
log.Printf("Remote subscriber %d: associated room has been destroyed, closing", p.handleId)
|
||||
go p.Close(ctx)
|
||||
case "event":
|
||||
// Handle renegotiations, but ignore other events like selected
|
||||
// substream / temporal layer.
|
||||
if getPluginStringValue(event.Plugindata, pluginVideoRoom, "configured") == "ok" &&
|
||||
event.Jsep != nil && event.Jsep["type"] == "offer" && event.Jsep["sdp"] != nil {
|
||||
p.listener.OnUpdateOffer(p, event.Jsep)
|
||||
}
|
||||
case "slow_link":
|
||||
// Ignore, processed through "handleSlowLink" in the general events.
|
||||
default:
|
||||
log.Printf("Unsupported videoroom event %s for remote subscriber %d: %+v", videoroom, p.handleId, event)
|
||||
}
|
||||
} else {
|
||||
log.Printf("Unsupported event for remote subscriber %d: %+v", p.handleId, event)
|
||||
}
|
||||
}
|
||||
|
||||
func (p *mcuJanusRemoteSubscriber) handleHangup(event *janus.HangupMsg) {
|
||||
log.Printf("Remote subscriber %d received hangup (%s), closing", p.handleId, event.Reason)
|
||||
go p.Close(context.Background())
|
||||
}
|
||||
|
||||
func (p *mcuJanusRemoteSubscriber) handleDetached(event *janus.DetachedMsg) {
|
||||
log.Printf("Remote subscriber %d received detached, closing", p.handleId)
|
||||
go p.Close(context.Background())
|
||||
}
|
||||
|
||||
func (p *mcuJanusRemoteSubscriber) handleConnected(event *janus.WebRTCUpMsg) {
|
||||
log.Printf("Remote subscriber %d received connected", p.handleId)
|
||||
p.mcu.SubscriberConnected(p.Id(), p.publisher, p.streamType)
|
||||
}
|
||||
|
||||
func (p *mcuJanusRemoteSubscriber) handleSlowLink(event *janus.SlowLinkMsg) {
|
||||
if event.Uplink {
|
||||
log.Printf("Remote subscriber %s (%d) is reporting %d lost packets on the uplink (Janus -> client)", p.listener.PublicId(), p.handleId, event.Lost)
|
||||
} else {
|
||||
log.Printf("Remote subscriber %s (%d) is reporting %d lost packets on the downlink (client -> Janus)", p.listener.PublicId(), p.handleId, event.Lost)
|
||||
}
|
||||
}
|
||||
|
||||
func (p *mcuJanusRemoteSubscriber) handleMedia(event *janus.MediaMsg) {
|
||||
// Only triggered for publishers
|
||||
}
|
||||
|
||||
func (p *mcuJanusRemoteSubscriber) NotifyReconnected() {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), p.mcu.mcuTimeout)
|
||||
defer cancel()
|
||||
handle, pub, err := p.mcu.getOrCreateSubscriberHandle(ctx, p.publisher, p.streamType)
|
||||
if err != nil {
|
||||
// TODO(jojo): Retry?
|
||||
log.Printf("Could not reconnect remote subscriber for publisher %s: %s", p.publisher, err)
|
||||
p.Close(context.Background())
|
||||
return
|
||||
}
|
||||
|
||||
p.handle = handle
|
||||
p.handleId = handle.Id
|
||||
p.roomId = pub.roomId
|
||||
p.sid = strconv.FormatUint(handle.Id, 10)
|
||||
p.listener.SubscriberSidUpdated(p)
|
||||
log.Printf("Subscriber %d for publisher %s reconnected on handle %d", p.id, p.publisher, p.handleId)
|
||||
}
|
||||
|
||||
func (p *mcuJanusRemoteSubscriber) Close(ctx context.Context) {
|
||||
p.mcuJanusSubscriber.Close(ctx)
|
||||
|
||||
if remote := p.remote.Swap(nil); remote != nil {
|
||||
remote.Close(context.Background())
|
||||
}
|
||||
}
|
||||
Loading…
Reference in a new issue