RxJS subscribe never finishes

Question

I'm fairly new with rxjs. I'm calling a function below and the complete stream is read and the read console statements are printed, but I never see a "Subscibe done" and I don't know why. What will it take to get this stream to finish? Is something obviously wrong?

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

I'm not sure about this *specific observable*, but not every observable will finish. — Omri Luzon, Jun 04 '17 at 02:06
What Node stream are you passing to `RxNode.fromReadLineStream`? Does the stream itself end? — cartant, Jun 04 '17 at 02:15
Yes, it's a RxNode.fromReadLineStream of a file with a finite length. — John Estess, Jun 04 '17 at 02:20
Ended up using a readline.on close event to achieve nearly the same thing. — John Estess, Jun 04 '17 at 04:42

score 1 · Answer 1 · answered Jun 04 '17 at 06:29

You can see from the source code that it send complete notification only when the source stream emits close event. https://github.com/Reactive-Extensions/rx-node/blob/master/index.js#L100-L102

So if you need the proper complete handler to be called you'll need to close the stream yourself, see How to close a readable stream (before end)?.
In other words the Observable doesn't complete automatically after reading the entire file.

RxJS subscribe never finishes

1 Answers1