자바스크립트에서 데이터 스트림 읽기 (ReadableStream)

ChatGPT와 같은 LLM(Large Language Model, 대형 언어 모델)이 등장하면서 웹에서 텍스트를 스트리밍하는 사례가 점점 늘어나고 있는데요. 그에 따라 별다른 라이브러리없이도 웹에서 스트림을 쓰고 읽을 수 있는 자바스크립트의 Streams API가 다시 주목받고 있는 것 같습니다.

이번 포스팅에서는 자바스크립트의 Streams API을 사용하여 스트림을 생성하고 데이터를 읽는 방법에 대해서 알아보겠습니다.

읽을 수 있는 스트림 생성

자바스크립트에서 데이터를 읽을 수 있는 스트림을 만들 때는 ReadableStream 클래스를 사용합니다.

이 클래스의 생성자는 두 개의 인자를 받는데요 두 번째 인자인 queuingStrategy는 고급 주제이기 때문에 생략하고, 첫 번째 인자인 underlyingSource 대해서만 다루겠습니다.

underlyingSource는 쉽게 말해 스트림의 데이터 원천지 또는 공급처를 나타내는 객체인데요. 이 객체의 start() 함수를 사용하여 스트림이 내부적으로 관리하는 큐(queue)에 데이터를 넣어줄 수 있습니다.

간단한 예로, 1초 간격으로 0부터 10까지의 숫자 데이터가 공급되는 스트림을 만들어볼까요?

const stream = new ReadableStream({
  start(controller) {
    console.log("start");
    let num = 0;
    const interval = setInterval(() => {
      controller.enqueue(num++);
      if (num === 10) {
        controller.close();
        clearInterval(interval);
      }
    }, 1_000);
  },
});

start() 함수를 보면 controller가 인자로 넘어오고 setInterval() 함수 안에서 controller.enqueue()를 통해 스트림에 숫자를 하나씩 내부 데이터 큐에 넣어주고 있습니다. 그리고 숫자가 10에 다다르면 데이터 공급을 끊기 위해서 controller.close()를 호출하고 있습니다.

start() 함수를 사용하면 데이터를 누가 읽든 말든 상관없이 스트림이 생성 시점에 데이터가 미리 공급이 되는 특징이 있습니다.

데이터 스트림을 읽는 방법 1

스트림으로 부터 데이터를 잆으려면 ReadableStream 객체를 상대로 getReader() 함수를 호출하여 리더(reader)를 얻어야하는데요. 스트림은 리더를 얻는 순간 잠겨버리기(locked) 때문에 여러 리더로 읽을 수 없습니다. (tee() 함수를 통해서 스트림을 2개로 쪼개서 읽는 방법이 있는데 이 부분은 추후 별도의 포스팅에서 다루도록 할께요.)

이 리더의 read() 함수를 호출하면 본격적으로 스트림에서 제공하는 데이터를 비동기로 읽을 수 있는데요. read() 함수는 done과 value 속성으로 이루어진 객체를 프라미스(promise)로 반환합니다. done은 스트림에서 데이터를 모두 읽었는지 여부를 나타내고, value에는 바로 읽은 데이터가 들어있죠.

그럼 위에서 만든 읽기 가능한 스트림의 데이터를 한번 읽어볼까요?

const reader = stream.getReader();
reader.read().then(function print({ done, value }) {
  if (done) return console.log("done");
  console.log({ value });
  reader.read().then(print);
});

리더의 read() 함수를 호출하여 얻은 프라미스의 then() 함수를 호출한 후, done이 true가 될 때까지 재귀적으로 리더의 read() 함수를 호출하고 있습니다.

작성한 코드를 실행해보면 다음과 같이 스트림이 생성하자 마자 start() 함수가 호출되어 start라는 메세지가 콘솔에 먼저 찍힐테고요. 그 밑으로는 1초 간격으로 start() 함수를 통해 공급된 숫자 데이터가 1초마다 출력되는 것을 볼 수 있으실 것입니다. 마지막에는 스트림의 데이터가 소진되어 then() 함수의 재귀 호출이 종료되고 done이라는 메세지가 찍히게 됩니다.

start
{ value: 0 }
{ value: 1 }
{ value: 2 }
{ value: 3 }
{ value: 4 }
{ value: 5 }
{ value: 6 }
{ value: 7 }
{ value: 8 }
{ value: 9 }
done

위 코드가 잘 이해되지 않으시는 분들은 자바스크립트의 Promise에 대한 포스팅를 먼저 읽어보시면 도움이 되실 겁니다.

필요 시 데이터를 공급해주기

읽기 가능한 스트림에 데이터를 공급하는 또 다른 방법으로 underlyingSource 객체의 pull() 함수를 구현해줄 수도 있는데요. 스트림이 생성될 때 딱 한 번 호출되는 start() 함수와 달리 pull() 함수는 스트림에서 데이터를 읽어갈 때 마다 호출이 됩니다.

예를 들어, 위 동일하게 1초 간격으로 숫자 0부터 10까지의 데이터를 리더가 요청할 때 마다 공급해주는 스트림을 만들어보겠습니다.

const stream = new ReadableStream({
  num: 0,
  async pull(controller) {
    console.log("pull");
    await new Promise((r) => setTimeout(r, 1_000));
    controller.enqueue(this.num++);
    if (this.num === 10) {
      controller.close();
    }
  },
});

