1번,5번은 한 게임당 한 번만 발생하지만 2~4번의 이벤트는 사용자가 설정한 Round 반복 회수에 따라서 1회에서 5회까지 반복될 수 있었다. 서버는 시간에 따른 이벤트 발생을 처리하고 클라이언트들에게 메세지를 보내줘야 했다. Thread.timesleep(); 으로 동기 처리를 하는 것은 쓰레드가 멈추는 문제가 있으니 비동기로 이벤트를 발생시켜야 했다. 스프링의 batch나 @schedule을 활용하는 방법도 찾아 봤으나 문제는

1. 일정 시간 동안 이벤트가 발생하는 것은 좋은데 특정 반복 후에 멈추는 기능이 없다는 점.

2. 사용자의 게임 반복 회수, Stage 설정 시간에 따라서 유동적으로 적용하기 어렵다는 점.

   이 2가지의 문제로 다른 비동기 Schedule Executer Service를 활용하기로 했다.

   ScheduledExecutorService

   Schedule Executer Service는 Executers 쓰레드를 생성해서 사용해야 한다.

   private final ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();

   Executors는 쓰레드 풀을 활용해도 괜찮지만 이는 스프링 쓰레드로 관리되지 않는 쓰레드라는 것과 Scheduling만 시킬것이기 때문에 SingleThread만 사용했다.

ScheduleFuture 또한 Future를 상속받은 인터페이스다. 따라서 비동기로 처리되는 것을 알 수 있다. 나는 schedule 메소드를 활용했는데 delay만큼의 시간이 지나면 Runnable 또는 Callable이 실행된다. (Callable은 리턴이 있고 Runnable은 리턴 값이 없는 인터페이스다)

    //게임 시작 Future
    public CompletableFuture<Integer> startGame(int gameId, int round) throws BaseException {
        log.info("{} game start after 5 sec : {}", gameId, LocalDateTime.now());
        notifyRemainingTime(gameId, 5, 1,0, ServerEvent.NOTIFY_LEFT_TIME);

        sendingOperations.convertAndSend("/game/sse/" + gameId,
                new ServerSendEvent(ServerEvent.START, round)); // #1201 game start
        CompletableFuture<Integer> future = new CompletableFuture<>();
        executorService.schedule(() -> {
            future.complete(round);
        }, 5, TimeUnit.SECONDS);
        return future;
    }

    public CompletableFuture<Integer> scheduleFuture(int gameId, int delayTime) throws BaseException {
        CompletableFuture<Integer> future = new CompletableFuture<>();
        executorService.schedule(() -> {
            future.complete(gameId);
        }, delayTime, TimeUnit.SECONDS);
        return future;

스케줄링이 연속적으로 이어져야 하는데 ScheduleFuture는 CompletableFuture처럼 조합할 수 없어서 return을 CompletableFuture로 반환하도록 만들었다.

executorService.schedule의 delayTime이 지나면 Runnable이 호출된다. 이 때 CompletableFuture의 future.complete가 호출되고 CompletableFuture에서 다음 Future를 실행한다.

한 라운드당 조합되는 ScheduleFuture

public CompletableFuture<Integer> roundScheduler(int gameId, GameStartRequestDTO gameStartRequestDTO, int currentRound) {
        CompletableFuture<Integer> future = new CompletableFuture<>();

        // Round 시작
        log.info("{} Round {} Start: {}", gameStartRequestDTO.getGameId(), currentRound, LocalDateTime.now());
        sendingOperations.convertAndSend("/game/sse/" + gameId,
                new ServerSendEvent(ServerEvent.ROUND_START, currentRound)); // Game Start # 12

        notifyRemainingTime(gameId, gameStartRequestDTO.getStage1Time(), 1, currentRound, ServerEvent.NOTIFY_LEFT_TIME);
        scheduleFuture(gameId, gameStartRequestDTO.getStage1Time())
                .thenCompose(r -> { // Round Start stage 1
                    log.info("{} game stage 1 End : {}", gameStartRequestDTO.getGameId(), LocalDateTime.now());
                    sendingOperations.convertAndSend("/game/sse/" + gameId,
                            new ServerSendEvent(ServerEvent.STAGE_1_END, currentRound)); // send Hint and Stage 1 End # 1203
                    notifyRemainingTime(gameId, gameStartRequestDTO.getStage2Time(), 2, currentRound, ServerEvent.NOTIFY_LEFT_TIME);
                    return scheduleFuture(gameId, gameStartRequestDTO.getStage2Time());  // Stage 2 기다리기
                }).thenCompose(r -> { // Stage 2
                    log.info("{} game stage 2 End : {}", gameStartRequestDTO.getGameId(), LocalDateTime.now());
                    sendingOperations.convertAndSend("/game/sse/" + gameId,
                            new ServerSendEvent(ServerEvent.STAGE_2_END, currentRound)); // Stage 2 End go To Score # 1204
                    // 2스테이지 종료 > 해당 라운드 결과 집계
                    RoundFinishRequestDTO finishRequestDTO = new RoundFinishRequestDTO(gameStartRequestDTO.getSenderNickname(), gameStartRequestDTO.getSenderGameId(), gameStartRequestDTO.getSenderTeamId(), currentRound);
                    gameService.finishRound(finishRequestDTO);

                    notifyRemainingTime(gameId, gameStartRequestDTO.getScorePageTime(), 3, currentRound, ServerEvent.NOTIFY_LEFT_TIME);
                    return scheduleFuture(gameId, gameStartRequestDTO.getScorePageTime());
                }).thenCompose(r -> {
                    log.info("{} game {} Round End  : {}", gameStartRequestDTO.getGameId(), currentRound, LocalDateTime.now());
                    sendingOperations.convertAndSend("/game/sse/" + gameId,
                            new ServerSendEvent(ServerEvent.ROUND_END, currentRound)); // Round End stage 1, 2 score # 1205

                    notifyRemainingTime(gameId, 1, 4, currentRound, ServerEvent.NOTIFY_LEFT_TIME);
                    return scheduleFuture(gameId, 1);
                }).thenRun(() -> {
                    future.complete(currentRound);
                });
        return future;
    }

한 라운드당 scheduleFuture가 여러개가 적용된다.

ScheduleFuture에 CompletableFuture를 결합해서 스케줄링이 연속적으로 진행될 수 있게 했다.

    @DisplayName("시작 시간 포함 라운드 시간 관리")
    @Test
    public void testGameProcessAsync() throws ExecutionException, InterruptedException {
        // given
        LocalDateTime startTime = LocalDateTime.now();
        AtomicReference<LocalDateTime> asyncEndTime = new AtomicReference<>();
        GameStartRequestDTO requestDTO = new GameStartRequestDTO("testUser", 1, 1, 1, 3, 2, 3, 4);
        int currentRound = 0;
        int roundCount = requestDTO.getRoundCount();
        AtomicBoolean isPass = new AtomicBoolean(false);
        AtomicInteger executedRounds = new AtomicInteger(0);

        // when
        scheduleProvider.startGame(requestDTO.getGameId(), currentRound)
                .thenCompose(v -> {
                    // 초기 체인 생성
                    CompletableFuture<Integer> roundChain = CompletableFuture.completedFuture(currentRound);

                    // 각 라운드에 대한 비동기 작업 체인 구축
                    for (int round = 1; round <= roundCount; round++) {
                        final int currentRoundInLoop = round;
                        roundChain = roundChain.thenCompose(ignored -> {
                            executedRounds.incrementAndGet(); // 라운드 실행 횟수 증가
                            return scheduleProvider.roundScheduler(requestDTO.getGameId(), requestDTO, currentRoundInLoop);
                        });
                        RoundFinishRequestDTO finishRequestDTO = new RoundFinishRequestDTO(requestDTO.getSenderNickname(), requestDTO.getSenderGameId(), requestDTO.getSenderTeamId(), currentRound);
                        gameService.finishRound(finishRequestDTO);
                    }
                    // 마지막 결과를 설정
                    return roundChain;
                }).get();

        log.info(startTime + " Async operation ended at: "+ LocalDateTime.now());
        asyncEndTime.set(LocalDateTime.now());
        long compare = ChronoUnit.SECONDS.between(startTime, asyncEndTime.get());
        long expectedTime = (2 + 3 + 4 + 1) * 3 + 5; // 각 스테이지 시간의 합 (초 단위로 계산) / stage 1, 2, score, wait

        // 비교를 위해 오차 범위 설정
        long tolerance = 1;
        long minTime = expectedTime - tolerance;
        long maxTime = expectedTime + tolerance;

        isPass.set(true);
        assertEquals(roundCount, executedRounds.get(), "roundScheduler 호출 횟수가 예상한 반복 횟수와 다릅니다.");
        assertTrue(compare >= minTime && compare <= maxTime, "scheduler가 예상 시간 범위에서 벗어났습니다. "+ compare +" "+ expectedTime);

    }

해당 Schedule의 기능들이 제대로 작동하는지 확인하기 위한 테스트 코드를 작성했다. 오차를 1초로 설정하긴 했으나 밀리세컨즈 단위의 오차가 있었다.

Java Future

자바는 1.5부터 Future 인터페이슬 통해 비동기를 제공했다. 초기 Future는 문제가 한계가 있었다. Future 작업 자체는 비동기로 될 지라도 그 결과값을 받아서 처리하기 위해선 get()을 통해 return 값을 활용할 수 있었다. get()이 없다면 return 값을 받을 수가 없었는데 막상 get()을 사용하면 future 작업이 끝날 때까지 기다린다. 따라서 비동기가 아니라 동기가 되는 것이다.

Callable<Integer> task = () -> {
    Thread.sleep(1000);  // 1초동안 대기
    return 123;  // 결과 반환
};

FutureTask<Integer> future = new FutureTask<>(task);
new Thread(future).start();  // Future 작업 실행

// 다른 작업 ...

Integer result = future.get();  // 결과가 준비될 때까지 Blocking
System.out.println("Result: " + result);

이 문제를 해결하기 위해 Java 1.8버전에서 CompletableFuture가 등장했다.

CompletableFuture

CompletableFuture는 비동기 작업 실행, 작업 콜백, Future의 조합 등의 기능을 제공한다. 따라서 여러 비동기 작업을 하나의 흐름으로 만들고 적용할 수 있다. CompletableFuture를 사용하는 방법은 여러 가지가 있다. 나는 thenCompose와 thenRun을 주로 사용했으나 개발 환경에 따라 적용하는 것이 적절하다.

 scheduleProvider.startGame(gameInitVO.getGameId(), currentRound)
                .thenCompose(v -> {
                    // IntStream으로 라운드 수만큼 체인 생성
                    CompletableFuture<Integer> roundChain = CompletableFuture.completedFuture(currentRound);

                    // 각 라운드에 대해 체인에 비동기 작업을 연결
                    for (int round = 1; round <= gameStartRequestDTO.getRoundCount(); round++) {
                        final int currentRoundInLoop = round;
                        roundChain = roundChain.thenCompose(ignored -> scheduleProvider.roundScheduler(gameId, gameStartRequestDTO, currentRoundInLoop));

                    }

                    // 마지막 결과를 `CompletableFuture<Void>`로 변환
                    return scheduleProvider.scheduleFuture(gameId, gameExistLimitTime);
                }).thenRun(() -> {
                    gameService.finishGame(new SocketDTO(gameStartRequestDTO.getSenderNickname(), gameId, gameStartRequestDTO.getSenderTeamId()));

                    log.info("{} Game is dead at {}", gameId, LocalDateTime.now());
                    gameManager.removeGame(gameId);
                }).exceptionally(ex -> {
                    log.error("Error occurred in the CompletableFuture chain: ", ex);
                    throw new BaseException(BaseResponseStatus.OOPS, gameId);
                });

이 외에도 후술할 schedule Future를 사용한 Schedule Executer Service의 schedule을 사용했다. scheduleProvider로 커스텀했다.

CompletableFuture 설명 CompletableFuture를 잘 설명해주신 블로그가 있어서 많이 참고했다.

Spring은 WebClient를 사용해서 HTTP request를 보내도록 권장하고 관련 레퍼런스를 제공하고 있다. WebClient WebClient는 Reactive 방식의 비동기 요청으로 처리된다.

[문제 조건]

1. 1월부터 12월까지의 기상 정보를 받아서 최고 값과 최소 값의 차이를 구해야 함.

2. 한 번의 요청당 1개월의 데이터만 받을 수 있음.

   해당 프로젝트에서는 1월부터 12월까지의 12개월의 데이터가 필요했는데 외부 서비스에서는 한번에 12개월의 데이터를 받을 수 없고 12번의 요청을 보내야만 했다.

   한 번의 요청당 약 2초 가량의 시간이 소요됐는데 동기적으로 보낼 경우 총 24초가 걸렸다. WebClient의 비동기 특징을 활용해 12월까지의 데이터 요청을 한번에 보냈다. 모든 데이터를 한 번에 비교해야 했다. 때문에 비동기 요청이 다 처리되면 이후 작업을 진행하기 위해서 Mono<List>꼴로 데이터를 처리했다.

public  Mono<List<WeatherAPIResponseDTO>> fetchWeatherDataWithMonth(float lat, float lon) {
        Flux<Integer> months = Flux.range(1, 12);
        Mono<List<WeatherAPIResponseDTO>> responses = months.flatMap(month -> webClient.get()
                        .uri(uriBuilder -> uriBuilder
                                .path("/month")
                                .queryParam("lat", lat)
                                .queryParam("lon", lon)
                                .queryParam("month", month)
                                .queryParam("appid", weatherAPI)
                                .build())
                        .retrieve()
                        .bodyToMono(WeatherAPIResponseDTO.class))
                .collectList()  // 모든 결과를 리스트로 모음
                .doOnSuccess(dataList -> {
                    System.out.println("Received Data: " + dataList);
                })
                .doOnError(error -> {
                    System.out.println("Error occurred: " + error.getMessage());
                });
        return responses;


    }

Mono<List>로 받은 데이터라서 streamAPI를 활용해서 최대값과 최소값을 구했다. StreamAPI가 익숙치 않아서 두 번에 걸쳐서 데이터를 모았는데 한번의 Stream에서 최대 최소를 비교하면 더 나을 것 같다.

    public Mono<Double> getAnnualTemperatureRange( Mono<List<WeatherAPIResponseDTO>> weatherDat) {
        Mono<Double> annualTemperatureRange = weatherDat.map(dataList -> {
            double highest = dataList.stream().map(data -> data.getResult().getTemp().getRecordMax())
                    .max(Comparator.naturalOrder()).orElse(Double.NEGATIVE_INFINITY);

            double lowestTemperature = dataList.stream()
                    .map(weatherData -> weatherData.getResult().getTemp().getRecordMin())
                    .min(Comparator.naturalOrder())
                    .orElse(Double.POSITIVE_INFINITY);

            return highest - lowestTemperature;
        });
        return annualTemperatureRange;
    }

import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.config.annotation.*;
import org.springframework.messaging.simp.config.MessageBrokerRegistry;


@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer  {

    public void configureMessageBroker(MessageBrokerRegistry config) {
        config.enableSimpleBroker(new String[]{"/topic","/busker","/audience"}); // sub
        config.setApplicationDestinationPrefixes(new String[]{"/app"});
    }


// EndPoint를 등록하기. Pub/Sub이전에 EndPoint로 채널을 분리함.
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        registry.addEndpoint(new String[]{"/api/chat"}).setAllowedOriginPatterns("*");
        registry.addEndpoint(new String[]{"/api/signal"}) 
                .setAllowedOriginPatterns("*");
    }


}

    // publishing 하는 쪽에서 prefix를 붙였을 때 라우팅된다.
    // 메세지를 받아서 가공하려고 할 때 사용함.
    @MessageMapping("/api/busker") // 메세지 받는 경로
    public void listenTestStomp(@Payload String message) {
        System.out.println(message);
//        System.out.println("test");
        HashMap<String, String> map = new HashMap<>();
        map.put("test", "test");

        simpMessagingTemplate.convertAndSend("/busker", map); // 메세지 보내기
        return;
    }

    @MessageMapping("/api/busker/{buskerName}")
    public void listenBusker(@DestinationVariable String buskerName, @Payload String message) {
        System.out.println(buskerName + " " + message);
        HashMap<String, String> map = new HashMap<>();
        map.put("buskerName", "test success");

        simpMessagingTemplate.convertAndSend("/busker/" + buskerName, map);
        return;
    }

  import { Client } from '@stomp/stompjs';

