실시간 화면 공유 플랫폼 만들기 With kurento media server

이 블로그는 https://terianp.tistory.com/184 를 참고하여 만들었습니다.

media server란?

• peer 간의 연결에서 각 peer들의 연결 상태를 확인 해 주는 Signaling server와 실제 media 데이터를 전달하는 STUN server 또는 TURN server가 필요하다.

• 위 그림은 media server를 구성하기 위한 3가지 방법을 표현한 것이다.
• 나는 그 중에서 1대 다 화상 채팅 및 화면 공유를 원하기 때문에 SFU 방식을 선택하였다.

핵심 코드 리뷰

KurentoRoomDto

import com.example.steam.steam.enums.ChatType;  
import com.example.steam.steam.handler.KurentoUserSession;  
import com.google.gson.JsonArray;  
import com.google.gson.JsonObject;  
import lombok.Getter;  
import lombok.NoArgsConstructor;  
import lombok.Setter;  
import lombok.extern.slf4j.Slf4j;  
import org.kurento.client.KurentoClient;  
import org.kurento.client.MediaPipeline;  
import org.springframework.web.socket.WebSocketSession;  

import java.io.IOException;  
import java.util.ArrayList;  
import java.util.Collection;  
import java.util.List;  
import java.util.concurrent.ConcurrentHashMap;  
import java.util.concurrent.ConcurrentMap;  

@Getter  
@Setter  
@NoArgsConstructor  
@Slf4j  
public class KurentoRoomDto{  
    private KurentoClient kurento;  
    private MediaPipeline pipeline;  
    private String userId;  
    private String roomId;  
    private String roomName;  
    private int userCount = 0;  
    private int maxUserCnt;  
    private ChatType chatType;  
    private boolean isHost = false;  

    private ConcurrentMap<String, KurentoUserSession> participants;  
    public void setRoomInfo(String roomId, String roomName, ChatType chatType, KurentoClient kurento, String userId){  
        this.userId = userId;  
        this.kurento = kurento;  
        this.roomId = roomId;  
        this.chatType = chatType;  
        this.roomName = roomName;  
        this.participants = new ConcurrentHashMap<>();  
    }  

    public void createPipeline(){  
        this.pipeline = this.kurento.createMediaPipeline();  
        log.info("pipline : {} ",this.pipeline);  
    }  

    public KurentoUserSession join(String name, WebSocketSession session, String roomName) throws IOException {  
        final KurentoUserSession participant = new KurentoUserSession(name, roomName, session, this.pipeline);  

        joinRoom(participant);  

        participants.put(participant.getName(), participant);  

        sendParticipantNames(participant);  
        userCount++;  

        return participant;  
    }  

    public void leave(KurentoUserSession user) throws IOException {  
        log.info("PARTICIPANT {}: Leaving room {}", user.getName(), this.roomId);  
        this.removeParticipant(user.getName());  

        user.close();  
    }  

    public void close(){  
        for (final KurentoUserSession user : participants.values()) {  
            // 유저 close            user.close();  
        } // for 문 끝  

        // 유저 정보를 담은 map - participants - 초기화  
        participants.clear();  

        pipeline.release();  
    }  

    public Collection<String> joinRoom(KurentoUserSession newParticipant){  

        final JsonObject newParticipantMsg = new JsonObject();  
        newParticipantMsg.addProperty("id", "newParticipantArrived");  
        JsonObject joNewParticipant = new JsonObject();  
        joNewParticipant.addProperty("name", newParticipant.getName());  
        newParticipantMsg.add("data", joNewParticipant);  

        final List<String> participantsList = new ArrayList<>(participants.values().size());  
        log.debug("ROOM {}: 다른 참여자들에게 새로운 참여자가 들어왔음을 알림 {} :: {}", roomId,  
                newParticipant.getName(), newParticipant.getRoomName());  

        for (final KurentoUserSession participant: participants.values()) {  
            try{  
                participant.sendMessage(newParticipantMsg);  
            }  
            catch (IOException e){  
                log.error("ROOM {}: participant {} could not be notified", roomId, participant.getName(), e);  
            }  

            participantsList.add(participant.getName());  
        }  
        return participantsList;  


    }  

