deblan/nextcloud-spreed-signaling
deblan
/
nextcloud-spreed-signaling
mirror of https://github.com/strukturag/nextcloud-spreed-signaling
1
0
Fork 0
Code Issues Releases Wiki Activity

Send more responses through the session to handle disconnected / resuming clients.

This commit is contained in:
Joachim Bauch 2021-05-31 16:40:21 +02:00
parent 15d5c93b0b
commit 21f3f506d1
No known key found for this signature in database
GPG Key ID: 77C1D22D53E15F02
4 changed files with 44 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
clientsession.go
View File

@ -559,6 +559,14 @@ func (s *ClientSession) sendMessageUnlocked(message *ServerMessage) bool {
return true return true
} }
func (s *ClientSession) SendError(e *Error) bool {
message := &ServerMessage{
Type: "error",
Error: e,
}
return s.SendMessage(message)
}
func (s *ClientSession) SendMessage(message *ServerMessage) bool { func (s *ClientSession) SendMessage(message *ServerMessage) bool {
s.mu.Lock() s.mu.Lock()
defer s.mu.Unlock() defer s.mu.Unlock()
@ -822,7 +830,7 @@ func (s *ClientSession) NotifySessionResumed(client *Client) {
if len(s.pendingClientMessages) == 0 { if len(s.pendingClientMessages) == 0 {
s.mu.Unlock() s.mu.Unlock()
if room := s.GetRoom(); room != nil { if room := s.GetRoom(); room != nil {
room.NotifySessionResumed(client) room.NotifySessionResumed(s)
} }
return return
} }
@ -841,7 +849,7 @@ func (s *ClientSession) NotifySessionResumed(client *Client) {
if !hasPendingParticipantsUpdate { if !hasPendingParticipantsUpdate {
// Only need to send initial participants list update if none was part of the pending messages. // Only need to send initial participants list update if none was part of the pending messages.
if room := s.GetRoom(); room != nil { if room := s.GetRoom(); room != nil {
room.NotifySessionResumed(client) room.NotifySessionResumed(s)
} }
} }
} }

44
hub.go
View File

@ -768,20 +768,22 @@ func (h *Hub) processMessage(client *Client, data []byte) {
if err := message.UnmarshalJSON(data); err != nil { if err := message.UnmarshalJSON(data); err != nil {
if session := client.GetSession(); session != nil { if session := client.GetSession(); session != nil {
log.Printf("Error decoding message from client %s: %v", session.PublicId(), err) log.Printf("Error decoding message from client %s: %v", session.PublicId(), err)
session.SendError(InvalidFormat)
} else { } else {
log.Printf("Error decoding message from %s: %v", client.RemoteAddr(), err) log.Printf("Error decoding message from %s: %v", client.RemoteAddr(), err)
client.SendError(InvalidFormat)
} }
client.SendError(InvalidFormat)
return return
} }
if err := message.CheckValid(); err != nil { if err := message.CheckValid(); err != nil {
if session := client.GetSession(); session != nil { if session := client.GetSession(); session != nil {
log.Printf("Invalid message %+v from client %s: %v", message, session.PublicId(), err) log.Printf("Invalid message %+v from client %s: %v", message, session.PublicId(), err)
session.SendMessage(message.NewErrorServerMessage(InvalidFormat))
} else { } else {
log.Printf("Invalid message %+v from %s: %v", message, client.RemoteAddr(), err) log.Printf("Invalid message %+v from %s: %v", message, client.RemoteAddr(), err)
client.SendMessage(message.NewErrorServerMessage(InvalidFormat))
} }
client.SendMessage(message.NewErrorServerMessage(InvalidFormat))
return return
} }
@ -1201,7 +1203,7 @@ func (h *Hub) processMessageMsg(client *Client, message *ClientMessage) {
case "requestoffer": case "requestoffer":
// Process asynchronously to avoid blocking regular // Process asynchronously to avoid blocking regular
// message processing for this client. // message processing for this client.
go h.processMcuMessage(client, client, session, message, msg, &data) go h.processMcuMessage(session, session, message, msg, &data)
return return
case "offer": case "offer":
fallthrough fallthrough
@ -1210,7 +1212,7 @@ func (h *Hub) processMessageMsg(client *Client, message *ClientMessage) {
case "endOfCandidates": case "endOfCandidates":
fallthrough fallthrough
case "candidate": case "candidate":
h.processMcuMessage(client, client, session, message, msg, &data) h.processMcuMessage(session, session, message, msg, &data)
return return
} }
} }
@ -1311,7 +1313,7 @@ func (h *Hub) processMessageMsg(client *Client, message *ClientMessage) {
if clientData != nil && clientData.Type == "sendoffer" { if clientData != nil && clientData.Type == "sendoffer" {
if !isAllowedToSend(session, clientData) { if !isAllowedToSend(session, clientData) {
log.Printf("Session %s is not allowed to send offer for %s, ignoring", session.PublicId(), clientData.RoomType) log.Printf("Session %s is not allowed to send offer for %s, ignoring", session.PublicId(), clientData.RoomType)
sendNotAllowed(client, message) sendNotAllowed(session, message)
return return
} }
@ -1320,7 +1322,7 @@ func (h *Hub) processMessageMsg(client *Client, message *ClientMessage) {
// It may take some time for the publisher (which is the current // It may take some time for the publisher (which is the current
// client) to start his stream, so we must not block the active // client) to start his stream, so we must not block the active
// goroutine. // goroutine.
go h.processMcuMessage(client, recipient, recipientSession, message, msg, clientData) go h.processMcuMessage(session, recipientSession, message, msg, clientData)
} else { // nolint } else { // nolint
// Client is not connected yet. // Client is not connected yet.
} }
@ -1573,7 +1575,7 @@ func (h *Hub) processInternalMsg(client *Client, message *ClientMessage) {
log.Printf("Session %s removed virtual session %s", session.PublicId(), sess.PublicId()) log.Printf("Session %s removed virtual session %s", session.PublicId(), sess.PublicId())
if vsess, ok := sess.(*VirtualSession); ok { if vsess, ok := sess.(*VirtualSession); ok {
// We should always have a VirtualSession here. // We should always have a VirtualSession here.
vsess.CloseWithFeedback(client, message) vsess.CloseWithFeedback(session, message)
} else { } else {
sess.Close() sess.Close()
} }
@ -1594,22 +1596,22 @@ func isAllowedToSend(session *ClientSession, data *MessageClientMessageData) boo
return session.HasPermission(permission) return session.HasPermission(permission)
} }
func sendNotAllowed(client *Client, message *ClientMessage) { func sendNotAllowed(session *ClientSession, message *ClientMessage) {
response := message.NewErrorServerMessage(NewError("not_allowed", "Not allowed to publish.")) response := message.NewErrorServerMessage(NewError("not_allowed", "Not allowed to publish."))
client.SendMessage(response) session.SendMessage(response)
} }
func sendMcuClientNotFound(client *Client, message *ClientMessage) { func sendMcuClientNotFound(session *ClientSession, message *ClientMessage) {
response := message.NewErrorServerMessage(NewError("client_not_found", "No MCU client found to send message to.")) response := message.NewErrorServerMessage(NewError("client_not_found", "No MCU client found to send message to."))
client.SendMessage(response) session.SendMessage(response)
} }
func sendMcuProcessingFailed(client *Client, message *ClientMessage) { func sendMcuProcessingFailed(session *ClientSession, message *ClientMessage) {
response := message.NewErrorServerMessage(NewError("processing_failed", "Processing of the message failed, please check server logs.")) response := message.NewErrorServerMessage(NewError("processing_failed", "Processing of the message failed, please check server logs."))
client.SendMessage(response) session.SendMessage(response)
} }
func (h *Hub) processMcuMessage(senderClient *Client, client *Client, session *ClientSession, client_message *ClientMessage, message *MessageClientMessage, data *MessageClientMessageData) { func (h *Hub) processMcuMessage(senderSession *ClientSession, session *ClientSession, client_message *ClientMessage, message *MessageClientMessage, data *MessageClientMessageData) {
ctx, cancel := context.WithTimeout(context.Background(), h.mcuTimeout) ctx, cancel := context.WithTimeout(context.Background(), h.mcuTimeout)
defer cancel() defer cancel()
@ -1632,7 +1634,7 @@ func (h *Hub) processMcuMessage(senderClient *Client, client *Client, session *C
case "offer": case "offer":
if !isAllowedToSend(session, data) { if !isAllowedToSend(session, data) {
log.Printf("Session %s is not allowed to offer %s, ignoring", session.PublicId(), data.RoomType) log.Printf("Session %s is not allowed to offer %s, ignoring", session.PublicId(), data.RoomType)
sendNotAllowed(senderClient, client_message) sendNotAllowed(senderSession, client_message)
return return
} }
@ -1642,7 +1644,7 @@ func (h *Hub) processMcuMessage(senderClient *Client, client *Client, session *C
if session.PublicId() == message.Recipient.SessionId { if session.PublicId() == message.Recipient.SessionId {
if !isAllowedToSend(session, data) { if !isAllowedToSend(session, data) {
log.Printf("Session %s is not allowed to send candidate for %s, ignoring", session.PublicId(), data.RoomType) log.Printf("Session %s is not allowed to send candidate for %s, ignoring", session.PublicId(), data.RoomType)
sendNotAllowed(senderClient, client_message) sendNotAllowed(senderSession, client_message)
return return
} }
@ -1655,29 +1657,29 @@ func (h *Hub) processMcuMessage(senderClient *Client, client *Client, session *C
} }
if err != nil { if err != nil {
log.Printf("Could not create MCU %s for session %s to send %+v to %s: %s", clientType, session.PublicId(), data, message.Recipient.SessionId, err) log.Printf("Could not create MCU %s for session %s to send %+v to %s: %s", clientType, session.PublicId(), data, message.Recipient.SessionId, err)
sendMcuClientNotFound(senderClient, client_message) sendMcuClientNotFound(senderSession, client_message)
return return
} else if mc == nil { } else if mc == nil {
log.Printf("No MCU %s found for session %s to send %+v to %s", clientType, session.PublicId(), data, message.Recipient.SessionId) log.Printf("No MCU %s found for session %s to send %+v to %s", clientType, session.PublicId(), data, message.Recipient.SessionId)
sendMcuClientNotFound(senderClient, client_message) sendMcuClientNotFound(senderSession, client_message)
return return
} }
mc.SendMessage(context.TODO(), message, data, func(err error, response map[string]interface{}) { mc.SendMessage(context.TODO(), message, data, func(err error, response map[string]interface{}) {
if err != nil { if err != nil {
log.Printf("Could not send MCU message %+v for session %s to %s: %s", data, session.PublicId(), message.Recipient.SessionId, err) log.Printf("Could not send MCU message %+v for session %s to %s: %s", data, session.PublicId(), message.Recipient.SessionId, err)
sendMcuProcessingFailed(senderClient, client_message) sendMcuProcessingFailed(senderSession, client_message)
return return
} else if response == nil { } else if response == nil {
// No response received // No response received
return return
} }
h.sendMcuMessageResponse(client, session, message, data, response) h.sendMcuMessageResponse(session, message, data, response)
}) })
} }
func (h *Hub) sendMcuMessageResponse(client *Client, session *ClientSession, message *MessageClientMessage, data *MessageClientMessageData, response map[string]interface{}) { func (h *Hub) sendMcuMessageResponse(session *ClientSession, message *MessageClientMessage, data *MessageClientMessageData, response map[string]interface{}) {
var response_message *ServerMessage var response_message *ServerMessage
switch response["type"] { switch response["type"] {
case "answer": case "answer":

4
room.go
View File

@ -590,13 +590,13 @@ func (r *Room) getParticipantsUpdateMessage(users []map[string]interface{}) *Ser
return message return message
} }
func (r *Room) NotifySessionResumed(client *Client) { func (r *Room) NotifySessionResumed(session *ClientSession) {
message := r.getParticipantsUpdateMessage(r.users) message := r.getParticipantsUpdateMessage(r.users)
if len(message.Event.Update.Users) == 0 { if len(message.Event.Update.Users) == 0 {
return return
} }
client.SendMessage(message) session.SendMessage(message)
} }
func (r *Room) NotifySessionChanged(session Session) { func (r *Room) NotifySessionChanged(session Session) {

18
virtualsession.go
View File

@ -135,16 +135,16 @@ func (s *VirtualSession) Close() {
s.CloseWithFeedback(nil, nil) s.CloseWithFeedback(nil, nil)
} }
func (s *VirtualSession) CloseWithFeedback(client *Client, message *ClientMessage) { func (s *VirtualSession) CloseWithFeedback(session *ClientSession, message *ClientMessage) {
room := s.GetRoom() room := s.GetRoom()
s.session.RemoveVirtualSession(s) s.session.RemoveVirtualSession(s)
removed := s.session.hub.removeSession(s) removed := s.session.hub.removeSession(s)
if removed && room != nil { if removed && room != nil {
go s.notifyBackendRemoved(room, client, message) go s.notifyBackendRemoved(room, session, message)
} }
} }
func (s *VirtualSession) notifyBackendRemoved(room *Room, client *Client, message *ClientMessage) { func (s *VirtualSession) notifyBackendRemoved(room *Room, session *ClientSession, message *ClientMessage) {
ctx, cancel := context.WithTimeout(context.Background(), s.hub.backendTimeout) ctx, cancel := context.WithTimeout(context.Background(), s.hub.backendTimeout)
defer cancel() defer cancel()
@ -160,9 +160,9 @@ func (s *VirtualSession) notifyBackendRemoved(room *Room, client *Client, messag
if err := s.hub.backend.PerformJSONRequest(ctx, s.ParsedBackendUrl(), request, &response); err != nil { if err := s.hub.backend.PerformJSONRequest(ctx, s.ParsedBackendUrl(), request, &response); err != nil {
virtualSessionId := GetVirtualSessionId(s.session, s.PublicId()) virtualSessionId := GetVirtualSessionId(s.session, s.PublicId())
log.Printf("Could not leave virtual session %s at backend %s: %s", virtualSessionId, s.BackendUrl(), err) log.Printf("Could not leave virtual session %s at backend %s: %s", virtualSessionId, s.BackendUrl(), err)
if client != nil && message != nil { if session != nil && message != nil {
reply := message.NewErrorServerMessage(NewError("remove_failed", "Could not remove virtual session from backend.")) reply := message.NewErrorServerMessage(NewError("remove_failed", "Could not remove virtual session from backend."))
client.SendMessage(reply) session.SendMessage(reply)
} }
return return
} }
@ -170,9 +170,9 @@ func (s *VirtualSession) notifyBackendRemoved(room *Room, client *Client, messag
if response.Type == "error" { if response.Type == "error" {
virtualSessionId := GetVirtualSessionId(s.session, s.PublicId()) virtualSessionId := GetVirtualSessionId(s.session, s.PublicId())
log.Printf("Could not leave virtual session %s at backend %s: %+v", virtualSessionId, s.BackendUrl(), response.Error) log.Printf("Could not leave virtual session %s at backend %s: %+v", virtualSessionId, s.BackendUrl(), response.Error)
if client != nil && message != nil { if session != nil && message != nil {
reply := message.NewErrorServerMessage(NewError("remove_failed", response.Error.Error())) reply := message.NewErrorServerMessage(NewError("remove_failed", response.Error.Error()))
client.SendMessage(reply) session.SendMessage(reply)
} }
return return
} }
@ -182,9 +182,9 @@ func (s *VirtualSession) notifyBackendRemoved(room *Room, client *Client, messag
err := s.hub.backend.PerformJSONRequest(ctx, s.ParsedBackendUrl(), request, &response) err := s.hub.backend.PerformJSONRequest(ctx, s.ParsedBackendUrl(), request, &response)
if err != nil { if err != nil {
log.Printf("Could not remove virtual session %s from backend %s: %s", s.PublicId(), s.BackendUrl(), err) log.Printf("Could not remove virtual session %s from backend %s: %s", s.PublicId(), s.BackendUrl(), err)
if client != nil && message != nil { if session != nil && message != nil {
reply := message.NewErrorServerMessage(NewError("remove_failed", "Could not remove virtual session from backend.")) reply := message.NewErrorServerMessage(NewError("remove_failed", "Could not remove virtual session from backend."))
client.SendMessage(reply) session.SendMessage(reply)
} }
} }
} }