Если у меня есть поток Node js, скажем, из чего-то вроде process.stdin или из fs.createReadStream, как я могу преобразовать его в поток RxJs Observable, используя RxJs5?
Я вижу, что RxJs-Node имеет метод fromReadableStream, но похоже, что его нет. был обновлен почти через год.
Для тех, кто ищет это, следуя рекомендации Марка, я адаптировал реализацию rx-node fromStream для 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();
}
Обратите внимание, что он по своей сути нарушает все функции обратного давления потоков. Observables — это технология push. Все входные фрагменты будут прочитаны и переданы наблюдателю как можно быстрее. В зависимости от вашего случая, это может быть не лучшим решением.
Следующее должно работать как для версии 4, так и для версии 5 (отказ от ответственности не проверено):
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();
},
Приведенные выше ответы будут работать, хотя противодавление не поддерживается. Если вы пытаетесь прочитать большой файл с помощью createReadStream, они прочитают его целиком в памяти.
Вот моя реализация с поддержкой противодавления: rxjs-stream
Реализация RxJs-Node основана на RxJs4, но может быть перенесена на RxJs5 без особых усилий https://github.com/Reactive-Extensions/rx-node/blob/87589c07be626c32c842bdafa782fca5924e749c/index.js#L52
Поскольку потоки Node v11.14.0 поддерживают for await https://nodejs.org/api/stream.html#stream_readable_symbol_asynciterator
это означает, что вы можете передать поток оператору from().
Под капотом rxjs(v7.x.x) вызовет fromAsyncIterable(), который вернет Observable
