Wie Knoten lesbaren Stream zu RX beobachtbaren

Question

konvertieren, wenn ich einen Knoten js Strom haben, sagen wir zum Beispiel von so etwas wie process.stdin oder aus fs.createReadStream, wie kann ich diese konvertieren mit RxJs5 einen RxJs beobachtbare Strom zu sein?

Ich sehe, dass RxJs-Node hat eine fromReadableStream Methode, aber das sieht aus wie es nicht in knapp einem Jahr aktualisiert wurde.

Answer 1

Für jeder, der danach sucht, folgt Marks Empfehlung 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

Die RxJs-Node-Implementierung RxJs4 basiert, kann aber ohne viel Arbeit https://github.com/Reactive-Extensions/rx-node/blob/87589c07be626c32c842bdafa782fca5924e749c/index.js#L52

Answer 3

Die auf RxJs5 portiert werden folgende sollte sowohl für die Arbeit v4 und v5 (Haftungsausschluss ungetestet):

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(); 
    },