Merge pull request #851 from strukturag/duplicate-virtual-sessions

Prevent duplicate virtual sessions in participant update events.

grpc_client.go

@@ -249,6 +249,36 @@ func (c *GrpcClient) IsSessionInCall(ctx context.Context, sessionId string, room
 	return response.GetInCall(), nil
 }
 
+func (c *GrpcClient) GetInternalSessions(ctx context.Context, roomId string, backend *Backend) (internal map[string]*InternalSessionData, virtual map[string]*VirtualSessionData, err error) {
+	statsGrpcClientCalls.WithLabelValues("GetInternalSessions").Inc()
+	// TODO: Remove debug logging
+	log.Printf("Get internal sessions for %s@%s on %s", roomId, backend.Id(), c.Target())
+	response, err := c.impl.GetInternalSessions(ctx, &GetInternalSessionsRequest{
+		RoomId:     roomId,
+		BackendUrl: backend.Url(),
+	}, grpc.WaitForReady(true))
+	if s, ok := status.FromError(err); ok && s.Code() == codes.NotFound {
+		return nil, nil, nil
+	} else if err != nil {
+		return nil, nil, err
+	}
+
+	if len(response.InternalSessions) > 0 {
+		internal = make(map[string]*InternalSessionData, len(response.InternalSessions))
+		for _, s := range response.InternalSessions {
+			internal[s.SessionId] = s
+		}
+	}
+	if len(response.VirtualSessions) > 0 {
+		virtual = make(map[string]*VirtualSessionData, len(response.VirtualSessions))
+		for _, s := range response.VirtualSessions {
+			virtual[s.SessionId] = s
+		}
+	}
+
+	return
+}
+
 func (c *GrpcClient) GetPublisherId(ctx context.Context, sessionId string, streamType StreamType) (string, string, net.IP, error) {
 	statsGrpcClientCalls.WithLabelValues("GetPublisherId").Inc()
 	// TODO: Remove debug logging

grpc_server.go

@@ -59,6 +59,7 @@ type GrpcServerHub interface {
 	GetSessionByResumeId(resumeId string) Session
 	GetSessionByPublicId(sessionId string) Session
 	GetSessionIdByRoomSessionId(roomSessionId string) (string, error)
+	GetRoomForBackend(roomId string, backend *Backend) *Room
 
 	GetBackend(u *url.URL) *Backend
 }
@@ -185,6 +186,52 @@ func (s *GrpcServer) IsSessionInCall(ctx context.Context, request *IsSessionInCa
 	return result, nil
 }
 
+func (s *GrpcServer) GetInternalSessions(ctx context.Context, request *GetInternalSessionsRequest) (*GetInternalSessionsReply, error) {
+	statsGrpcServerCalls.WithLabelValues("GetInternalSessions").Inc()
+	// TODO: Remove debug logging
+	log.Printf("Get internal sessions from %s on %s", request.RoomId, request.BackendUrl)
+
+	var u *url.URL
+	if request.BackendUrl != "" {
+		var err error
+		u, err = url.Parse(request.BackendUrl)
+		if err != nil {
+			return nil, status.Error(codes.InvalidArgument, "invalid url")
+		}
+	}
+
+	backend := s.hub.GetBackend(u)
+	if backend == nil {
+		return nil, status.Error(codes.NotFound, "no such backend")
+	}
+
+	room := s.hub.GetRoomForBackend(request.RoomId, backend)
+	if room == nil {
+		return nil, status.Error(codes.NotFound, "no such room")
+	}
+
+	result := &GetInternalSessionsReply{}
+	room.mu.RLock()
+	defer room.mu.RUnlock()
+
+	for session := range room.internalSessions {
+		result.InternalSessions = append(result.InternalSessions, &InternalSessionData{
+			SessionId: session.PublicId(),
+			InCall:    uint32(session.GetInCall()),
+			Features:  session.GetFeatures(),
+		})
+	}
+
+	for session := range room.virtualSessions {
+		result.VirtualSessions = append(result.VirtualSessions, &VirtualSessionData{
+			SessionId: session.PublicId(),
+			InCall:    uint32(session.GetInCall()),
+		})
+	}
+
+	return result, nil
+}
+
 func (s *GrpcServer) GetPublisherId(ctx context.Context, request *GetPublisherIdRequest) (*GetPublisherIdReply, error) {
 	statsGrpcServerCalls.WithLabelValues("GetPublisherId").Inc()
 	// TODO: Remove debug logging