Node.js Stream API에 대해 알아보자!

시작하기 전에

해당 블로그에 작성되는 글은 주인장의 지극히 주관적인 생각이 다수이며, 대부분의 지식은 구글링을 통해 얻고 있기 때문에 옳지않은 정보가 있습니다. 

잘못된 부분이나 수정해야 하는 부분이 있다면 과감히 덧글을 남겨주세요! 모르는게 많은 새싹입니다

오늘의 주제

오늘의 주제는 Node.js v11.5.0 공식 Document에 있는 Stream API에 대한 내용입니다.

스트림과 버퍼에 대해 아직 잘 모르겠다 싶으신 분은 제가 지난 시간에 작성했던 "스트림과버퍼"에 관한 글을 통해 글을 읽고 오시면 좀 더 도움이 많이 될 것 같아요.  그럼 이제 시작해 보도록 하겠습니다.

Nodejs - stream

아래는 node.js v11 공식 document에 있는 Stream에 대한 설명입니다.

스트림은 Node.js에서 데이터 스트리밍 작업을 위한 추상적잉ㄴ 인터페이스이다. 스트림 모듈은 스트림 인터페이스를 구현하는 객체를 쉽게 만들 수 있는 기본 API를 제공한다.

Node.js에서는 많은 스트림 객체를 지원합니다. 예를들어 HTTP server 요청을 위한 것이나. process.stdout이 모두 스트림 객체를 통해 이루어 집니다.

스트림은 읽거나 쓸수 있으며, 모든 스트림 객체는 EventEmitter의 인스턴스 입니다.

라고 나와있네요.

Node.js에서 스트림의 타입은 총 4가지가 있습니다.

Writable - 데이터를 작성할 수 있는 스트림입니다. ex) fs.createWriteStream()

Readable - 데이터를 읽어들일 수 있는 스트림입니다. ex) fs.createReadStream()

Duplex - 데이터의 읽기(Readable)과 쓰기(Writable) 모두 가능합니다. (ex net.Socket)

Transform - Duplex 스트림은 수정하거나 변환이 가능합니다. ex) zlib.createDeflate()

부가적으로 이 모듈은 pipeline과 finished를 포함하고 있습니다.

Node.js API에서 생성된 모든 스트림은 문자열 및 버퍼 객체에서만 작동하지만 다른 유형의 javascript 값과도 함께 작동이 가능합니다. 이러한 스트림은 "객체 모드"에서 작동하는 것으로 간주합니다. Node.js Stream에서는 Buffer기능 또한 제공합니다.

Writable 그리고 Readable 스트림은 내부 버퍼에 데이터를 저장 하며, writable.writableBuffer 혹은 readableBuffer를 통해 검색이 가능합니다.

버퍼데이터의 양은 highWatermark에 따라 달라질 수 있습니다. 일반 버퍼의 경우 highWatermark는 버퍼의 크기를 나타내며 객체모드의 경우는 객체의 수를 나타냅니다.

Node.js에서 Readable Streams과 Writable 스트림에 속하는 네이티브 노드 객체들 입니다.

Writable Streams

Writable 객체에서 제공하는 Event는 다음과 같습니다.

close => 스트림이 닫힌 경우 발생하는 이벤트입니다. 그 후의 이벤트는 발생하지 않습니다.

drain => writable stream이 추가로 데이터 쓰기 작업이 가능한 경우 발생합니다.

error => 쓰기작업이나 파이핑 데이터가 오류가 생긴경운 발생하는 이벤트 입니다.

finish => stream.end()(모든 데이터가 flush())후에 발생하는 이벤트 입니다.

pipe => 읽기 스트림이 쓰기스트림을 목적지로 감을 경우 발생합니다. ex) reader.pipe(writer)

unpipe => 읽기 스트림이 쓰기 스트림을 unpipe하는 경우 발생합니다. ex) reader.unpipe(wirter)

모든 이벤트는 EventEmitter의 인스턴스이며 데이터를 읽거나 쓸 때 각각의 이벤트를 방출 합니다.

Writable's Method

writable 객체에서 제공하는 메소드는 다음과 같습니다.

