video streaming을 위한 webflux 적용시 오류

[ydj515]

issue 사항

현재 세팅되어있는 webmvc에서 mongodb에 저장된 video를 streaming형태로 client에 전달해주길 기대하였으나 webflux 설정에 오류 발생.

현 상황

현재 mongodb에서 가져온 stream의 응답이 아래와 같이 application/json형태로 응답을 내려줌.
응답이 json형태여서 blob으로 video가 만들어지지않음.

[
    {
        "nativeBuffer": "AAAAGGZ0eXBtcDQyAAA..."
    },
    {
        "nativeBuffer": "AAAAGGZ0eXBtcDQyAAA..."
    }
]

그래서 react client에서 video태그에서 재생이 안되는 현상 발생

• fetchVideo.js

const fetchVideo = async () => {
  try {
    const response = await axiosClient.get(
      `/videos/search/�검색키워드`,
      {
        responseType: "blob",
        headers: {
          Range: "bytes=0-" // Range 요청 추가
        }
      }
    );
    debugger; // TODO: response 확인
    const mimeType = response.headers["Content-type"] || "video/mp4";
    const blob = new Blob([response.data], { type: "video/mp4" });
    const url = URL.createObjectURL(blob);
    setVideoUrl(url);
  } catch (error) {
    console.log("Error fetching video: " + error);
  } finally {
    setLoading(false);
  }
};

• video.html

<div>
  {loading ? (
    <div>Loading...</div>
  ) : (
    <video src={videoUrl} controls autoPlay width="600" />
  )}
</div>

attempt to

1. application.yml을 설정하여 sprintboot의 web application type을 강제로 reactive로 지정.

spring:
  main:
    web-application-type: reactive

=> tomcat을 사용하지않고 netty를 강제로 설정해주면 mvc 코드가 동작하지 않음.

2. ResponseEntity를 Mono로 return

• VideoController

@RestController
@RequestMapping("/videos")
class VideoController(
   private val videoService: VideoService,
) {
   @GetMapping("/search/{title}")
   fun streamVideo(
       @PathVariable title: String,
       request: HttpServletRequest,
   ): Mono<ResponseEntity<Flux<DataBuffer>>> =
       videoService
           .findVideoByTitleLike(title)
           .flatMap { gridFSFile ->
               val contentType =
                   gridFSFile.metadata?.getString("contentType")
                       ?: MediaType.APPLICATION_OCTET_STREAM_VALUE
               val encodedFilename =
                   URLEncoder
                       .encode(
                           gridFSFile.filename,
                           "UTF-8",
                       ).replace("+", "%20")

               // Range 헤더를 파싱하여 시작과 끝 바이트를 결정합니다.
               val rangeHeader = request.getHeader(HttpHeaders.RANGE)
               val range = parseRange(rangeHeader, gridFSFile)

               val responseEntity =
                   ResponseEntity
                       .ok()
                       .header(HttpHeaders.CONTENT_TYPE, contentType)
                       .header(
                           HttpHeaders.CONTENT_DISPOSITION,
                           "inline; filename*=UTF-8''$encodedFilename",
                       ).header(HttpHeaders.ACCEPT_RANGES, "bytes")

               // 스트리밍을 시작합니다.
               val videoStream = videoService.streamVideo(gridFSFile, range)
               Mono.just(responseEntity.body(videoStream))
           }

   @GetMapping("/test-stream", produces = [MediaType.APPLICATION_OCTET_STREAM_VALUE])
   fun testStream(): Flux<DataBuffer> {
       val content = "Test content".toByteArray()
       return Flux.just(DefaultDataBufferFactory().wrap(ByteBuffer.wrap(content)))
   }

   private fun parseRange(
       rangeHeader: String?,
       gridFSFile: GridFSFile,
   ): LongRange =
       if (!rangeHeader.isNullOrBlank()) {
           val ranges = rangeHeader.replace("bytes=", "").split("-")
           val start = ranges[0].toLongOrNull() ?: 0L
           val end = ranges.getOrNull(1)?.toLongOrNull() ?: (gridFSFile.length - 1)
           start..end
       } else {
           0L..(gridFSFile.length - 1) // Range 헤더가 없으면 전체 파일 범위를 반환
       }
}

• VideoService