pull() 함수도 start() 함수와 비슷한 방식으로 controller.enqueue()와 controller.close()를 사용하여 데이터를 공급하거나 공급을 끊을 수 있습니다.

데이터 스트림을 읽는 방법 2

위에서 데이터 스트림을 읽을 때 프라미스의 then() 함수를 재귀적으로 호출했었는데 코드가 읽기 좀 힘들었죠? 이번에는 await 키워드를 사용하여 좀 더 깔끔하게 동일한 코드를 재작성해보겠습니다.

const reader = stream.getReader();
while (true) {
  const { done, value } = await reader.read();
  if (done) {
    console.log("done");
    break;
  }
  console.log({ value });
}

while 문 안에서 리더의 read() 함수를 await 키워드를 붙여서 호출하면 바로 done과 value을 담은 객체가 반환됩니다. done이 true라면 반복을 마치고 done이 false라면 value를 출력하도록 구현하였습니다.

이 코드를 실행해보면 1초 간격으로 pull() 함수가 호출되어 pull 메시지와 함께 숫자 데이터가 출력되는 것을 볼 수 있으실 겁니다.

pull
{ value: 0 }
pull
{ value: 1 }
pull
{ value: 2 }
pull
{ value: 3 }
pull
{ value: 4 }
pull
{ value: 5 }
pull
{ value: 6 }
pull
{ value: 7 }
pull
{ value: 8 }
pull
{ value: 9 }
done

자바스크립트의 async/await에 대한 자세한 설명은 관련 포스팅를 참고 바랍니다.

start vs. pull

스트림에 데이터를 공급할 때 start() 함수와 pull() 함수 중에서 어떤 것을 사용하는 게 나을지 궁금하실 것 같습니다.

start() 함수를 사용하면 스트림에 데이터를 미리 준비해놓을 수 있기 때문에 스트림 데이터를 빨리 읽어가야하는 상황에서 유리합니다. 하지만 데이터를 미리 만들어놓을 수 없거나 데이터가 엄청 큰 경우에는 start() 함수 사용 시 메모리의 비효율적인 사용으로 이어질 수 있습니다.

반면에 pull() 함수를 사용하면 데이터를 요청량에 맞춰서 실시간으로 제공할 수 있다는 이점이 있습니다. 따라서 메모리에 한번에 올리기에는 부담스러운 대용량 데이터를 스트리밍하거나 LLM(대형 언어 모델)처럼 데이터 출력 속도가 느린 경우에는 pull() 함수를 사용하는 편이 유리할 것입니다.

물론 쾌적한 읽기 경험을 위해서 start() 함수와 pull() 함수를 적절히 섞어서 사용하는 전략도 있습니다. start() 함수를 통해 데이터의 일부를 미리 공급해놓고, pull() 함수를 통해서는 그 이후 데이터를 실시간으로 분할 공급하는 것이지요.

좀 억지스러운 예제이기는 하지만 0부터 4까지는 start() 함수로 미리 공급하고, 5부터 9까지는 pull() 함수로 실시간 공급을 하는 스트림을 한번 만들어보았습니다.

const stream = new ReadableStream({
  start(controller) {
    console.log("start");
    for (let num = 0; num < 5; num++) {
      controller.enqueue(num);
    }
  },
  num: 6,
  async pull(controller) {
    console.log("pull");
    await new Promise((r) => setTimeout(r, 1_000));
    controller.enqueue(this.num++);
    if (this.num === 10) {
      controller.close();
    }
  },
});

데이터 스트림을 읽는 방법 3

데이터 스트림을 읽는 가장 최신 방법으로 for await 문법이 있는데요. 위 두 가지 방법에서는 여러 줄이 필요했던 코드를 다음과 같이 초간단하게 작성할 수가 있습니다.

for await (const value of stream) {
  console.log({ value });
}

이 코드를 실행해보면 우선 start()를 통해 공급된 숫자가 우선 출력되고, pull()을 통해 공급된 숫자가 나중에 출력되는 것을 볼 수 있습니다.

start
{ value: 0 }
{ value: 1 }
{ value: 2 }
{ value: 3 }
pull
{ value: 4 }
{ value: 6 }
pull
{ value: 7 }
pull
{ value: 8 }
pull
{ value: 9 }
done

아쉽게도 읽기 가능한 현재 글 작성일 기준으로 파이어폭스에서만 스트림을 상대로 for await를 사용할 수 있으며 다른 브라우저에서 지원될 때까지는 좀 기다려셔야 할 것 같습니다. 다행히도 Node.js에서는 지원이 되기 때문에 백엔드에서 바로 쓰실 수 있는 문법입니다.

마치면서

지금까지 다양한 예제를 통해서 자바스크립트에서 어떻게 스트림을 생성하고 데이터를 읽는지에 대해서 살펴보았습니다.

실전에서는 이렇게 읽기 기능한 스트림을 직접 만들기보다는 이미 만들어진 스트림을 읽어야하는 경우가 더 많을 것입니다. 대표적으로 fetch API를 통해서 원격 스트림을 읽어오는 것을 들 수 있겠는데요. 이 부분에 대해서는 추후 포스팅해보도록 하겠습니다.