writable.cork() => 데이터를 버퍼에[ 강제로 저장한다. uncork()이나 .end()의 경우 flush됩니다.

writable.destroy([error]) => 스트림을 제거하고 error와 close 이벤트를 발생 시킵니다.

writable.end([chunk][, encoding][, callback]) => writable의 종료를 알린다.

writable.setDefaultEncoding => Default 인코딩을 설정해준다.

writable.uncork() => 버퍼에 담긴 데이터를 flush시킨다.

writable.writable => boolean true or false

writable.writableLength => highwatermark

writable.write(chunk[, encoding][,callback])

Readable Streams

Readable Stream은 flowing이나 paused 모드중 하나에서 효율 적으로 작동합니다. 이 모드들은 객체 모드로 분리됩니다. Readable stream이 object 모드이건 아니건 두개의 모드와 관계 없이 작동합니다.

flowing 모드에서는 데이터가 기본 시스템에서 자동으로 읽혀지며 EventEmitter 인터페이스를 이용하여 어플리케이션에 제공된다. puase 모드에서는 데터를 읽기위해서는 stream.read() 메서드를 명시적으로 호출해야 합니다.

그렇다면 pause 모드에서 다시 flowing 모드로 가려면 어떻게 해야할까? 총 세가지의 방법이 있습니다.

'data' 이벤트 핸들러 추가

stream.resume() 메서드 호출

Writable로 보내기 위하여 stream.pipe() 메소드를 호출한 경우

Readable 객체에서 제공하는 이벤트는 다음과 같습니다.

1. data 이벤트 => 스트림에서 데이터를 소비자로 전달해 줄 때 마다, 이벤트가 발생합니다.

   close 이벤트 => 모든 readable 스트림이 colse 이벤트를 가지고 있는것은 아니며 스트림이나 스트림 내부의 리소스가 종료되었을 때 발생합니다.

2. data 이벤트 => 스트림에서 데이터를 소비자로 전달해 줄 때 마다, 이벤트가 발생합니다.

   const { Readable } = require('stream')
   const readable = new Readable();
   // readable.setEncoding()을 설정할 수 있습니다.
   ​
   readable.on('data', (chunk : <buffer>|<String>|<array>) => {
     console.log(`Received ${chunk.length} bytes of data.`);
   })

3. end 이벤트 => 스트림에 더이상 소모할 데이터가 남아있지 않은 경우 발생합니다.

   const { Readable } = require('stream')
   const readable = new Readable();
   ​
   readable.on('data', (chunk) => {
     console.log(`Received ${chunk.length} bytes of data.`);
   })
   ​
   readable.on('end', (chunk) => {
     console.log('There will be no more data.');
   })

4. error 이벤트 => Readable이 실행되는 시간에 내부적으로 스트림을 생성할 수 없거나 유효하지 않은 청크를 읽어들일때 발생합니다. callback은 Error 객체를 반환합니다.

5. readable 이벤트 => readable 스트림으로 부터 읽어들일 수 있는 데이터가 있을 때 발생합니다. readable 이벤트에 대한 리스너를 연결하면 일부 데이터를 내부 버퍼로도 읽어들일 수 있습니다. 데이터의 끝에서 readable 이벤트가 발생한 후에 end 이벤트도 발생하게 됩니다.

   const fs = require('fs');
   const rr = fs.createReadStream('foo.txt');
   rr.on('readable', () => {
     console.log(`readable: ${rr.read()}`);
   });
   rr.on('end', () => {
     console.log('end');
   });
   ​
   // redable: null
   // end

Readable's Methods

1. readable.destroy([error])

2. readable.pause() => redable 객체를 pause상태로 만들어 줍니다.

3. readable.pipe(destination[, options]) => writable과 바로 연결 시켜 줍니다.

4. readable.read([size]) => Returns: <string> | <buffer> | <null> | <any>

5. readable.readable => boolean : true or false

6. readable.readableHighWaterMark => readable에 등록된 highWaterMark , return Number

7. redable.redableLength => 프로퍼티가 가지고 있는 바이트의 수

8. readable.resume() => pause 모드에서 flowing 모드로 변환시켜 줄떄 사용

9. readable.setEncoding(encoding) => 초기 인코딩을 설정해준다.

10. readable.unpipe([destination])

11. redable.unshift(chunk) => 받아온 데이터를 다시 내부 버퍼로 이동하는 작업을 실행 합니다. 데이터의 일부를 다른곳으로 넘겨주기 위해서!

글이 너무 길어져서 Duplex와 Transform에 대한 부분은 나눠서 포스팅 하도록 하겠습니다. 거의 도큐에 있는 내용을 번역해온 글이지만 혹시나 엉어 울렁증이 심한 분들을 위해서 짧게 정리해 보았습니다.

좀 더 자세한 내용이 필요하신 분은 직접 Node.js 공식문서(여기)를 통해 읽어보시면 좋을것 같습니다.