노드가 읽을 수있는 스트림을 RX 관찰 가능으로 변환하는 방법

Question

process.stdin 또는 fs.createReadStream과 같은 노드 js 스트림이있는 경우 RxJs5를 사용하여이 스트림을 RxJs Observable 스트림으로 변환하려면 어떻게해야합니까?

나는 RxJs-Node에 fromReadableStream 메서드가 있지만 1 년 내에 업데이트되지 않은 것 같습니다.

Answer 1

들어 마크의 추천에 따라 이것을 찾는 사람은 I adapted rx-node fromStream implementation for rxjs5입니다.

import { Observable } from 'rxjs'; 

// Adapted from https://github.com/Reactive-Extensions/rx-node/blob/87589c07be626c32c842bdafa782fca5924e749c/index.js#L52 
export default function fromStream(stream, finishEventName = 'end', dataEventName = 'data') { 
    stream.pause(); 

    return new Observable((observer) => { 
    function dataHandler(data) { 
     observer.next(data); 
    } 

    function errorHandler(err) { 
     observer.error(err); 
    } 

    function endHandler() { 
     observer.complete(); 
    } 

    stream.addListener(dataEventName, dataHandler); 
    stream.addListener('error', errorHandler); 
    stream.addListener(finishEventName, endHandler); 

    stream.resume(); 

    return() => { 
     stream.removeListener(dataEventName, dataHandler); 
     stream.removeListener('error', errorHandler); 
     stream.removeListener(finishEventName, endHandler); 
    }; 
    }).share(); 
} 

Answer 2

RxJs 노드 구현 RxJs4을 기반으로하지만, 많은 작업없이 RxJs5에 이식 할 수 https://github.com/Reactive-Extensions/rx-node/blob/87589c07be626c32c842bdafa782fca5924e749c/index.js#L52

Answer 3

V4 및 V5 (면책 조항 검증되지 않은) 모두에 대해 다음과 같은 작업을해야합니다 :

fromStream: function (stream, finishEventName, dataEventName) { 
    stream.pause(); 

    finishEventName || (finishEventName = 'end'); 
    dataEventName || (dataEventName = 'data'); 

    return Observable.create(function (observer) { 

     // This is the "next" event 
     const data$ = Observable.fromEvent(stream, dataEventName); 

     // Map this into an error event 
     const error$ = Observable.fromEvent(stream, 'error') 
     .flatMap(err => Observable.throw(err)); 

     // Shut down the stream 
     const complete$ = Observable.fromEvent(stream, finishEventName); 

     // Put it all together and subscribe 
     const sub = data$ 
     .merge(error$) 
     .takeUntil(complete$) 
     .subscribe(observer); 

     // Start the underlying node stream 
     stream.resume(); 

     // Return a handle to destroy the stream 
     return sub; 
    }) 

    // Avoid recreating the stream on duplicate subscriptions 
    .share(); 
    },