// 연결하기 -> 공식 문서 내용
const client = new StompJs.Client({
  brokerURL: 'ws://localhost:15674/ws',
  connectHeaders: {
    login: 'user',
    passcode: 'password',
  },
  debug: function (str) {
    console.log(str);
  },
  reconnectDelay: 5000,
  heartbeatIncoming: 4000,
  heartbeatOutgoing: 4000,
});

client.onConnect = function (frame) {
  // Do something, all subscribes must be done is this callback
  // This is needed because this will be executed after a (re)connect
};

client.onStompError = function (frame) {
  // Will be invoked in case of error encountered at Broker
  // Bad login/passcode typically will cause an error
  // Complaint brokers will set `message` header with a brief message. Body may contain details.
  // Compliant brokers will terminate the connection after any error
  console.log('Broker reported error: ' + frame.headers['message']);
  console.log('Additional details: ' + frame.body);
};

client.activate();


=====================================================================
내가 짠 코드

const clientRef = useRef(
        new StompJS.Client({
            brokerURL: `${process.env.REACT_APP_API_WEBSOCKET_BASE_URL}`,
        })
    );
const client = clientRef.current;

client.publish({ // 메세지 보내기 body에 메세지 담기
  destination: `/app/api/busker/${userId}/offer`,
  body: JSON.stringify({
    userId,
    offer,
  })
})