@Service
class VideoService(
    private val gridFsTemplate: GridFsTemplate,
) {
    fun findVideoByTitleLike(title: String): Mono<GridFSFile> {
        val query =
            Query.query(
                Criteria
                    .where("metadata.title")
                    .regex(".*$title.*", "i"),
            )
        return Mono
            .fromCallable { gridFsTemplate.findOne(query) }
            .subscribeOn(Schedulers.boundedElastic())
    }

    fun findVideoByVideoId(videoId: String): Mono<GridFSFile> {
        val query =
            Query.query(
                Criteria
                    .where("video_id")
                    .regex(".*$videoId.*", "i"),
            )
        return Mono
            .fromCallable { gridFsTemplate.findOne(query) }
            .subscribeOn(Schedulers.boundedElastic())
    }

    fun findVideoByTagsLike(tag: String): Mono<GridFSFile> {
        val query =
            Query.query(
                Criteria
                    .where("metadata.tags")
                    .regex(".*$tag.*", "i"),
            )
        return Mono
            .fromCallable { gridFsTemplate.findOne(query) }
            .subscribeOn(Schedulers.boundedElastic())
    }

    fun streamVideo(
        gridFSFile: GridFSFile,
        range: LongRange,
    ): Flux<DataBuffer> {
        val inputStream: InputStream = gridFsTemplate.getResource(gridFSFile).inputStream
        val bufferFactory = DefaultDataBufferFactory()

        // 범위에 해당하는 바이트를 읽도록 스트리밍 함수를 수정합니다.
        inputStream.skip(range.start)

        // 범위의 크기
        val rangeSize = range.endInclusive - range.start + 1
        var bytesReadTotal = 0L // 전체 읽은 바이트 수 추적

        return Flux
            .create { sink ->
                try {
                    val buffer = ByteArray(4096)
                    var bytesRead = inputStream.read(buffer)

                    while (bytesRead != -1 && bytesReadTotal < rangeSize) {
                        bytesReadTotal += bytesRead

                        // 범위를 초과하지 않도록 처리
                        val remainingBytes = rangeSize - bytesReadTotal
                        val bytesToSend =
                            if (remainingBytes < 4096) {
                                remainingBytes.toInt()
                            } else {
                                bytesRead
                            }

                        // 명시적으로 wrap 메서드에서 DataBuffer를 생성
                        val dataBuffer: DataBuffer =
                            bufferFactory.wrap(
                                ByteBuffer.wrap(buffer, 0, bytesToSend),
                            )
                        sink.next(dataBuffer)

                        if (bytesReadTotal >= rangeSize) {
                            sink.complete()
                        }

                        bytesRead = inputStream.read(buffer)
                    }

                    sink.complete()
                } catch (e: IOException) {
                    sink.error(e) // 에러 발생 시 스트림 종료
                }
            }.doFinally { inputStream.close() }
    }
}

=> 아래와 같은 오류 발생.

2024-09-22T01:56:22.757  WARN  24167 --- [catch-weak] [http-nio-8080-exec-4] o.s.w.s.m.s.DefaultHandlerExceptionResolver - Resolved [org.springframework.http.converter.HttpMessageNotWritableException: No converter for [class org.springframework.web.reactive.function.server.DefaultServerResponseBuilder$BodyInserterResponse] with preset Content-Type 'null']
2024-09-22T01:56:32.849  WARN  24167 --- [catch-weak] [http-nio-8080-exec-4] o.s.w.s.m.m.a.ExceptionHandlerExceptionResolver - Resolved [org.springframework.web.HttpMediaTypeNotAcceptableException: No acceptable representation]

3. ServerResponse를 Mono를 return
   서비스 코드는 2번과 동일하게 구성

• VideoController

@RestController
@RequestMapping("/videos")
class VideoController(
    private val videoService: VideoService,
) {
    @GetMapping("/search/{title}", produces = [MediaType.APPLICATION_OCTET_STREAM_VALUE])
    fun streamVideo(
        @PathVariable title: String,
        request: HttpServletRequest,
    ): Mono<ServerResponse> =
        videoService.findVideoByTitleLike(title).flatMap { gridFSFile ->
            val contentType =
                gridFSFile.metadata?.getString("contentType")
                    ?: MediaType.APPLICATION_OCTET_STREAM_VALUE
            val encodedFilename =
                URLEncoder
                    .encode(
                        gridFSFile.filename,
                        "UTF-8",
                    ).replace("+", "%20")

            // Range 헤더를 파싱하여 시작과 끝 바이트를 결정합니다.
            val rangeHeader = request.getHeader(HttpHeaders.RANGE)
            val range = parseRange(rangeHeader, gridFSFile)

            // 스트리밍을 시작합니다.
            val videoStream = videoService.streamVideo(gridFSFile, range)

            // ServerResponse를 생성하여 반환
            ServerResponse
                .ok()
                .contentType(MediaType.parseMediaType("application/octet-stream"))
                .header(
                    HttpHeaders.CONTENT_DISPOSITION,
                    "inline; filename*=UTF-8''$encodedFilename",
                ).header(HttpHeaders.ACCEPT_RANGES, "bytes")
                .body(BodyInserters.fromPublisher(videoStream, DataBuffer::class.java))
        }

    private fun parseRange(
        rangeHeader: String?,
        gridFSFile: GridFSFile,
    ): LongRange =
        if (!rangeHeader.isNullOrBlank()) {
            val ranges = rangeHeader.replace("bytes=", "").split("-")
            val start = ranges[0].toLongOrNull() ?: 0L
            val end = ranges.getOrNull(1)?.toLongOrNull() ?: (gridFSFile.length - 1)
            start..end
        } else {
            0L..(gridFSFile.length - 1) // Range 헤더가 없으면 전체 파일 범위를 반환
        }
}

=> 아래와 같은 오류 발생.

2024-09-22T01:56:22.757  WARN  24167 --- [catch-weak] [http-nio-8080-exec-4] o.s.w.s.m.s.DefaultHandlerExceptionResolver - Resolved [org.springframework.http.converter.HttpMessageNotWritableException: No converter for [class org.springframework.web.reactive.function.server.DefaultServerResponseBuilder$BodyInserterResponse] with preset Content-Type 'null']
2024-09-22T01:56:32.849  WARN  24167 --- [catch-weak] [http-nio-8080-exec-4] o.s.w.s.m.m.a.ExceptionHandlerExceptionResolver - Resolved [org.springframework.web.HttpMediaTypeNotAcceptableException: No acceptable representation]

2,3 번 에러

2,3번 에러의 경우 springboot쪽 media type이 지정된게 없다고 나온다. mvc 설정과 webflux 설정이 충돌나기에 mediatype이 null이라고 나옴.
그러나, log level이 error가 아닌 warn이기에 놓치기 쉬움을 유의.