RxJS訂閱永遠不會完成

Question

我是rxjs的新手。我在下面調用一個函數，讀完整個流並打印讀取控制檯語句，但我從來沒有看到「Subscibe done」，我不知道爲什麼。要完成此流將需要什麼？是明顯的錯嗎？

const readline$ = RxNode.fromReadLineStream(rl) 
    .filter((element, index, observable) => { 
     if (index >= range.start && index < range.stop) { 
      console.log(`kept line is ${JSON.stringify(element)}`); 
      return true; 
     } else { 
      console.log(`not keeping line ${JSON.stringify(element)}`); 
      return false; 
     } 
    }) 
    .concatMap(line => Rx.Observable.fromPromise(myFunction(line))) 
    .do(response => console.log(JSON.stringify(response))); 

readline$.subscribe(i => { console.log(`Subscribe object: ${util.inspect(i)}`); }, 
        err => { console.error(`Subscribe error: ${util.inspect(err)}`); }, 
       done => { console.log("Subscribe done."); // NEVER CALLED 
          anotherFunc();     // NEVER CALLED 
        } 
); 

Answer 1

您可以從源代碼，它發送完成通知，只有當源流發出close事件看。 https://github.com/Reactive-Extensions/rx-node/blob/master/index.js#L100-L102

因此，如果您需要調用正確的完整處理程序，則需要自行關閉流，請參閱How to close a readable stream (before end)?。
換句話說，在讀完整個文件後，Observable不會自動完成。