    private void removeParticipant(String name) throws IOException {  

        // participants map 에서 제거된 유저 - 방에서 나간 유저 - 를 제거함  
        participants.remove(name);  

        log.debug("ROOM {}: notifying all users that {} is leaving the room", this.roomId, name);  

        // String list 생성  
        final List<String> unNotifiedParticipants = new ArrayList<>();  

        // json 객체 생성  
        final JsonObject participantLeftJson = new JsonObject();  

        // json 객체에 유저가 떠났음을 알리는 jsonObject        // newParticipantMsg : { "id" : "participantLeft", "name" : "참여자 유저명"}  
        participantLeftJson.addProperty("id", "participantLeft");  
        participantLeftJson.addProperty("name", name);  

        // participants 의 value 로 for 문 돌림  
        for (final KurentoUserSession participant : participants.values()) {  
            try {  
                // 나간 유저의 video 를 cancel 하기 위한 메서드  
                participant.cancelVideoFrom(name);  

                // 다른 유저들에게 현재 유저가 나갔음을 알리는 jsonMsg 를 전달  
                participant.sendMessage(participantLeftJson);  

            } catch (final IOException e) {  
                unNotifiedParticipants.add(participant.getName());  
            }  
        }  

        // 만약 unNotifiedParticipants 가 비어있지 않다면  
        if (!unNotifiedParticipants.isEmpty()) {  
            log.debug("ROOM {}: The users {} could not be notified that {} left the room", this.roomId,  
                    unNotifiedParticipants, name);  
        }  

    }  

    public void sendParticipantNames(KurentoUserSession user) throws IOException {  
        final JsonObject existingParticipantsMsg = new JsonObject();  

        final JsonArray participantsArray = new JsonArray();  

        for (final KurentoUserSession participant: this.getParticipants().values()) {  
            if(!participant.getName().equals(user.getName())){  
                JsonObject exisingUser  = new JsonObject();  
                exisingUser.addProperty("name", participant.getName());  

                participantsArray.add(exisingUser);  
            }  
        }  

        existingParticipantsMsg.addProperty("id", "existingParticipants");  
        existingParticipantsMsg.add("data", participantsArray);  
        log.debug("PARTICIPANT {}: sending a list of {} participants", user.getName(), participantsArray.size());  

        // user 에게 existingParticipantsMsg 전달  
        user.sendMessage(existingParticipantsMsg);  
    }  

    public void sendHostIsOut(){  
        final JsonObject msg = new JsonObject();  
        msg.addProperty("id", "hostExit");  
        msg.addProperty("name", "name");  
        for (final KurentoUserSession participant : participants.values()) {  
            try {  
                // 다른 유저들에게 현재 유저가 나갔음을 알리는 jsonMsg 를 전달  
                participant.sendMessage(msg);  

            } catch (final IOException e) {  
                log.info("host exit error");  
            }  
        }  
    }  




}

• 진짜 저 위의 블로그가 없었다면 구현할 수 없었을 것 같았다.
• Websocket 기반으로 Signaling server로 구현하였다.
• 위 코드에서는 방에 사람이 들어오고 나갔을 때에 처리 방법과, 방을 만든 Host가 방을 떠났을 시에 대해 어떻게 처리해야 되는지 대한 로직이 담겨있다.

KurentoHandler

import com.example.steam.steam.dto.ChatRoomMap;  
import com.example.steam.steam.dto.KurentoRoomDto;  
import com.example.steam.steam.service.KurentoManager;  
import com.example.steam.steam.service.KurentoRegistryService;  
import com.google.gson.Gson;  
import com.google.gson.GsonBuilder;  
import com.google.gson.JsonObject;  
import lombok.RequiredArgsConstructor;  
import lombok.extern.slf4j.Slf4j;  
import org.kurento.client.IceCandidate;  
import org.springframework.web.socket.CloseStatus;  
import org.springframework.web.socket.TextMessage;  
import org.springframework.web.socket.WebSocketSession;  
import org.springframework.web.socket.handler.TextWebSocketHandler;  

import java.io.IOException;  
import java.util.Objects;  

@Slf4j  
@RequiredArgsConstructor  
public class KurentoHandler extends TextWebSocketHandler {  
    private static final Gson gson = new GsonBuilder().create();  


