nextcloud-spreed-signaling/room_ping.go

229 lines
5.9 KiB
Go

/**
* Standalone signaling server for the Nextcloud Spreed app.
* Copyright (C) 2022 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"
"net/url"
"sync"
"time"
)
const (
ConfigGroupSignaling = "signaling"
ConfigKeySessionPingLimit = "session-ping-limit"
)
type pingEntries struct {
url *url.URL
entries map[*Room][]BackendPingEntry
}
func newPingEntries(url *url.URL, room *Room, entries []BackendPingEntry) *pingEntries {
return &pingEntries{
url: url,
entries: map[*Room][]BackendPingEntry{
room: entries,
},
}
}
func (e *pingEntries) Add(room *Room, entries []BackendPingEntry) {
if existing, found := e.entries[room]; found {
e.entries[room] = append(existing, entries...)
} else {
e.entries[room] = entries
}
}
func (e *pingEntries) RemoveRoom(room *Room) {
delete(e.entries, room)
}
// RoomPing sends ping requests for active sessions in rooms. It evaluates the
// capabilities of the Nextcloud server to determine if sessions from different
// rooms can be grouped together.
//
// For that, all ping requests across rooms of enabled instances are combined
// and sent out batched every "updateActiveSessionsInterval" seconds.
type RoomPing struct {
mu sync.Mutex
closeChan chan bool
backend *BackendClient
capabilities *Capabilities
entries map[string]*pingEntries
}
func NewRoomPing(backend *BackendClient, capabilities *Capabilities) (*RoomPing, error) {
result := &RoomPing{
closeChan: make(chan bool, 1),
backend: backend,
capabilities: capabilities,
}
return result, nil
}
func (p *RoomPing) Start() {
go p.run()
}
func (p *RoomPing) Stop() {
select {
case p.closeChan <- true:
default:
}
}
func (p *RoomPing) run() {
ticker := time.NewTicker(updateActiveSessionsInterval)
loop:
for {
select {
case <-p.closeChan:
break loop
case <-ticker.C:
p.publishActiveSessions()
}
}
}
func (p *RoomPing) getAndClearEntries() map[string]*pingEntries {
p.mu.Lock()
defer p.mu.Unlock()
entries := p.entries
p.entries = nil
return entries
}
func (p *RoomPing) publishEntries(entries *pingEntries, timeout time.Duration) {
ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
limit, found := p.capabilities.GetIntegerConfig(ctx, entries.url, ConfigGroupSignaling, ConfigKeySessionPingLimit)
if !found || limit <= 0 {
// Limit disabled while waiting for the next iteration, fallback to sending
// one request per room.
for room, e := range entries.entries {
ctx2, cancel2 := context.WithTimeout(context.Background(), timeout)
defer cancel2()
if err := p.sendPingsDirect(ctx2, room, entries.url, e); err != nil {
log.Printf("Error pinging room %s for active entries %+v: %s", room.Id(), e, err)
}
}
return
}
var allEntries []BackendPingEntry
for _, e := range entries.entries {
allEntries = append(allEntries, e...)
}
p.sendPingsCombined(entries.url, allEntries, limit, timeout)
}
func (p *RoomPing) publishActiveSessions() {
var timeout time.Duration
if p.backend.hub != nil {
timeout = p.backend.hub.backendTimeout
} else {
// Running from tests.
timeout = time.Second * time.Duration(defaultBackendTimeoutSeconds)
}
entries := p.getAndClearEntries()
var wg sync.WaitGroup
wg.Add(len(entries))
for _, e := range entries {
go func(e *pingEntries) {
defer wg.Done()
p.publishEntries(e, timeout)
}(e)
}
wg.Wait()
}
func (p *RoomPing) sendPingsDirect(ctx context.Context, room *Room, url *url.URL, entries []BackendPingEntry) error {
request := NewBackendClientPingRequest(room.Id(), entries)
var response BackendClientResponse
return p.backend.PerformJSONRequest(ctx, url, request, &response)
}
func (p *RoomPing) sendPingsCombined(url *url.URL, entries []BackendPingEntry, limit int, timeout time.Duration) {
total := len(entries)
for idx := 0; idx < total; idx += limit {
end := idx + limit
if end > total {
end = total
}
tosend := entries[idx:end]
ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
request := NewBackendClientPingRequest("", tosend)
var response BackendClientResponse
if err := p.backend.PerformJSONRequest(ctx, url, request, &response); err != nil {
log.Printf("Error sending combined ping session entries %+v to %s: %s", tosend, url, err)
}
}
}
func (p *RoomPing) SendPings(ctx context.Context, room *Room, url *url.URL, entries []BackendPingEntry) error {
limit, found := p.capabilities.GetIntegerConfig(ctx, url, ConfigGroupSignaling, ConfigKeySessionPingLimit)
if !found || limit <= 0 {
// Old-style Nextcloud or session limit not configured. Perform one request
// per room. Don't queue to avoid sending all ping requests to old-style
// instances at the same time but distribute across the interval.
return p.sendPingsDirect(ctx, room, url, entries)
}
key := url.String()
p.mu.Lock()
defer p.mu.Unlock()
if existing, found := p.entries[key]; found {
existing.Add(room, entries)
return nil
}
if p.entries == nil {
p.entries = make(map[string]*pingEntries)
}
p.entries[key] = newPingEntries(url, room, entries)
return nil
}
func (p *RoomPing) DeleteRoom(room *Room) {
p.mu.Lock()
defer p.mu.Unlock()
for _, entries := range p.entries {
entries.RemoveRoom(room)
}
}