client.onConnect = (frame) => { // sub는 onconnect 이벤트에 걸어두는 것이 안정적이라고 봄.
  console.log("streaming stomp : " + frame);
  // sdpOffer를 보내고 Answer를 받음

  client.subscribe(`/busker/${userId}/sdpAnswer`, (res) => { //메세지 받기 res로 메세지 받음.
    const offerResponse = JSON.parse(res.body);
    const answerId = offerResponse.id;
    const response = offerResponse.response;
    const sdpAnswer = offerResponse.sdpAnswer;

    client.subscribe(`/busker/${userId}/iceCandidate`, (res) => {
      const iceResponse = JSON.parse(res.body);
      if (iceResponse.id === "iceCandidate") {
        console.log(koreaTime + " server send ice \n" + iceResponse.candidate.candidate)
        const icecandidate = new RTCIceCandidate(iceResponse.candidate)
        }
    })
  }
                   client.onStompError = (frame) => {
    console.log('Broker reported error: ' + frame.headers['message']);
    console.log('Additional details: ' + frame.body);
  };



// 연결 끊기
client.deactivate().then(r => console.log(r)); // Close the WebSocket connection

@Service
public class ImageService {
....

private final ManagedChannel channel
            = ManagedChannelBuilder.forTarget("localhost:9090").usePlaintext().build();

    private final CreateImageGrpc.CreateImageBlockingStub imageStub
            = CreateImageGrpc.newBlockingStub(channel);

    public ByteArrayResource sendImage(MultipartFile image, ImageOption imageOption) throws IOException {
        ByteString imageData = ByteString.copyFrom(image.getBytes());
        BufferedImage bufferedImage = null;
        Image.Options options = Image.Options.newBuilder()
                .setBackground(imageOption.getBackground())
                .setHair(imageOption.getHair())
                .setSuit(imageOption.getSuit()).build();

        Image.ProcessedImageInfo receiveData =
                this.imageStub.sendImage(Image.OriginalImageInfo.newBuilder()
                        .setOriginalImage(imageData)
                        .setOptions(options)
                        .build());

syntax = "proto3";

package com.ssafy.pjt.grpc;

service CreateImage {
  rpc sendImage (OriginalImageInfo) returns (ProcessedImageInfo);
}

message OriginalImageInfo {
  bytes originalImage = 1;
  Options options = 2;
}

message processedImage{
  string name = 1;
  bytes image = 2;
}


....

Controller가 Image를 보내주면 proto 포멧에 맞춰서 bytes로 변환해야한다. image 파일은 bytes로 날아오긴 했지만 타입이 MultipartFile이기 때문에 ByteString으로 변환해주고 Python 서버에 전송한다. Python 서버에서 이미지 처리가 끝나면 Spring서버에 bytes로 다시 보내준다.

** (Client)MultipartFile ->(Spring) bytes ->(Python) bytes -> (Spring)**

MutipartFile을 쉽게 byte[]로 변환하고 Python에서 이미지 처리를 했으니 Spring에서 byte[]파일을 받으면 쉽게 Image 파일로 변환할 수 있을 것 같았는데 그게 안됨

byte[]를 Image로 다시 복구시키는 작업이 필요했음. byte를 width*height Image로 복구 시켜야하는데 파이썬으로 이미지를 처리하면 width, height 정보가 사라져서 Java가 눈치껏 복구할 수가 없고 그저 byte[]의 데이터로만 사용할 수가 있음. 그래서 width, height, 색타입을 주고 byte[]를 BufferedImage타입으로 만듬. 이후 픽셀에 맞는 색 주입.

gRPC로 받은 데이터를 복구하는 메소드 getBufferedImage 호출

 byte[] processedImageData = receiveData.getProcessedImage().toByteArray();
            ByteArrayResource byteArrayResource = getBufferedImage(processedImageData,768,1024);
            Image.ResponseUrl responseUrl = receiveData.getResponseUrl();

byte[] to ByteArrayResource로 복구

    private ByteArrayResource getBufferedImage(byte[] processedImageData,int width, int height) throws IOException {
        BufferedImage bufferedImage = new BufferedImage(width, height, BufferedImage.TYPE_3BYTE_BGR);

        // BufferedImage에 byte 배열 데이터 채우기
        int index = 0;
        for (int y = 0; y < bufferedImage.getHeight(); y++) {
            for (int x = 0; x < bufferedImage.getWidth(); x++) {
                int red = processedImageData[index++] & 0xFF;
                int green = processedImageData[index++] & 0xFF;
                int blue = processedImageData[index++] & 0xFF;

                // RGB 값으로 Pixel 생성 및 설정
                bufferedImage.setRGB(x, y, new Color(blue, green, red).getRGB());
            }
        }

        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        ImageIO.write(bufferedImage, "jpg", baos);

        ByteArrayResource resource = new ByteArrayResource(baos.toByteArray());

        return resource;
    }

width와 height는 파이썬에서 만들기로한 이미지 사이즈를 미리 정해놓았음. static 변수로 설정해도 좋았을 것 같음.

1. Python이 여러 요청을 한번에 처리하려고 하는데 Image processing에 어차피 시간 오래 걸리는데 한번에 하다보니 하나도 만들지 못하고 time out이 발생해버렸음.
2. 이럴바에 하나씩 완성해서 응답하기로 함.
3. gRPC pool size를 1개로 만들어서 spring에서 요청을 동기적으로 보내도록 만들었음.

이미지 사이즈도 client에서 resize해서 작게 만들었음. python 처리 시간을 줄이기 위해서.

gRPC stub을 통해 요청이 마구마구 날아갈 수 있는데 stub 객체를 하나만 쓸 수 있게 만듬.

이렇게 만들면 gRPC를 쓰는 이유도 없고 서버가 한번에 한 요청만 처리할 수 있는 몹시 나쁜 구조지만 AI 이미지 생성이 느려서 모든 요청을 처리하지 못하는 것보다는 낫다는 판단으로 어쩔 수 없었음.

추후에 GPU를 연결해서 AI처리 속도를 높이면 그때는 설계대로 복구할 예정

grpc Stub

 public class GrpcStubPool {
    private final ManagedChannel channel;
    private final BlockingQueue<CreateImageGrpc.CreateImageBlockingStub> stubPool;
    private final String aiUrl = System.getenv("AI_URL");
    private final int poolSize = 1;
    public GrpcStubPool() {
        this.channel = ManagedChannelBuilder.forTarget(aiUrl).usePlaintext().build();
        this.stubPool = new ArrayBlockingQueue<>(poolSize);

        for (int i = 0; i < poolSize; i++) {
            CreateImageGrpc.CreateImageBlockingStub stub = CreateImageGrpc.newBlockingStub(channel);
            stubPool.offer(stub);
        }
    }

    public CreateImageGrpc.CreateImageBlockingStub getStub() throws InterruptedException {
        return stubPool.take();
    }

    public void returnStub(CreateImageGrpc.CreateImageBlockingStub stub) throws InterruptedException {
        stubPool.put(stub);
    }

    @PreDestroy
    public void close() {
        channel.shutdown();
    }
}

파이썬이 이미지 하나 만들어서 response를 보내면 해당 service가 stub을 반납함. 기다리던 다른 service가 stub을 가져가서 사용하고 response 받으면 stub을 반납함. pool size를 하나로 만들었음.

gRPC Client

 try {
            imageStub = grpcStubPool.getStub();
            Image.OriginalImageInfo buildImageInfo = Image.OriginalImageInfo.newBuilder()
                    .setOriginalImage(imageData)
                    .setOptions(options)
                    .build();
            System.out.println(buildImageInfo.getOptions().getSex());
            receiveData = imageStub.sendImage(buildImageInfo);

        } catch (IllegalStateException | InterruptedException e) {
            throw ApiExceptionFactory.fromExceptionEnum(ImageExceptionEnum.GRPC_ERROR);
        }

        if (Image.ImageProcessingResult.SUCCESS.equals(receiveData.getResult())) {
            byte[] processedImageData = receiveData.getProcessedImage().toByteArray();
            ByteArrayResource byteArrayResource = getBufferedImage(processedImageData,768,1024);
            Image.ResponseUrl responseUrl = receiveData.getResponseUrl();

            ImageInfo imageInfo = new ImageInfo(
                    userId,
                    responseUrl.getOriginalImageUrl(),
                    responseUrl.getThumbnailImageUrl(),
                    responseUrl.getProcessedImageUrl(),
                    optionStore.get()
                    );

            ImageInfo insertResult = imageRepository.insertImageUrls(imageInfo,optionStore.get());
            log.info("DB insert Image info : " + insertResult.getImageInfoId());

            CreateImageDto imageInfoDto = new CreateImageDto(
                    imageInfo.getImageInfoId(),
                    imageInfo.getThumbnailImageUrl(),
                    imageInfo.getOriginalImageUrl(),
                    imageInfo.getProcessedImageUrl(),
                    byteArrayResource
            );
            try{
                grpcStubPool.returnStub(imageStub);
            } catch (InterruptedException e){
                ApiExceptionFactory.fromExceptionEnum(GrpcExceptionEnum.NO_STUB);
            }

            return imageInfoDto;

Rust는 클래스가 아니라 struct를 사용한다. struct의 원소를 console에서 보고 싶은 개발자를 위한 impl을 제공하는데 그것이 Display와 Debug다.

Display는 자바의 toString과 유사하게 개발자가 출력문을 커스텀할 수 있다. Debug는 struct를 디버깅하기 위해서 struct + 원소들을 출력하는 기능을 한다.(이것도 본인 커스텀할 수 있긴 하다.)

// #[derive(Debug)]이걸 통해서 개발자가 fmt출력문을 따로 만들지 않아도 debug 출력문의
// 포멧을 사용하면 struct의 이름과 원소들이 출력된다.
#[derive(Debug)]
struct Point2D {
    x: f64,
    y: f64,
}

// fmt::Display를 impl하고 fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result{}
// 의 내용물을 채우면 struct를 출력할 수 있다. 이때 fmt의 write가 적용되어서 출력된다.
impl fmt::Display for Point2D {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        // Customize so only `x` and `y` are denoted.
        write!(f, "x: {}, y: {}", self.x, self.y)
    }
}

let point = Point2D { x: 3.3, y: 7.2 };

println!("Compare points:");
println!("Display: {}", point);
println!("Debug: {:?}", point);

std::fmt contains many traits which govern the display of text. The base form of two important ones are listed below:

fmt::Debug: Uses the {:?} marker. Format text for debugging purposes. fmt::Display: Uses the {} marker. Format text in a more elegant, user friendly fashion.

** 디버깅하고 싶으면 Debug를 사용하고 특정 원소들만 출력하고 싶으면 Display를 사용 ** 그 외에도 fmt 관련 함수들이 많음. 공식문서 잘 볼 것.

https://leetcode.com/problems/subarrays-with-k-different-integers

배열 nums와 int k가 주어지면 nums의 부분 집합(subarray) 중 특정 조건을 만족하는 부분 집합 개수의 합을 구하라.

특정 조건은 부분 집합의 원소들은 모두 k개의 다른 원소로만 이루어져야한다. 부분 집합(sub array)는 nums의 연속 배열이다. A subarray is a contiguous part of an array. 1,4,6이 있다면 sub array의 후보는 [1], [3], [6], [1,4],[4,6],[1,4,6] 이고 [1,6]은 연속적이지 않기 때문에 subarray가 아니다. Example 1:

Input: nums = [1,2,1,2,3], k = 2 Output: 7 Explanation: Subarrays formed with exactly 2 different integers: [1,2], [2,1], [1,2], [2,3], [1,2,1], [2,1,2], [1,2,1,2]

JAVA

class Solution {
    public int subarraysWithKDistinct(int[] nums, int k) {
        int a = sub(nums,k);
        int b = sub(nums,k-1);

        return a-b;

    }

    public int sub(int[] nums, int k){
        HashMap<Integer, Integer> map = new HashMap();
        int left = 0, right = 0, ans = 0;

        while( right < nums.length){
            map.put(nums[right], map.getOrDefault(nums[right],0)+1);

            while(map.size() > k){
                map.put(nums[left],map.get(nums[left])-1);
                if(map.get(nums[left]) == 0){
                    map.remove(nums[left]);
                }
                left++;
            }

            ans += right - left +1;
            right ++;
        }
        return ans;
    }

}

Rust

use std::collections::HashMap;

impl Solution {
    pub fn subarrays_with_k_distinct(nums: Vec<i32>, k: i32) -> i32 {
        let a : i32 = Solution::sub(&nums, k);
        let b : i32 = Solution::sub(&nums, k-1);
        a-b
    }

    fn sub(nums: &Vec<i32>, k:i32) -> i32 {
        let mut map = HashMap::new();
        let mut ans : usize = 0;
        let mut right: usize = 0;
        let mut left: usize = 0;

        while right < nums.len(){
            *map.entry(&nums[right]).or_insert(0) += 1;

            while map.len() > k as usize{

                *map.get_mut(&nums[left]).unwrap()-=1;

                if map.get(&nums[left]).unwrap() == &0{
                    map.remove(&nums[left]);
                }
                left+=1;
            }

            ans += right - left + 1;
            right += 1;
        }
        ans as i32
    }
}

go - gin, fiber rust - actix,axum python - fastapi, django java - spring javascript - nestjs

https://web-frameworks-benchmark.netlify.app/compare?f=actix,axum,fiber,gin,django,fastapi,spring,nestjs-express 복잡한 어플리케이션은 아니고 print(hello world) 같이 간단한 요청이라고 알고 있다.

여러 지표가 있는데 Rust가 성능은 제일 좋은 것 같다. 물론 세상이 성능이 다는 아니고 장고로도 충분히 서비스도 제공하고 팀원들과 언어나 프레임워크 스펙 같은걸 통일해야 하는걸 감수해야 하지만 아주 흥미로운 결과인 것 같다. 클라우드 서비스에서는 사용하는 리소스가 다 돈이고 초기 투자 비용보다 사용료가 더 거대해질 수 있는 세상에서 go나 rust가 사용료 대비 성능을 뽑아낼 수 있다는 걸 감안하면 왜 개발자들이 익숙하고 많은 오픈소스가 있는 자바 대신 선택하는지 이해가 간다. C++은 빠르고 메모리도 절약하지만 어렵고 사고가 많이 나잖아.

주말에 심심해서 Golang이나 Rust 둘 중 하나 맛이나 봐야겠다 싶은 생각이 들었다. Rust가 GC도 없고 성능도 좋고 Golang보다 조금이라도 나은 것 같고 백엔드나 시스템 프로그래밍, Wasm이나 확장성 높은 것 같아서 Rust를 좀만 알아보자는 마음이 있었다.

생각보다 어렵다. 한국어 자료도 찾기 힘들고 생소한 개념도 많아서 좋은 러스트 코드를 작성하는게 어려울 것 같다. 객체의 소유권을 고민하면서 코딩을 한다거나 Box, Rc, Arc, Cow 등 진입장벽이 꽤 높다. 차라리 내가 자바가 아니라 C++이나 C를 좀 경험해봤다면 그나마 나을 것 같기도 하다.

공식 문서 읽기나 간단한 튜토리얼 정도는 경험했는데 러스트를 "잘"쓰는거 생각보다 쉽지 않을 것 같다. 대충 쓰면 버그 고치기도 힘들고 러스트 장점 살리지도 못 할 것 같다. 그래도 잘 하게 되면 뭐든 할 수 있다는 자신감이 생길 것도 같고....

원래 WebRTC연결을 WebSocket이 연결되면 바로 시작하도록 동기적으로 만들었다. 그런데 한 번 연결이 맺어지고 나서는 WebSocket을 거의 사용하지 않는데 이 때 WebSocket은 데이터를 계속 주고 받지 않으면 연결 상태를 확인하고 다시 연결한다. 그 과정에서 Event 기반으로 만들지 않고 Socket 연결되면 바로 동기적으로 WebRTC연결을 맺도록 만든 코드가 버그를 일으켰다.

 client.onConnect = (frame) => { // client는 WebSocket 연결 Client임.
   pc.createOffer({
     offerToReceiveAudio:true,
     offerToReceiveVideo:true
   })
     .then((offer) => {
     console.log("sdp offer created") // sdp status
     pc.setLocalDescription(offer)
       .then((r) => {
       client.publish({
         destination: `/app/api/audience/${userId}/offer`,
         body: JSON.stringify({
           buskerId,
           audienceId: userId,
           offer,
         })
       })
       new RTCPeerConnectionIceEvent("onicecandidate")
     })
   })
     .catch((error) => {
     console.log(error)
   })

WebSocket이 1분 간격으로 다시 연결되는데 이 때마다 클라이언트가 다시 연결을 시도했다. 클라이언트는 연결을 SDPOffer를 만들어서 Signaling Server로 보내는데 Server는 Offer를 받아서 Ice Candidate까지 진행을 한다. 그런데 브라우저는 Offer만 보내고 Server의 메세지에 응답을 하지 않아서 서버와 Kurento에 data가 쌓이다가 죽어버렸다. Event 중심으로 코딩을 하지 않으면 이러한 일들이 발생할 수 있다. 다양한 노드와 정해진 프로토콜로 메세지를 주고 받으며 연결을 해야하기 때문에 되도록 EventListner를 사용하는 것이 좋다.

Kurento

아직 이유를 제대로 알지 못하고 있으나 어쩌다보니 해결해버린 문제임.

@SpringBootApplication
@EnableWebSocket
public class ChocolateApplication {

    @Bean
    public KurentoClient kurentoClient() {

        String kurentoValue;
        return KurentoClient.create();
    }

Coturn은 EC2에 있는데 개발할 때 Kurento는 로컬환경에서 Docker로 작동시켰다. KurentoClient()에 파라미터를 주지 않으면 로컬 Kurento를 사용한다. Coturn도 문제가 없고 Signaling과 Broswer, Coturn은 문제가 없는 것을 확인했다. 모든 Event에 log를 달고 wireshark로 coturn의 응답도 확인을 했는데 막상 영상이 Kurento로 전달이 안됐다. 모든 상태가 다 좋은 것 같은데 안되서 Kurento가 문제라는 것은 확인했는데 막상 그걸 어떻게 해야할지 몰라서 Kurento의 Logging을 최대로 만들고 모니터링을 하려다가 무슨 생각이었는지 모르겠는데 EC2에 올려놓은 Kurento로 수정하니까 WebRTC연결이 되어버렸다.

    @Bean
    public KurentoClient kurentoClient() {

        String kurentoValue;
        return KurentoClient.create("여기에 EC2 Kurento url:port를 입력해주세요."
        );
    }

이유는 잘 모르겠다. SSAFY의 port block때문이었을까? 나머지 개발에 쫒겨서 더 이상 디버깅은 어려웠고 이 상태로 나머지 기능을 빨리 개발했다.

Coturn setting

realm 설정 안했더니 401 UnAuthrizaion Error가 발생했다. 유의... realm이 뭔지 모르겠어서 안했더니 문제가 있었다. 나중에 Kurento doc보니까 설정값을 다 알려줬었다. 그 때 realm도 설정하라고 했다. 임의의 문자를 입력하면 된다.

PC.close, Stomp.release()

소켓이나 PC를 잘 정리하는 것의 중요성 처음 개발할 때 연결을 끊는 것을 생각하지 않고 연결하는 기능만 생각해서 개발을 했다. WebSocket이나 Http나 제대로 연결을 끊는 로직을 별로 짜지 않았다. 프레임워크에서 알아서 연결을 끊어버렸기 때문이다. 그런데 Kurento를 사용할때는 개발자가 연결을 끊어버리지 않으면 그 세션 데이터 상태가 계속 남게된다.

 @Getter
@Setter
public class Busking extends UserSession implements Closeable {
    private final Logger log = LoggerFactory.getLogger(Busking.class);
    private final ConcurrentHashMap<String, UserSession> buskingSession = new ConcurrentHashMap<>();
    private final KurentoClient kurentoClient;
    private final IceMessageSendService iceMessageSendService;
    private final String buskerEmail;
    private final String buskingTitle;
    private final String buskingReport;
    private final String buskingHashtag;
    private final String buskingInfo;
    private GeoLocation geoLocation;
    private int audienceCount = 0;
    private WebRtcEndpoint buskerWebRtcEndpoint;
    private MediaPipeline buskerPipeline;
    ... ( 생략 )
       String sdpAnswer = audienceWebRtcEndpoint.processOffer(sdpOffer.getOffer().getSdp());
        JsonObject sdpResponse = new JsonObject();
        sdpResponse.addProperty("id", "audienceSdpAnswer");
        sdpResponse.addProperty("response", "accepted");
        sdpResponse.addProperty("sdpAnswer", sdpAnswer);

        iceMessageSendService.audienceSendSdpAnswer(audienceId, sdpResponse);
        audienceWebRtcEndpoint.gatherCandidates();

        buskingSession.put(sdpOffer.getAudienceId(), audienceSession);

이렇게 HashMap에 세션 정보를 관리하는데 이 세션에서 WebRtcEndPoint를 저장한다. 이 때 세션을 remove하는 기능이 없다면 브라우저에서는 웹 페이지를 꺼버리거나 비정상적으로 연결을 끊었을 때 세션 정보가 남아있어서 서비스를 사용할 때 충돌이 발생하거나 정보가 입력되지 않는 등 여러가지 에러가 발생했다. 뿐만 아니라 한번 방송에 들어가서 WebRTC 연결이 맺어지면 해제가 되지 않아서 다른 방송방에 들어가도 볼 수 없는 등 예측하기 어려운 문제들이 발생한다. 그래서 네트워크 연결과 관련된, 정상적인 종료, 비정상적인 연결 끊김 등 다양한 경우를 고려해줘야한다. TCP에서 3-HandShake,4-HandShake로 연결 상태를 관리하는 과정을 생각해보면 1번의 데이터 req,res를 위해 7번의 데이터 송수신이 낭비인 것 같으나 좋은 품질의 서비스를 위해 필수적인 요소라는 것을 배울 수 있었다.

개발

• 네트워크 통신이라는게 어려운 여러 기능들의 결합이라는걸 느낄 수 있었다. 웹 서비스에서 TCP/IP 통신을 위해서 내가 보이지 않는 low-level의 코드와 구현이 많겠다는 것을 새삼 느낄 수 있었다.

• coturn 서버의 작동을 확인하기 위해서 wireshark로 패킷을 확인해보고 정상 작동하고 있다는 것을 확인했는데 학부 때 배운걸 잘 써먹었다는 생각도 들고 wireshark로 패킷 검사하는 성공적인 경험 자체가 아주 좋다는 생각도 든다.

• 브라우저에서 정상적이지 않게 통신을 꺼버리거나 갑자기 네트워크가 끊기는 경우 등 네트워크에서 다양한 경우를 고려해야하는 것을 배웠다. 그런 경우에 재연결하거나 또는 연결 세션 상태를 지워버리는 등의 전략이 필요했다. 시간이 없어서 그 부분을 고려하지 못하고 개발한 상태에서 사람들에게 서비스를 공개하게 되어서 매우 아쉽다.

• 연결 세션을 redis로 저장하고 싶어서 세션 로그인하자고 했는데 시간이 없어서 redis로 하지 못해서 아쉽다....

• 개선 사항과 추가 기능 개발해야 할 것들이 많이 보인다. 아쉽게 흘려보내게 되는 것 같다.

개인

• 학부에서 WebRTC를 사용해서 Object Detecting을 구현하고 싶었는데 그때 제대로 못했다. 늘 아쉬움이 남았고 내가 못한 것에 대한 아쉬움이 남았었는데 이번에는 부족함이 많지만 충분히 해냈다고 생각이 들어서 만족스럽다. 이걸 하기 위해서 WebRTC의 프로토콜과 여러 네트워크 지식과 인프라 지식을 학습한 내가 자랑스럽다.

• 내가 쿠렌토를 개발한다고 팀원들의 개발 상황을 체크하지 못한게 좀 아쉽다. 팀 내 컨벤션이 제대로 지켜지지 못하고 있었는데 그걸 몰랐고 내 기능이 개발되지 않아서 FE 개발자가 기능을 산발적으로 만들고 있었다. 산발적인 UI나 Request를 나중에 통합할 때 고생이 있었다.

• Kurento를 이용한 스트리밍 기능의 개발이 예상보다 1주나 지체되었다.내 업무였는데 너무 눈치보이고 미안했다. 내가 Open vidu 안쓰고 Kurento Midea Server쓰기로한거였는데.... 예상하지 못한 문제들이 너무 많았다. 예상했다면 애초에 벌어지지 않게 막았겠지만. 재촉하지 않고 기다려준 팀원들에게 고맙다.

Kurento 1:N Streaming example Kurento Configure Kurento Install

1:N 스트리밍 설명

사람이 많이 출입하는 곳은 출구와 입구가 다릅니다. 문이 크든 작든 나가려는 사람과 들어오는 사람이 만나면 부딪치고 병목현상이 일어납니다. 네트워크도 이와 유사합니다. A->B로 가는 흐름이 있다면 흐름이 끝나고 나서 A<-B의 흐름을 보내거나 문을 2개 만들고 A->B와 A<-B의 흐름을 따로 관리합니다. KMS도 이 상태를 벗어날 수 없고 InBound와 OutBound를 처리하는 문이 다릅니다.이 문을 EndPoint라고 표현할 수 있을 것 같습니다.

KMS는 데이터를 보내는 EndPoint를 Src, 들어오는 EndPoint를 Sink라고 표현합니다. 문은 하나라도 통로는 여러개가 될 수 있죠. KMS는 한 Src에서 보내는 데이터를 여러 Sink로 보내야합니다. 1:N 스트리밍을 구현하는 것은 이러한 EndPoint와 통로(Pipeline)을 관리하고 개발자의 의도에 맞게 할당하는 것입니다.

아직 시그널링 서버에서 PeerConnection을 연결하지도 않았음에도 이것을 설명드리는 이유는 이러한 패턴이 낯설어서 Kurento example code를 읽을 때 곤혹스러웠기 때문입니다. 물론 아직도 익숙하지는 않지만 조금이나마 독자분께 도움이 되셨으면 좋겠습니다.

• 제 코드를 보면서 설명할까 했는데 프로젝트 구현 내용이 섞여 있어서 Peer Connection을 이해하기에는 오히려 복잡할 것 같아서 Kurento Example OneToMany Code를 보겠습니다.

중요한건 KurentoClient입니다. 이것을 통해서 Signaling Server와 KMS가 연결됩니다. 파라미터를 주지 않는다면 같은 서버에 있는 것을 디폴트로 실행됩니다. KMS와 Signaling이 다른 서버에 있다면

    @Bean
    public KurentoClient kurentoClient() {

        String kurentoValue;
        return KurentoClient.create("ws://someURL");
    }

이렇게 설정할 수 있습니다.

private synchronized void presenter(final WebSocketSession session, JsonObject jsonMessage)
      throws IOException {
    if (presenterUserSession == null) {
      presenterUserSession = new UserSession(session);

      pipeline = kurento.createMediaPipeline();
      presenterUserSession.setWebRtcEndpoint(new WebRtcEndpoint.Builder(pipeline).build());

      WebRtcEndpoint presenterWebRtc = presenterUserSession.getWebRtcEndpoint();

      presenterWebRtc.addIceCandidateFoundListener(new EventListener<IceCandidateFoundEvent>() {

        @Override
        public void onEvent(IceCandidateFoundEvent event) {
          JsonObject response = new JsonObject();
          response.addProperty("id", "iceCandidate");
          response.add("candidate", JsonUtils.toJsonObject(event.getCandidate()));
          try {
            synchronized (session) {
              session.sendMessage(new TextMessage(response.toString()));
            }
          } catch (IOException e) {
            log.debug(e.getMessage());
          }
        }
      });

      String sdpOffer = jsonMessage.getAsJsonPrimitive("sdpOffer").getAsString();
      String sdpAnswer = presenterWebRtc.processOffer(sdpOffer);

      JsonObject response = new JsonObject();
      response.addProperty("id", "presenterResponse");
      response.addProperty("response", "accepted");
      response.addProperty("sdpAnswer", sdpAnswer);

      synchronized (session) {
        presenterUserSession.sendMessage(response);
      }
      presenterWebRtc.gatherCandidates();

    } else {
      JsonObject response = new JsonObject();
      response.addProperty("id", "presenterResponse");
      response.addProperty("response", "rejected");
      response.addProperty("message",
          "Another user is currently acting as sender. Try again later ...");
      session.sendMessage(new TextMessage(response.toString()));
    }
  }

전체 코드는 Kurento OneToMany example에 있습니다.

  private synchronized void presenter(final WebSocketSession session, JsonObject jsonMessage)
      throws IOException {
    if (presenterUserSession == null) {
      presenterUserSession = new UserSession(session);

      pipeline = kurento.createMediaPipeline();
      presenterUserSession.setWebRtcEndpoint(new WebRtcEndpoint.Builder(pipeline).build());

      WebRtcEndpoint presenterWebRtc = presenterUserSession.getWebRtcEndpoint();

KMS를 사용하기 위해서는 EndPoint와 Pipeline이 필수적입니다. 데이터를 받고 보내는 문과 통로에 해당하기 때문입니다. 처음에 시작하면 KMS의 Pipeline을 만들고 그것을 활용하는 EndPoint를 생성합니다. 아직 Sink일지 Src일지 그 성격을 정해지지 않았습니다. 그저 문과 통로일뿐입니다.

String sdpOffer = jsonMessage.getAsJsonPrimitive("sdpOffer").getAsString();
String sdpAnswer = presenterWebRtc.processOffer(sdpOffer);

JsonObject response = new JsonObject();
response.addProperty("id", "presenterResponse");
response.addProperty("response", "accepted");
response.addProperty("sdpAnswer", sdpAnswer);

presenterUserSession.sendMessage(response);

presenterWebRtc.gatherCandidates();

상대 Peer가 보낸 SDPOffer를 받고 WebRtcEndPoint에 등록하면 SDP를 기반으로 EndPoint의 성격이 설정되고 이에 맞는 SDPAnswer가 생성됩니다. 이것을 상대 Peer에게 보냅니다. SDP교환 이후에는 Ice Candidate가 있어야겠죠 KMS와 상대 Peer가 통신하기 위한 Ice Candidate를 수집합니다.

들여쓰기가 잘 안되서 캡쳐했습니다. ㅠㅠ 여기도 EventListner로 작동하는데 Ice Candidate가 수집되면 상대 Peer에게 전송합니다.

• 다시 말씀드리지만 ICE Candidate를 수집하는 것까지 구현하는 내용이고 ICE 연결은 개발자가 구현하지 않아도 알아서 작동합니다.

ICE Candidate는 쌍방이 체크하는 것입니다. 따라서 KMS에게도 Coturn 값을 줘야겠죠. 제 프로젝트에서는 Broswer와 KMS의 Coturn이 같아서 굳이 써줘야 하나 싶기도 했는데 KMS 입장에서는 Stun/Turn 설정을 주지 않으면 ICE Candidate를 수집하지 못해서 ICE교환 과정이 원활하지 못한 것 같습니다. 일단 설정해줍니다. Kurento 공식문서 보시면 설정위치가 있습니다.

Kurento는 도커로 실행했습니다. 도커 배쉬에 들어가줍니다.

sudo docker exec -it e758f0f87f42 /bin/bash

vim /etc/kurento/modules/kurento/WebRtcEndpoint.conf.ini

ini 파일에 들어가서

stunServerAddress=stun.l.google.com
stunServerPort=19302
turnURL=name:credential@URL:port

stun/turn 설정 값을 주면 KMS가 상대 Peer와 알아서 ICE Candidate를 수집합니다. turnURL은 Coturn에서 설정한 값을 주면 됩니다.

Coturn도 설치하고 설정 값을 주셔야합니다. 관련 블로그가 많으니 참고하시면 될 것 같습니다만 유의할 점이 있습니다. realm이 무엇인지 이해하지 못해서 값을 주지 않았는데 401 unauthorize Error가 계속 발생했습니다. realm을 설정해주셔야합니다. 추가적으로 Coturn 모니터링 팁을 드릴 수는 있을 것 같습니다.

sudo service coturn status

로 coturn에 들어오는 값들과 coturn이 어떻게 반응했는지 확인할 수 있습니다.

coturn.service - coTURN STUN/TURN Server
     Loaded: loaded (/lib/systemd/system/coturn.service; enabled; vendor preset: enabled)
     Active: active (running) since Mon 2024-02-05 03:05:45 KST; 5 days ago
       Docs: man:coturn(1)
             man:turnadmin(1)
             man:turnserver(1)
   Main PID: 292618 (turnserver)
      Tasks: 9 (limit: 19165)
     Memory: 6.8M
     CGroup: /system.slice/coturn.service
             └─292618 /usr/bin/turnserver --daemon -c /etc/turnserver.conf --pidfile /run/turnserver/turnserver.pid

Feb 10 09:01:41 ip-xxx-xx-xx-xx turnserver[292618]: 453357: session 003000000000000210: closed (2nd stage), user <> realm <cat>
Feb 10 09:11:24 ip-xxx-xx-xx-xx turnserver[292618]: 453940: IPv4. tcp or tls connected to: 162.216.149.101:61724
Feb 10 09:11:34 ip-xxx-xx-xx-xx turnserver[292618]: 453950: IPv4. tcp or tls connected to: 162.216.149.101:59722
Feb 10 09:11:45 ip-xxx-xx-xx-xx turnserver[292618]: 453961: session 000000000000000206: TCP socket closed remotely 162.216.149>
Feb 10 09:11:45 ip-xxx-xx-xx-xx turnserver[292618]: 453961: session 000000000000000206: usage: realm=<catchup>, username=<>, r>
Feb 10 09:11:45 ip-xxx-xx-xx-xx turnserver[292618]: 453961: session 000000000000000206: closed (2nd stage), user <> realm <cat>
Feb 10 11:38:57 ip-xxx-xx-xx-xx turnserver[292618]: 462793: handle_udp_packet: New UDP endpoint: local addr 172.26.11.74:3478,>
Feb 10 11:38:57 ip-xxx-xx-xx-xx turnserver[292618]: 462793: session 003000000000000211: realm <catchup> user <>: incoming pack>
Feb 10 11:39:57 ip-xxx-xx-xx-xx turnserver[292618]: 462853: session 003000000000000211: usage: realm=<catchup>, username=<>, r>
Feb 10 11:39:57 ip-xxx-xx-xx-xx turnserver[292618]: 462853: session 003000000000000211: closed (2nd stage), user <> realm <cat

* 싸피 보안 정책으로 IP 지웠습니다!*

Trickle ICE로 Stun/Turn을 테스트해봤을 때 내가 생각하는 반응이 나오지 않는다면 위의 명령어로 모니터링 해볼 수 있습니다.

* 싸피 보안 정책으로 IP와 URL 지웠습니다!*

하단에 701 Error로 당황하시더라도 그건 무시하셔도 괜찮습니다. 위에 srflx,relay가 잘 나온다면 괜찮습니다. srflx는 stun,relay는 turn 서버의 응답 타입이기 때문입니다. 결과 값이 안나오고 Error가 발생한다면 그건 문제가 맞습니다. wireshark로 보내는 패킷, 응답 패킷 확인하거나 coturn status나 다른 서버들의 log를 확인해야 합니다.

sion 003000000000000212: realm <catchup> user <>: incoming packet BINDING processed, success
sion 003000000000000212: realm <catchup> user <>: incoming packet message processed, error 401: Unauthorized
4. Local relay addr: 172.26.11.74:52825
sion 003000000000000212: new, realm=<catchup>, username=<username1>, lifetime=600
sion 003000000000000212: realm <catchup> user <username1>: incoming packet ALLOCATE processed, success
sion 003000000000000212: refreshed, realm=<catchup>, username=<username1>, lifetime=0
sion 003000000000000212: realm <catchup> user <username1>: incoming packet REFRESH processed, success
sion 003000000000000212: usage: realm=<catchup>, username=<username1>, rp=4, rb=248, sp=4, sb=416

앞부분을 짤라서 복사하긴 했는데 Trickle ICE와 coturn status를 봤을 때 turn 서버에 잘 접속하고 사용하고 있습니다. username이나 password 값을 안주면 error 401 Unauthroized가 발생합니다. 또는 Coturn 설정의 문제일 수도 있습니다. WireShark를 사용하면 Stun,Turn 응답값을 패킷마다 확인할 수도 있습니다. 저는 realm 때문에 Coturn이 말을 안들어서 너무 답답해 WireShark로 모니터링까지 해보긴 했습니다만 그럴 필요까지는 없을 것 같습니다.

소회

어중간하게 WebRTC 프로토콜과 Kurento를 이해해서 더 많은 내용을 읽지 않아서 고생한 것 같습니다. 디버깅하는 방법을 몰라서 이것저것 시도하며 날린 시간이 참 아쉽습니다. 문제가 있을 때 공식 문서를 더 찾아보거나 많은 내용을 애초에 더 읽어보았다면 이해하기 위해서 다른 블로그를 더 찾아보았던 시간을 단축할 수 있었을 것 같습니다.

WebRTC connectivity Lifetime of WebRTC Sessioin Signaling and Video Call Signaling Connecting

위의 3가지는 MDN의 설명이고 아래 2개는 각 과정을 자세하게 설명한 영어 문서입니다. 저는 JS의 구현을 MDN의 설명을 주로 참고하였습니다.

RTCPeerConnection

RTCPeerConnectioin

브라우저로 WebRTC를 사용하기 위해서 사용하는 API입니다. Peer를 연결하고 유지하고 관리할 수 있습니다. EventTarget 인터페이스를 상속받고 있어 Event 위주로 구현합니다. 지속적으로 Peer 통신을 하고 있는 상태에서 네트워크 상태에 따라 연결 상태가 변할 수 있습니다. 이때 Event가 발생하도록 API가 설계되어 있어서 EventListner 중심으로 코딩하는게 좋은 것 같습니다.

연결 주의 사항

• KMS를 사용하는 Spring Signaling Server와 Broswer가 PeerConnection을 맺는 순서입니다. 모든 경우에 통용되지 않습니다.
• MDN의 WebRTC Connectivity를 읽으시면 어떻게 연결되어야 하는지 알 수 있습니다.
• STOMP 연결된 상황에서 PeerConnection을 진행합니다.

1. local peer는 RTCPeerConnection 객체를 생성합니다.

const Streaming = ({ isStreaming }) => {
    const pcRef = useRef(new RTCPeerConnection(PCConfig));
    const pc = pcRef.current;

이 때 파라미터를 줘야하는데 이는 Stun/Turn 서버 객체를 줍니다.

export const PCConfig = {
    iceServers:[
        {
            urls:"stun:stun.l.google.com:19302"
        },
        {
            urls:"turn:URL:PORT",
            username:"name",
            credential:"password"
        },
    ]
}

이 정보를 추후 IceCandidate에서 사용합니다. 미리 만들어 놓은 Coturn의 정보를 입력하시면 됩니다.

2. 브라우저 Camera,Audio 등록하기

navigator.mediaDevices.getUserMedia({video:true,audio:false})
  .then((stream) => {
  for (const track of stream.getTracks()){
    pc.addTrack(track,stream)
  }
  console.log("buskerId : "+ userId)
  videoElement.srcObject = stream
}).catch(error => {
  if (error.name === "OverconstrainedError") {
    console.error(
      `The resolution ${video.width.exact}x${video.height.exact} px is not supported by your device.`,
    );
  } else if (error.name === "NotAllowedError") {
    console.error("You need to grant this page permission to access your camera and microphone.",);
  } else {
    console.error(`getUserMedia error: ${error.name}`, error);
  }
})

3. SDP 교환

1. 처음 Connection을 맺을 때 SDP를 교환해야합니다. local은 createOffer()를 통해 자신의 세션 정보를 생성합니다. 이 때 addTrack이나 DataChannel을 미리 생성해두고 createOffer()를 하는게 정확한 세션 정보를 포함하는 SDP를 만들 수 있습니다.

2. 생성된 SDP를 setLocalDescription을 통해 등록합니다.

3. STOMP를 통해서 Signaling Server에게 자신의 sdpOffer를 전달합니다.

4. Server는 Offer를 받고 SDPAnswer를 생성 후 Peer에게 전송합니다.

5. Broswer는 Answer를 받고 자신의 RemoteDescription에 등록합니다.

   pc.addTrack을 통해서 우리가 어떤 정보를 보낼 것인지 세션 정보를 등록할 수 있습니다. 연결이 맺어지면 PC는 등록된 track을 stream으로 상대 peer에게 전달합니다.

   pc.onnegotiationneeded = (event) => {
   console.log(koreaTime+ " Negotiation을 진행합니다.")
   pc.createOffer({ // 1
   })
    .then((offer) => {
    console.log("sdp offer created") // sdp status
    pc.setLocalDescription(offer) // 2
      .then((r) => {
      client.publish({ // 3 
        destination: `/app/api/busker/${userId}/offer`,
        body: JSON.stringify({
          userId,
          offer,
        })
      })
    })
   })
    .catch((error) => {
    console.log(error)
   })
   }

   • onnegotiationneeded는 PeerConnection의 연결 협상이 필요하게 되면 호출되는 EventHandler입니다. Track에 미디어가 추가되거나 통신환경이 변경되어서 재협상을 해야할 때도 사용됩니다.

client.onConnect = (frame) => {
  console.log("streaming stomp : " + frame);
  // sdpOffer를 보내고 Answer를 받음
  client.subscribe(`/busker/${userId}/sdpAnswer`, (res) => {
    const offerResponse = JSON.parse(res.body);
    const answerId = offerResponse.id;
    const response = offerResponse.response;
    const sdpAnswer = offerResponse.sdpAnswer;

    // console.log("Received SDP Answer \n");
    console.log("Received SDP Answer \n"+offerResponse)
    pc.setRemoteDescription({
      type: "answer",
      sdp: sdpAnswer
    }).then(() => {
      console.log("Remote description set successfully");
    }).catch((error) => {
      console.error("Error setting remote description:", error);
    });
  });

STOMP에서 subcribe는 되도록 onConnect 내에서 등록하는게 안정적인 것 같습니다. 메세지를 보낼 때는 EventHandler로 필요한 상황에 보내도록 하고 메세지는 늘 받을 수 있는 상태로 관리하는 것입니다.

4. Ice Candidate

pc.onicecandidate = (event) => { //setLocalDescription이 불러옴.
            if (event.candidate) {
                // console.log("Client Send Ice Candidate : [ " + event.candidate.candidate + " ] ")
                // candidateList.push({iceCandidate: event.candidate})
                client.publish({
                    destination: `/app/api/busker/${userId}/iceCandidate`,
                    body: JSON.stringify({iceCandidate: event.candidate})
                });
            }
            if ( event.target.iceGatheringState === 'complete') {
                console.log('done gathering candidates - got iceGatheringState complete');
            }
        }

SDP교환과 ICE Candidate는 동기적이지 않습니다. SDP교환이 다 이루어지고 나서 ICE Candidate를 교환하지 않습니다. setLocalDescription()이 끝나면 Local측에서 ICE Candidate를 수집하고 ICE 과정을 진행합니다. onicecandidate는 setLocalDescription가 실행시키는 Event입니다. Broswer가 ice Candidate를 탐색하고 나면 서버에 전송합니다. 서버도 그것을 받으면 자신의 Ice Candidate를 탐색하고 Broswer에 전송합니다.

client.subscribe(`/busker/${userId}/iceCandidate`, (res) => {
  const iceResponse = JSON.parse(res.body);
  if (iceResponse.id === "iceCandidate") {
    console.log(koreaTime + " server send ice \n" + iceResponse.candidate.candidate)
    const icecandidate = new RTCIceCandidate(iceResponse.candidate)
    pc.addIceCandidate(icecandidate)
      .then()
  }
})

Broswer가 IceCandidate를 받고 나면 addIceCandidate에 등록합니다.

pc.oniceconnectionstatechange = (event) => {
    if (pc.iceConnectionState === 'new'){
      console.log(koreaTime +' 피어 연결을 시작 합니다. ')
    }
    console.log(koreaTime +' ICE 연결 상태:', pc.iceConnectionState);
    if (pc.iceConnectionState === 'connected') {
      console.log(pc.getStats().then(r=> console.log(koreaTime+" "+r)))
      console.log(koreaTime +' 피어 간 연결이 성공적으로 수립되었습니다.');
    } else if (pc.iceConnectionState === 'disconnected'){

      console.log(koreaTime +' 피어 간 연결이  끊어졌습니다.')
    } else if(pc.iceConnectionState === 'failed') {
      pc.restartIce()
      console.log(koreaTime +' 피어 간 연결이  실패.');
    }
};
pc.onconnectionstatechange = (event) => { // 데이터 연결 상태 확인
    console.log('데이터 연결 상태:', pc.connectionState);
    if (pc.connectionState === 'connected') {
      console.log(koreaTime +' 데이터 연결이 확립되었습니다.');
    } else if (pc.connectionState === 'disconnected') {
      console.log(koreaTime +' 데이터 연결이 끊어졌습니다.');
    }
};

이것으로 RTCPeerConnection의 중요한 연결이 끝났습니다. 연결 상태를 Event로 모니터링하면서 필요한 경우 대처하면 됩니다.

STOMP로 RTCPeerConnection을 구현하는 코드를 다 쓰고 나니 200줄이 조금 안되는군요. 물론 서버측 코드도 있습니다만... 이 적은 코드를 작성하기 위해서 공식문서와 갖은 블로그와 영어 문서를 2주 동안 매일 읽고 공부한 것 같습니다. 포스팅에 제가 했던 고민과 나름의 답을 나누고 싶었는데 너무 난잡해지는 것 같아서 다 쳐냈습니다. 제가 몇 줄 코드를 적고 뭐가 뭐라고 말씀 드리는 것보다 위의 문서들을 읽는게 WebRTC를 이해하시는데 훨씬 나은 것 같다고 생각했기 때문입니다.

그래도 참고해보시라고 RTCPeerConnection을 관리하는 컴포넌트의 전체 코드를 공개합니다.

import React, {useEffect, useRef, useState} from "react";
import CustomText from "../components/CustomText";
import {PCConfig} from "../WebRTC/RTCConfig";
import * as StompJS from "@stomp/stompjs";
import * as SockJS from "sockjs-client";
import {koreaTime} from "../WebRTC/PCEvent";
import {useRecoilState} from "recoil";
import {userInfoState} from "../RecoilState/userRecoilState";
import {useNavigate} from "react-router-dom";

// const userId = "buskerID"
let makingOffer = false


const Streaming = ({ isStreaming }) => {
    const [userInfo, setUserInfo] = useRecoilState(userInfoState);
    const pcRef = useRef(new RTCPeerConnection(PCConfig));
    const clientRef = useRef(
        new StompJS.Client({
            brokerURL: `${process.env.REACT_APP_API_WEBSOCKET_BASE_URL}`,
        })
    );
    const pc = pcRef.current;
    const client = clientRef.current;
    const userId = userInfo.userId
    const navigate = useNavigate();
    // Set Peer Connection
    useEffect(() => {
        if (isStreaming === false){
            pc.getSenders().forEach(sender => pc.removeTrack(sender))
            client.deactivate()
            pc.close()

            client.publish({
                destination: `app/api/busker/${userId}/stopBusking`
            })
            navigate("/")
        }
        const videoElement = document.getElementById("streamingVideo")
        pc.onicecandidate = (event) => { //setLocalDescription이 불러옴.
            if (event.candidate) {
                // console.log("Client Send Ice Candidate : [ " + event.candidate.candidate + " ] ")
                // candidateList.push({iceCandidate: event.candidate})
                client.publish({
                    destination: `/app/api/busker/${userId}/iceCandidate`,
                    body: JSON.stringify({iceCandidate: event.candidate})
                });
            }
            if ( event.target.iceGatheringState === 'complete') {
                console.log('done gathering candidates - got iceGatheringState complete');
            }
        }
        pc.oniceconnectionstatechange = (event) => {
            if (pc.iceConnectionState === 'new'){
                console.log(koreaTime +' 피어 연결을 시작 합니다. ')
            }
            console.log(koreaTime +' ICE 연결 상태:', pc.iceConnectionState);
            if (pc.iceConnectionState === 'connected') {
                console.log(pc.getStats().then(r=> console.log(koreaTime+" "+r)))
                console.log(koreaTime +' 피어 간 연결이 성공적으로 수립되었습니다.');
            } else if (pc.iceConnectionState === 'disconnected'){

                console.log(koreaTime +' 피어 간 연결이  끊어졌습니다.')
            } else if(pc.iceConnectionState === 'failed') {
                pc.restartIce()
                console.log(koreaTime +' 피어 간 연결이  실패.');
            }
        };
        pc.onconnectionstatechange = (event) => { // 데이터 연결 상태 확인
            console.log('데이터 연결 상태:', pc.connectionState);
            if (pc.connectionState === 'connected') {
                console.log(koreaTime +' 데이터 연결이 확립되었습니다.');
            } else if (pc.connectionState === 'disconnected') {
                console.log(koreaTime +' 데이터 연결이 끊어졌습니다.');
            }
        };
        pc.onnegotiationneeded = (event) => {
            console.log(koreaTime+ " Negotiation을 진행합니다.")
            pc.createOffer({
            })
                .then((offer) => {
                    console.log("sdp offer created") // sdp status
                    pc.setLocalDescription(offer)
                        .then((r) => {
                            client.publish({
                                destination: `/app/api/busker/${userId}/offer`,
                                body: JSON.stringify({
                                    userId,
                                    offer,
                                })
                            })
                        })
                })
                .catch((error) => {
                    console.log(error)
                })
        }

        const constraints = {video: false, audio: true}

         navigator.mediaDevices.getUserMedia(constraints)
            .then((stream) => {
                for (const track of stream.getTracks()){
                    pc.addTrack(track,stream)
                }
                console.log("buskerId : "+ userId)
                videoElement.srcObject = stream
            }).catch(error => {
                if (error.name === "OverconstrainedError") {
                    console.error(
                        `The resolution ${constraints.video.width.exact}x${constraints.video.height.exact} px is not supported by your device.`,
                    );
                } else if (error.name === "NotAllowedError") {
                    console.error("You need to grant this page permission to access your camera and microphone.",);
                } else {
                    console.error(`getUserMedia error: ${error.name}`, error);
                }
        })

        if (typeof WebSocket !== 'function') {
            client.webSocketFactory = function () {
                console.log("Stomp error sockjs is running");
                return new SockJS(`${process.env.REACT_APP_API_BASE_URL}/api`);
            };
        }

        client.onConnect = (frame) => {
            console.log("streaming stomp : " + frame);
            // sdpOffer를 보내고 Answer를 받음
            client.subscribe(`/busker/${userId}/sdpAnswer`, (res) => {
                const offerResponse = JSON.parse(res.body);
                const answerId = offerResponse.id;
                const response = offerResponse.response;
                const sdpAnswer = offerResponse.sdpAnswer;

                // console.log("Received SDP Answer \n");
                console.log("Received SDP Answer \n"+offerResponse)
                pc.setRemoteDescription({
                    type: "answer",
                    sdp: sdpAnswer
                }).then(() => {
                    console.log("Remote description set successfully");
                }).catch((error) => {
                    console.error("Error setting remote description:", error);
                });
            });
            client.subscribe(`/busker/${userId}/iceCandidate`, (res) => {
                const iceResponse = JSON.parse(res.body);
                if (iceResponse.id === "iceCandidate") {
                    console.log(koreaTime + " server send ice \n" + iceResponse.candidate.candidate)
                    const icecandidate = new RTCIceCandidate(iceResponse.candidate)
                    pc.addIceCandidate(icecandidate)
                        .then()
                }
            })
        }
        client.onStompError = (frame) => {
            console.log('Broker reported error: ' + frame.headers['message']);
            console.log('Additional details: ' + frame.body);
        };

        client.activate();

        return () => {
            // Cleanup function to be executed on component unmount
            console.log("Closing WebSocket connection and Peer Connection");
            client.deactivate(); // Close the WebSocket connection
            pc.close(); // Close the Peer Connection
        };

    }, [isStreaming]);
    return (
        <>
            <video id="streamingVideo" style={{width: '100%'}} autoPlay controls></video>
        </>
    )
}

export default Streaming;

KMS(Kurento Media Server)

Kurento

KMS는 미디어 서버를 오픈소스로 제공하는 Kurento의 미디어 서버입니다. Kurento는 Coturn을 이용한 Turn서버와 각종 기능을 제공하고 있으며 Nodejs, Java Spring을 이용한 시그널링 및 클라이언트 예시 코드를 제공하고 있어서 참고하여 개발하기 수월합니다. 다소 어려운 부분이 있으나 상용 프로그램을 구매하지 않고 공부하기 좋은 것 같습니다. KMS를 사용할 경우 Peer가 KMS와 브라우저로 이루어집니다. ICE를 하는 대상이 KMS와 브라우저가 되는 것입니다. 따라서 시그널링 서버에서는 KMS를 대신해서 상대 Peer와 SDP교환, ICE를 대신 처리해줍니다. 그리고 KMS의 연결 상태를 관리합니다.

KMS는 따로 설정하지 않고 Docker로 사용하지만 Coturn과 사용하거나 특수한 목적으로 사용해야할 때는 필요에 따라 내부의 환경 설정을 변경해야합니다.

Coturn(Open source Turn Server)

Coturn

Coturn은 Turn 서버의 표준을 구현한 오픈 소스 Turn 서버입니다. ICE를 위해서는 Stun과 Turn 둘 다 필요하다고 말했는데 Coturn은 Stun 기능을 제공하기 때문에 Stun 서버가 따로 필요하지 않습니다. Coturn이 둘 다 제공합니다. Turn 서버는 릴레이 서버이기 때문에 서버에 Inbound, Outbound가 발생합니다. 그래서 한 대의 Turn 서버에 많은 리소스가 생기면 지연이 심해질 수도 있고 네트워크 사용 비용이 많이 청구될 수 있습니다. Turn 서버를 사용하기 위해서는 Credential을 요구합니다. Credential을 구현하는 방법은 다양하지만 저는 lt-cred-mech을 사용했고 이 방식은 username:password을 Coturn 내부에 설정하고 Coturn에서는 이 username:password을 검증합니다.

Signaling Server (Java Spring Server)

Kurento는 NodeJS를 활용한 코드도 제공해주고 있지만 저는 Spring을 사용해서 시그널링 서버를 구현했습니다. 이후 포스팅에서 상세하게 설명하겠습니다.

Peer (broswer)

Peer는 브라우저나 모바일이 WebRTC API만 지원되면 가능합니다. 저는 브라우저로 개발했습니다. MDN에서 제공하는 WebRTC의 PeerConnection API를 활용해서 SDP를 만들고 Ice Candidate를 송수신합니다. 이후 ICE 연결이 완료되면 미디어 스트림을 관리합니다. 이후 포스팅에서 스트리밍 구현 내용 설명하겠습니다.

EC2(공인 IP 필요)

굳이 EC2일 필요는 없고 공인 IP를 할당받을 수 있는 서버면 됩니다. 본인 컴퓨터라도 포트 포워딩을 하거나 공인 IP을 구매후 할당하셔도 충분합니다. 다만 이 프로젝트에서는 EC2로 서버를 관리하고 있습니다.

SSAFY의 프로젝트로 WebRTC를 사용하는 서비스를 개발했습니다. WebRTC로 확장성있는 프로젝트를 위해서 Peer가 N개의 연결을 만드는 것이 아니라 Media Server를 사용하여 Midea Service를 제공했습니다. Media Server는 Kurento를 사용했습니다.(KMS) Signaling Server는 Spring boot로 개발했습니다.

기획

• 버스킹은 오프라인에서 일어난다.

• 음악 스트리밍은 온라인에서 일어난다.

• 버스킹과 음악 스트리밍의 간극을 줄일수는 없을까?

• 버스킹은 길에서 행인들에게 자신의 노래를 들려주려는 목적이 있다.

• 음악 스트리밍은 온라인으로 음악을 들려주고 시청자들과 소통하는데 목적이 있다.

• 본질은 관객들에게 음악을 들려주고 소통하기 위함이다.

• 온라인과 오프라인에서 오는 차이를 줄이고 온라인에서 오프라인으로 접근할 수 있도록 하자

• 버스킹은 특정 장소에서 짧지 않은 시간동안 음악을 한다는 특징이 있다. 장소를 옮기지 않는다.

• 그렇다면 버스킹 장소를 온라인에 공개하고 온라인에서 버스킹을 보다가 맘에 내키면 그 곳으로 이동할 수 있도록 하자.

• 즉 버스킹의 위치정보를 공개하여 온라인과 오프라인의 간격을 줄이고 버스킹과 스트리밍의 장점을 합쳐보자.

기능

SSAFY 발표회에서 시연보여줬던 내용들입니다!

0. 메인페이지

1. 방송하기

• 해당 기능이 발표회에서 제대로 작동하지 않아서 개인 화면에서 캡쳐했습니다.

  • 방송정보 입력(제목, 해쉬태그, 설명)

  • 위치 정보 공개 동의 (필수) 방송화면

2. 시청하기

   • 영상, 음성 지원

3. 채팅하기

   • 로그인했다면 ID, 로그인하지 않았다면 익명의 사용자

4. 위치 정보 기반으로 방송중인 스트리밍은 위치 지도에 표시하기

5. 스트리밍을 했었다면 Short 영상 등록하기

6. Short 영상에 댓글달기

7. 사용자 프로필 공개 (방송중이라면 프로필에서 길찾기 버튼으로 방송 찾아)

   • 현재 위치에서 방송중인 위치로 길찾기

아키텍처

EC2 안에 Coturn서버도 있습니다.

개발 구성

1. Java Spring boot(gradle) - Signaling and Application Server
2. React - Broswer(Peer)
3. Kurento - Media Server (Peer)
4. Coturn - (Stun/Turn Server)
5. AWS EC2
6. Docker

• 주의 : React를 잘 모르는 상태에서 팀이 프론트를 React를 쓰기로 해서 공부하면서 적용하느라 이상할 수 있음.

Signaling with WebSocket

Signaling

Signaling을 다시 말하자면 WebRTC를 위한 P2P(Peer to Peer) 연결 과정입니다. 이를 위해 Signaling Server와 Peer가 통신을 해야합니다. 이것은 클라이언트 - 서버 통신을 생각하시면 됩니다. Signaling을 통해서 P2P 연결이 이루어지고 나면 그 때는 Peer끼리 알아서 통신을 하게됩니다. 그 연결까지는 Peer가 Signaling Server를 통해서 상대 Peer와 SDP와 Ice Candidate를 교환하게 됩니다.

Signaling data 교환

일반적인 웹 서비스에서 클라이언트 - 서버는 HTTP Request, Response로 설계하게 됩니다. 요청이 있어야 응답이 있죠. 그런데 이 방법은 Signaling에 적용하기에는 문제가 있습니다. Peer와 Signaling Server가 순서없이 데이터를 주고 받기 때문입니다. 물론 SDP는 순서가 있지만 Ice Candidate과정에서는 누가 응답을 보내고 요청을 받는 순서가 정해지기 어렵고 잦은 데이터 교환 과정에서 3-handshake,4-handshake가 일종의 비용이 되기 때문입니다. Ice Candidate가 이루어지는 것을 추후 log로 찍어보시면 이해가 되실 겁니다. 그래서 시그널링을 WebSocket을 통해서 구현하게 됩니다. WebSocket으로 데이터를 주고 받고 각 피어는 EventHandler를 통해서 Signaling을 관리합니다.

WebSocket 여기서 웹 소켓이 무엇인지 한 번 살펴보세요!!

WebSocket - STOMP

자 그러면 WebSocket을 이용해서 통신을 구현해보기 전에 우리가 알아야할게 있습니다. WebSocket은 누구나 쓸 수 있지만 목적에 따라 편하게 쓰기 쉬운 라이브러리가 있습니다. 보통 백엔드가 NodeJS일 때는 SocketIO를 사용하고 Spring 환경에서는 STOMP를 사용합니다. 여기서는 STOMP를 사용합니다. STOMP까지 설명하는 것은 논지에서 벗어난 것 같아서 생략합니다. 사용할 때 겪었던 어려움만 말씀드리고 자세하게는 말씀드리지 않겠습니다. 공식문서를 찾아보세요.

STOMP

Stomp with spring

Java Config

 @Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer  {

    public void configureMessageBroker(MessageBrokerRegistry config) {
        config.enableSimpleBroker(new String[]{"/topic","/busker","/audience"}); // sub
        config.setApplicationDestinationPrefixes(new String[]{"/app"});
    }

    public void registerStompEndpoints(StompEndpointRegistry registry) {
        registry.addEndpoint(new String[]{"/api/chat"}).setAllowedOriginPatterns("*");
        registry.addEndpoint(new String[]{"/api/signal"})
                .setAllowedOriginPatterns("*");
    }

}

Java Stomp를 사용할 때 3가지를 유의해야 합니다.

1. config.enableSimpleBroker(new String[]{"/topic","/busker","/audience"}); // sub
2. config.setApplicationDestinationPrefixes(new String[]{"/app"}); // pub
3. registry.addEndpoint(new String[]{"/api/signal"})
               .setAllowedOriginPatterns("*");

1은 STOMP의 메세지를 구독하는 경우에 앞에 붙입니다. 2는 STOMP에 메세지를 보낼 때 사용합니다. 3은 STOMP WebSocket을 연결할 때 목적이나 클라이언트를 식별하기 위해 사용합니다.

JS connection

const Streaming = ({ isStreaming }) => {
    const pcRef = useRef(new RTCPeerConnection(PCConfig));
    const clientRef = useRef(
        new StompJS.Client({
            brokerURL: `wss://이건 비밀이에요./api/signal`,
        })
    );
    const pc = pcRef.current;
    const client = clientRef.current; //useRef를 사용합니다.

brokerURL에서 /api/signal을 붙였는데 EndPoint에 대응하는 Connection을 사용하시면 됩니다. 이 소켓은 Signaling을 위한 소켓입니다.

Subcribe

java

    public void buskerSendIceCandidate(String userId, HashMap<String, Object> iceCandidate){
        simpMessagingTemplate.convertAndSend(
                "/busker/" + userId + "/iceCandidate", iceCandidate
        );
    }

JS

client.subscribe(`/busker/${userId}/iceCandidate`, (res) => {
                const iceResponse = JSON.parse(res.body);
                if (iceResponse.id === "iceCandidate") {
                    console.log(koreaTime + " server send ice \n" + iceResponse.candidate.candidate)
                    const icecandidate = new RTCIceCandidate(iceResponse.candidate)
                    pc.addIceCandidate(icecandidate)
                        .then()
                }

1의 prefix busker로 구독 신청해놨습니다. 자바에서 보낼 때나 JS에서 받을 때 subcribe의 prefix를 유의해야합니다. prefix의 뒤로 특정 대상을 식별하지만 prefix가 없다면 애시당초 메세지를 읽을 수 없을겁니다.

JAVA Publish

@RestController
@RequiredArgsConstructor
public class SignalController {
    private final Logger log = LoggerFactory.getLogger(SignalController.class);
    private final SimpMessagingTemplate simpMessagingTemplate;
    private final BuskingManagingService buskingManagingService;

    @MessageMapping("/api/busker")
    public void listenTestStomp(@Payload String message) {
        HashMap<String, String> map = new HashMap<>();
        map.put("test", "test");

        simpMessagingTemplate.convertAndSend("/busker", map);
        return;
    }

JS publish

client.publish({
  destination: `/app/api/busker/${userId}/offer`,
  body: JSON.stringify({
       userId,
       offer,
       })
  })

JAVA에서 Controller가 JS가 보낸 메세지의 destination을 통해 매핑합니다.

주의

1. 동기들이 stomp를 적용할 때 블로그에 있는 많은 코드들을 참고했다가 어려움을 겪었습니다. latest 버전으로 업데이트 된 사용법과 블로그에 있는 내용이 다르거나 ChatGPT가 엉뚱하게 알려준 코드를 참고했을 문제를 겪었습니다. JAVA는 괜찮은데 Stomp-js를 쓰시는 경우는 유의하실 필요가 있는 것 같습니다. STOMP-js는 공식 문서를 참고하세요.
2. Sock-js를 적용하려다가 스프링에서 말썽을 부렸습니다. Sock-js는 웹소켓을 지원하지 않는 브라우저가 사용할 수 있도록 하기 위한 방법인데 IE8,9 정도의 예전 브라우저를 위한 방법입니다. 전 그냥 Sock-js 빼버렸습니다. Sock-js는 Pulling을 사용하기도 하고 IE8,9를 사용하시는 분이면 이 서비스를 거의 사용하지 않으실 것 같아서 그랬습니다. 문제 해결이 어렵더라고요.
3. wss는 HTTPS처럼 SSL이 추가된 소켓입니다. 서버에서 HTTPS를 적용하셨다면 딱히 하실 것은 없습니다.
4. 소켓 통신을 하신다면 ws를 http의 자리에 넣어주세요.

성공

WebSocket 연결이 끝났다면 축하합니다. WebRTC의 10분 능선 중 1분 능선을 건너셨습니다. 나머지 9분 능선 중 8이 이 Peer Connection입니다. (1은 종료와 마무리)

다음은 Peer Connection입니다.

내용이 너무 많아서 다음 포스팅으로 넘기겠습니다.

WebRTC(Web Real-Time Communication)는 웹 브라우저 간에 플러그인 없이 서로 통신할 수 있도록 설계된 API입니다. 또한 WebRTC는 표준임과 동시에 표준을 구현한 오픈소스 프로젝트입니다. WebRTC에서 브라우저가 일반적인 피어(Peer)인데 이들이 서버가 없이 통신하기 위한 여러 문제들을 해결하기 위한 요소들이 필요합니다. 이때 실시간으로 통신할 때 단순한 텍스트만 보내는 것이 아니라 영상, 음성 등 다양한 포멧의 데이터를 송,수신 할 수 있습니다. WebRTC는 빠른 데이터 송신, 수신을 위해서 RTP, RTCP, SDTP 등의 UDP를 활용하는 프로토콜을 활용하고 있습니다. 브라우저에서는 WebRTC를 위한 JavaScript API를 제공하고 있습니다. 개발자가 특별한 라이브러리를 사용하지 않는다면 표준화된 이 API를 사용하여 구현합니다.

WebRTC를 위한 개념

P2P(Peer To Peer)

피어는 일종의 노드이며 클라이언트라고 생각하시면 됩니다. 다만 기존의 웹 구조에서는 클라이언트 - 서버로 데이터 전달과 서비스 제공이 이루어졌다면 WebRTC에서는 쌍방이 데이터 송수신을 할 수 있기 때문에 피어(Peer)라고 지칭합니다. 웹 서비스는 네트워크 연결의 어려움이 없었습니다. 프레임워크나 브라우저 레벨에서 TCP/IP 서비스를 잘 만들어놓았고 기존 개발자들이 네트워크에 대해서 잘 몰라도 클라이언트- 서버의 네트워크 연결이 거의 자동으로 진행되었습니다. 그저 클라이언트가 서버의 IP:port로 요청(request)를 보내면 3-handshake, 4-handshake를 통한 통신,통신 종료가 개발자들이 신경안쓰는 사이에 진행되었습니다. 그러나 P2P(Peer to Peer)는 이러한 일종의 자동화된 네트워크 연결이 어렵습니다. 후술하겠지만 Peer는 일반적인 IP를 쓰지 않기 때문입니다. 따라서 P2P 서비스는 일반적인 클라이언트 - 서버와는 다른 연결과정과 인프라 가 필요합니다.

Protocol of WebRTC

1. Signailing Server

WebRTC는 P2P를 위한 서비스이지만 아이러니하게도 그 연결을 위해서는 P2P의 통신을 중계해줄 시그널링 서버가 필요합니다. WebRTC는 다양한 프로토콜을 통해서 피어 통신의 저지연성을 지원하고 있습니다. 이를 위해서 각 피어는 본인의 IP와 포트, 미디어 코덱 등의 통신 세션 정보를 교환해야합니다. Peer 통신을 하기 위해서는 나의 정보를 상대에게 보내야하는데 상대에게 내 정보를 보낼 방법이 없기 때문에 시그널링 서버라는 중계자가 필요합니다.(상대의 IP, port를 몰라서 보낼 수가 없습니다.) 따라서 시그널링 서버에게 내 정보를 보내면 시그널링 서버는 상대에게 나의 정보를 중계해주고 상대도 저에게 본인의 정보를 시그널링 서버를 통해서 보내게 됩니다. 시그널링 서버를 통해서 각자의 세션 정보를 교환하면 PeerConnection이 생깁니다. 이 PeerConnection을 통해서 필요한 데이터를 교환하게 됩니다.

PeerConnection

이 부분은 피어가 연결을 맺는 중요한 부분인데 저는 여기서 곤혹스럽기도 하고 머리아픈 많은 디버깅을 처리했던 부분이기도 합니다. 나중에 코드와 함께 설명할 수 있도록 하겠습니다. 우선 축약한 개념과 순서를 설명하겠습니다. 추후 이 두 웹사이트에서 읽어보시면 좋을 것 같습니다. WebRTC 정보 MDN WebRTC connectivity

1. 두 피어는 SDP(Session Description Protocol)를 교환하고 쌍방의 SDP를 저장합니다.
2. Ice Candidate를 교환하면서 최적의 네트워크 통신 경로를 탐색합니다.
3. ICE를 통해서 각자의 환경에 최적의 네트워크 통신을 확정합니다.
4. Negotiation이 끝나면 Peer Connection 생성되고 미디어를 송수신할 수 있습니다.

2. Connecting for NAT Traversal with Stun/Turn server

NAT Traversal

IP기반의 네트워크 통신에서는 IP가 디바이스마다 설정됩니다. 이 IP를 통해서 특정 피어를 식별하고 통신을 가능하게 합니다. 웹 서비스에서도 클라이언트는 DNS를 통해 서버의 IP를 알고 요청을 보낼 수 있습니다. 서버도 클라이언트의 IP를 식별해서 응답을 보낼 수 있습니다. P2P에서도 이와 같이 상대의 IP를 알아야 네트워크 통신을 할 수 있다는 점은 동일하지만 웹 서비스와 다른 점은 개인은 NAT(Network Address Tranlation)로 야기되는 문제가 있다는 것입니다. 이를 통해서 Private Ip/Public Ip가 생깁니다. 내 Public IP 알아보기 위의 링크를 통해서 나오는 IP가 본인의 Public IP 입니다. 윈도우라면 CMD에 ipconfig, 맥이나 리눅스라면 ifconfig를 치면 나오는 네트워크 ip가 private IP입니다. P2P 통신을 하기 위해서는 본인의 Private IP와 Public IP 둘 다 알고 상대에 전해줘야 합니다. 마치 본인 아파트 1층 비밀번호와 집 현관문 비밀번호를 알아야 집에 들어갈 수 있는 것과 같이 말입니다. NAT 알아보기

Stun/Turn server

Stun server

Stun server는 개인의 Private,Public IP를 반환해주는 서버입니다. Peer는 Stun를 통해서 개인의 Network 정보를 알 수 있습니다. Stun을 통해 반환되어 P2P에 사용할 때는 경우 Server Reflexive라고 합니다.

Turn server

Stun 서버를 통해서 본인의 정보를 알고 통신하는 것이 제일 좋습니다. 그러나 NAT의 보안 정책이 엄격하거나 기타 이유로 인해서 Stun으로는 통신이 어려운 경우 Turn 서버를 사용해야 합니다. Turn 서버는 공인IP를 가진 서버를 통해서 구현해야 합니다. Turn서버는 피어가 상대에게 직접 데이터를 보내지 못하기 때문에 그 데이터를 받아서 상대에게 전달해주는 릴레이 역할을 하는 서버입니다. 이 방법을 사용하면 직접 데이터를 보내는 것이 아니라 데이터를 릴레이하기 때문에 필연적으로 지연되고 엄밀히 말해 P2P통신이 아니기도 합니다. 따라서 Turn 서버를 사용하는 경우는 지양하게 됩니다.

ICE

SDP를 주고 받고 나서 각 피어는 ICE(Interactive Connectivity Establishment)를 실행합니다. ICE는 NAT를 우회하는 방법 Stun/Turn을 통해서 두 피어가 WebRTC 통신을 할 수 있도록 연결을 만드는 일종의 프레임워크입니다. Stun/Turn를 통해서 피어가 통신할 수 있는 후보군을 Ice Candidate라고 하는데 이 후보군들을 각 피어가 테스트해보면서 최적의 경로를 발견하면 이를 통해서 두 피어는 그 경로를 사용하기로 합의하면서 연결 상태를 유지합니다. 이 경로는 추후 데이터 연결 상태에 따라서 다시 수정될 수 있습니다. ICE는 개발자가 손대지 않고 표준에 따른 구현을 사용하며 개발자가 임의로 수정하지 않을 것을 권유합니다. 개발자는 ICE는 이해하지 못해도 Peer Connection 구현에는 문제가 없으며 단지 SDP 교환 이후에 Ice Candidate를 또한 교환하는 것까지 구현하기만 하면 됩니다.

3. Securing the transport layer with DTLS and SRTP

WebRTC는 UDP를 사용합니다. 아무리 그래도 우리... 보안은 신경써야겠죠? HTTPS와 같이 Secure Layer를 추가합니다. DTLS (Datagram Transport Layer Security) and SRTP (Secure Real-Time Transport Protocol)를 통해 기본적인 보안은 챙겨줍니다. 그러나 P2P에서 HTTPS와 같이 인증된 인증서를 쓰기는 어렵습니다. 따라서 시그널링 과정에서 사용한 HTTPS의 인증서를 재활용하여 DTLS를 사용하게 됩니다.

4. Communication with RTP,SCTP

이제 ICE를 통해서 두 피어가 연결되었습니다. 그러면 데이터를 교환해야겠죠? RTP (Real-time Transport Protocol), and SCTP (Stream Control Transmission Protocol)를 사용합니다. RTP는 실시간 스트리밍을 위해서 사용하며 SCTP는 안정적이고 순서가 보장된 메세지 전달을 위해서 사용합니다.

Media server (Option)

WebRTC는 P2P를 지원하기 위한 표준이지만 참여하는 사람들이 늘어날수록 각 피어의 네트워크 및 데이터 처리 리소스가 많이 사용되는 한계가 있습니다. 일반인들이 데스트탑으로 WebRTC 서비스를 사용하고 소수의 인원이 참여하는 것이 아니라 많은 인원이 이 서비스에 참여하려고 한다면 Media Server를 사용하는 것은 필수가 됩니다.

사실상 엄밀한 P2P가 아니지만 웹 환경에서 실시간 통신을 하기 위한 최선의 선택이라고 볼 수 있을 것 같습니다. 다만 필수는 아니고 이를 위한 피어 - Media server 시그널링 + Media Server 구현을 하기 위한 품이 더 들어갑니다.

참고

https://tech.kakaoenterprise.com/121 [카카오엔터프라이즈 기술블로그 Tech&(테크앤):티스토리] https://developer.mozilla.org/ko/docs/Web/API/WebRTC_API https://webrtcforthecurious.com/ https://hyperconnect.github.io/2022/12/30/introduction-to-media-server.html

바이너리로 인코딩하는 이유를 알면 왜 protobuf가 바이너리로 직렬화하는지 알 수 있을 것이다.

컴퓨터는 데이터를 어떻게 표현할까?

컴퓨터에서 모든 것은 bit와 byte다.(8bit == 1byte) 8bit를 가지고 만들 수 있는 숫자는 0~255 인데 이걸로 컴퓨터는 뭘 할 수 있는데? 바로 문자를 표현할 수 있다. 이게 아스키 코드다. 예를 들어 'A'를 표현하고 싶다면 A에 해당하는 아스키 숫자 65를 바이너리로 저장한다. 01000001. 아스키코드

그래서 뭐

그래 뭐 알겠다. 우리가 보는 65와 컴퓨터의 65는 뭔 차이인데? 아주 큰 차이가 있다. 우리는 65를 그대로 컴퓨터에 저장했다고 생각한다. 그런데 우리는 두 가지 숫자를 입력한 것이다. '6' , '5' 이 때 아스키 코드는 두개가 사용된다. 그래서 2byte가 사용됐다. 컴퓨터 입장에서 65는 그냥 01000001로 저장할 수 있는데 사람이 입력한 것은 6,5로 나눠서 저장하게 되는 것이다.

더 나아가서 우리가 숫자 4,000,000,000을 저장한다고 치자. 사람은 10개의 아스키 코드를 사용하게 된다. 컴퓨터는? 4byte만 사용하면 된다.

위와 같이 숫자를 컴퓨터 저장 공간에 저장하는 것은 공간이 효율적이다.

근데 왜 안 써?

바이너리 포멧은 매우 효과적이지만 우리는 왜 항상 쓰지 않는가?

1. 읽기가 넘 힘들다.

사람이 4byte 숫자를 읽을 때 (1,0이 32개..!) 이게 아스키로 저장된 4byte문자인지 숫자인지 알 방도가 없다. 그런데 10개의 아스키 코드로 4000000000을 보면 숫자라는걸 알 수 있다.

2. 편집하기가 어렵다.

4000000000을 2000000000으로 변경하고 싶다면 이에 해당하는 이진수 표현법을 생각해야 한다. 그런데 아스키 코드에서는? 4를 2로 변경하면 된다.

3. 생각보다 효율적이지 않다.

숫자를 2진수로 표현하면 이상적으로는 3의 계수를 절약할 수 있습니다(4바이트 숫자는 10바이트의 텍스트를 나타낼 수 있음). 그러나 이는 표현하고자 하는 숫자가 크다는 가정 하에 이루어진 것입니다(999와 같은 3자리 숫자는 4바이트 숫자보다 ASCII로 표현하는 것이 더 낫습니다). 마지막으로, ASCII는 실제로 바이트당 7비트만 사용하므로 이론적으로 ASCII를 함께 패킹하면 1/8 또는 12%의 이득을 얻을 수 있습니다. 하지만 이런 방식으로 텍스트를 저장하는 것은 일반적으로 번거로울 수 있습니다. 출처

텍스트 압축하면 효율적임.

바이너리 파일이 효율적인 이유는 1바이트의 8비트를 모두 사용할 수 있지만 대부분의 텍스트는 패턴이 고정되서 사용하지 않는 공간이 남기 때문입니다. 그러나 텍스트 데이터를 압축하면 공간을 줄이고 효율적으로 만들 수 있습니다.

언제 바이터리 파일 포멧을 사용하는게 유용할까?

여튼 우리가 항상 효율적이란 이유로 바이너리 포멧을 사용하지는 않는다. 효율은 저장공간, 코딩하는 노력, 시간 등등이 포함되기 때문이다. 이를 다 포함해서 바이너리를 사용할 때 이점을 가지는 것이 무엇일까?

PNG는 바이너리 포멧을 사용하는데 작은 이미지 파일을 만들 때 데이터 효율성이 중요하기 때문이다.

비즈니스를 위해서 사용하기도 한다. 바이너리를 읽고 리버스 엔지니어링을 하기 어렵기 때문이다.

데이터 통신할 때, 통신 데이터가 많다는 것은 더 많은 리소스가 사용된다는 뜻이다. 데이터를 압축하기도 하고 직렬화하고 역직렬화 하는 경우가 많다. 이럴 때 사용한다. 특히 서버 클라이언트, 블록체인 네트워크 상에서는 더욱 중요하다. 데이터 무결성과 보안, 효율성을 확보하기 위해서 바이너리로 포멧을 바꾸고 직렬화하는 경우가 필요하다.

결론

바이너리는 효율적이다. 사람이 읽기는 어렵지만 컴퓨터에게는 큰 문제가 없고 오히려 좋다. 코딩할때는 직렬화하지 않고 송신할 때 직렬화하고 수신할 때 역직렬화하면 컴퓨터도 좋고 나도 좋고 네트워크도 좋아한다.