    private final KurentoRegistryService registry;  

    private final KurentoManager roomManger;  


    @Override  
    protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {  
        final JsonObject jsonMessage = gson.fromJson(message.getPayload(), JsonObject.class);  
        KurentoUserSession user = null;  
        try{  
            user = registry.getBySession(session);  
        }  
        catch (Exception e){  
            if(user != null){  
                log.info("Incoming message from user '{}': {}", user.getName(), jsonMessage);  
            }  
            else{  
                log.info("Incoming message from new user '{}'", jsonMessage);  
            }  
        }  

        switch (jsonMessage.get("id").getAsString()){  
            case "joinRoom": // value : joinRoom 인 경우  
                joinRoom(jsonMessage, session); // joinRoom 메서드를 실행  
                break;  

            case "receiveVideoFrom": // receiveVideoFrom 인 경우  
                try {  
                    // sender 명 - 사용자명 - 과  
                    final String senderName = jsonMessage.get("sender").getAsString();  
                    // 유저명을 통해 session 값을 가져온다  
                    final KurentoUserSession sender = registry.getByName(senderName);  
                    // jsonMessage 에서 sdpOffer 값을 가져온다  
                    final String sdpOffer = jsonMessage.get("sdpOffer").getAsString();  

                    user.receiveVideoFrom(sender, sdpOffer);  
                } catch (Exception e){  
                    connectException(user, e);  
                }  
                break;  

            case "leaveRoom": // 유저가 나간 경우  
                leaveRoom(user);  
                break;  
            case "onIceCandidate": // 유저에 대해 IceCandidate 프로토콜을 실행할 때  
                JsonObject candidate = jsonMessage.get("candidate").getAsJsonObject();  

                if (user != null) {  
                    IceCandidate cand = new IceCandidate(candidate.get("candidate").getAsString(),  
                            candidate.get("sdpMid").getAsString(), candidate.get("sdpMLineIndex").getAsInt());  
                    user.addCandidate(cand, jsonMessage.get("name").getAsString());  
                }  
                break;  

            default:  
                break;  
        }  
    }  

    @Override  
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {  
        KurentoUserSession user = registry.removeBySession(session);  
        this.leaveRoom(user);  
    }  



    private void joinRoom(JsonObject params, WebSocketSession session) throws IOException {  
        final String roomName = params.get("room").getAsString();  
        final String name = params.get("name").getAsString();  
        log.info("participant '{}': trying to join room {}", name, roomName);  

        KurentoRoomDto room = roomManger.getRoom(roomName);  

        final KurentoUserSession user = room.join(name, session, roomName);  

        registry.register(user);  
    }  

    private void leaveRoom(KurentoUserSession user) throws IOException {  

        if(Objects.isNull(user)) return;  

        final KurentoRoomDto room = roomManger.getRoom(user.getRoomName());  

        if(room.getParticipants() == null) return;  
        if(!room.getParticipants().containsKey(user.getName())) return;  


        if(isHost(room.getRoomName(), user.getName())){  
            room.sendHostIsOut();  
        }  
        room.leave(user);  
        room.setUserCount(room.getUserCount() - 1);  
        if(room.getUserCount() == 0){  
            roomManger.removeRoom(room);  
        }  
    }  


    private boolean isHost(String roomName, String name){  
        if(ChatRoomMap.getInstance().getChatRooms().get(roomName).getUserId().equals(name)){  
            return true;  
        }  
        return false;  
    }  

    private void connectException(KurentoUserSession user, Exception e) throws IOException {  
        JsonObject message = new JsonObject();  
        message.addProperty("id", "ConnectionFail");  
        message.addProperty("data", e.getMessage());  

        user.sendMessage(message);  

    }  
}

• Websocket 연결로 첫 Kurento와 연결될 때, 종료 등 다른 이벤트가 발생할 때 메세지 타입에 따라 로직을 구현하였다.
• 실제 peer간에 미디어 데이터 통신은 소켓 연결 수 pipeline을 통해 kurento media server에서 통신하므로 Spring server의 부담이 줄어들게 된다.

 

이 글을 쓰면서 다시 한번 Media server에 대해 블로그를 작성해준 분께 감사합니다.
너무 어렵다.