RxJs Basing a Filter on Another Observable

Question

У меня есть список слов с данными произношения в текстовом файле. Я хотел бы сделать, чтобы пользователь вводил слово, а затем проверил программу, чтобы узнать, есть ли у меня данные на этом слове в этом файле. Я бы хотел сделать это в RxJs, для чего я новичок.

Код, приведенный ниже, ближе всего подходит для того, что я хочу. В моем основном потоке у меня есть вызов фильтра, который создает зависимый поток, называемый «checkstream». Я не понимаю, как использовать результаты этого зависимого потока в методе фильтрации моего основного потока. В настоящее время этот метод фильтра выходит из строя, но я до сих пор получаю данные на экране при помощи консоли.

Если в текстовом файле для слова есть данные, тогда контрольный поток будет являться наблюдаемым, содержащим только данные, которые я хочу получить и показать пользователю. Я хочу каким-то образом передать эти данные потребителю моего основного потока, но я не понимаю, как это сделать.

Я хотел бы получить любую помощь, которую вы можете предоставить. Даже некоторая интуиция была бы полезна.

var Rx = require('rx'); 
var RxNode = require('rx-node'); 
var fs = require('fs'); 
var split = require('split'); 

    RxNode.fromReadableStream(process.stdin) 
     .map((inputData) => { 
      var inputWord = inputData.toString().trim(); 
      return inputWord; 
     }) 
     .map((inputWord) => { 

      var checkStream = fs.createReadStream('./dict.txt'); 

      RxNode.fromReadableStream(checkStream.pipe(split())) 
       .map((lineFromFile) => { 
        return JSON.parse(lineFromFile); 
       }) 
       .filter((parsedDataToCheck) => { 
        return parsedDataToCheck.word.toLowerCase().trim() === inputWord; }) 
       .subscribe((dataThatMatches) => { console.log(dataThatMatches) }); 

      return dataToReturn; 
     }) 
     .subscribe(function(dataToReturn) { 
      console.log(dataToReturn); 
     }); 

Answer 1

Может быть что-то вроде этого:

var Rx = require('rx'); 
var RxNode = require('rx-node'); 
var fs = require('fs'); 
var split = require('split'); 

RxNode.fromReadableStream(process.stdin).map( 
    inputData => inputData.toString().trim() 
).flatMap(inputWord => { 
    var checkStream = fs.createReadStream('./dict.txt'); 
    return RxNode.fromReadableStream(
     checkStream.pipe(split()) 
    ).map(
     lineFromFile => JSON.parse(lineFromFile) 
    ).find(
     parsedDataToCheck => parsedDataToCheck.word.toLowerCase().trim() === inputWord 
    ); 
}).subscribe(dataToReturn => { 
    console.log(dataToReturn); 
}); 

Обратите внимание, что это возможно, что входные слова из стандартного ввода получить заказана после фильтрации из-за асинхронной фс читает.