stream 사용(Node.js와 web client 간 통신)

API Server에서 list를 불러오는 GET method를 구현하면서 매우 많이 양의 데이터가 list될 수 있는 상황이 생겨서 stream를 구현하게 되었다.

server측 stream은 Node.js 내장 module에 stream이 구현되어 있어 쉽게 사용할 수 있다.

const stream = require(‘stream’);

stream은 4가지의 기본 type을 가진다.

• Writable: data를 쓸 수 있는 stream
• Readable: data를 읽을 수 있는 stream
• Duplex: Readable과 Writable 모두 할 수 있는 stream
• Transform: data를 수정하거나 변형할 수 있는 stream - Duplex stream이다.

mongodb에서 cursor를 사용하여 stream을 만들 수 있다. Mongoose에서 cursor는 node.js의 readable stream을 확장한 것이다.

import { QueryCursor } from ‘mongoose’;
import * as jsonStream from ‘JSONStream’;

async function getList(req, res, next) {
    const cursor: QueryCursor = model.find({ enabled: true }).cursor();
    cursor
        .pipe(jsonStream.stringify()) // jsonStream.stringify()을 사용하지 않으면 error 발생
        .pipe(res.type('json’)) // header에 Content-Type: application/json을 설정
}

 

cursor(Readable stream)을 jsonStream.stringify(Transform stream)으로 연결한다. jsonStream.stringify는 json객체는 string으로 변형하는 Transform stream이다. express의 response는 Writable stream이다. 위와 같이 stream를 연결하면 db에서 읽은 전체 list를 memory에 올리지 않고도 응답을 stream으로 전송할 수 있다.

client에서도 stream 을 받을 수 있도록 구현해야한다. client에서 request를 위해 axios를 사용한다. axios를 browser에서 사용하는 경우 XMLHttpRequests를 사용하게 된다. XMLHttpRequests를 사용하는 경우 stream 방식은 binary형태로 응답을 받아 처리하는 로직을 구현해야하는 것 같다.

 

fetch를 사용하면 Readable stream을 반환받을 수 있다. fetch를 사용하여 stream 구현은 이전 포스팅 했었다. 참고

 

이번에는 gaxios를 사용하여 구현했다.

import * as gaxios from "gaxios";
import ndjsonStream from "can-ndjson-stream";

const result = await gaxios.request({
    method: "get",
    url: "https://example.com/asyc",
    responseType: "stream",
});
const exampleReader = ndjsonStream(result.data).getReader();
let readerResult;
while (!readerResult || !readerResult.done) {
    readerResult = await exampleReader.read();
}

gaxios를 사용하면 위와 같이 steam을 처리할 수 있다. result.data가 readable 객체로 반환된다. ndjsonStream을 사용하여 multi line json을 쉽게 parsing할 수 있다.

 

stream을 처리하면서 어려웠던 점은 node.js에서 readable 객체와 client에서의 readable 객체가 다르다는 것이었다. Node.js의 readable 객체는 on으로 callback listener를 사용할 수 있지만 client readable은 callback설정을 할 수가 없었다. client readable은 위와같이 추가 구현이 필요한 형태이다.

 

stream을 대체할 방법을 찾을 때 몇가지 대안으로 확인했던 방법을 적어본다.

 

1. GraphQL을 사용한다면 subscription을 사용할 수 있다.
2. client에서 서버에 요청을 보내지 않고 server에서 응답만 받는 형태라면 SSE(Server Sent Event)를 구현하는 방법도 있다. SSE는 HTML5의 표준이다.
3. Web socket를 구현하는 방법이다. API gateway와 dynamoDB를 사용하여 socket server를 운영하는 예시를 많이 찾와봤다. 참고