javascript - rxjs オブザーバーの強制完了

Question

tail -f のように、ファイルを永久に追跡する rxjs オブザーバー (実際にはサブジェクト) があります。たとえば、ログファイルを監視するのに最適です。

この「永久に」動作は、私のアプリケーションには最適ですが、テストにはひどいものです。現在、アプリケーションは動作しますが、テストが永遠にハングします。

私のテストコードはファイルに何行あるべきかを知っているので、オブザーバーの変更を強制的に早期に完了させたいと思っています。どうすればいいですか？

返された Subject ハンドルで onCompleted を呼び出してみましたが、その時点では基本的にオブザーバーとしてキャストされており、強制的に閉じることはできません。エラーは次のとおりです。

オブジェクト # にはメソッド「onCompleted」がありません

ソースコードは次のとおりです。

function ObserveTail(filename) {

source = new Rx.Subject();

if (fs.existsSync(filename) == false) {
    console.error("file doesn't exist: " + filename);
}

var lineSep = /[\r]{0,1}\n/;
tail = new Tail(filename, lineSep, {}, true);

tail.on("line", function(line) {
        source.onNext(line);
});
tail.on('close', function(data) {
    console.log("tail closed");
    source.onCompleted();
});     
tail.on('error', function(error) {
    console.error(error);
});     

this.source = source;
}           

そして、永久に強制的に終了させる方法を理解できないテスト コード (テープ スタイル テスト) を次に示します。「ILLEGAL」行に注意してください。

test('tailing a file works correctly', function(tid) {

var lines = 8;
var i = 0;
var filename = 'tape/tail.json';
var handle = new ObserveTail(filename);
touch(filename);

handle.source
.filter(function (x) {
    try {
        JSON.parse(x);
        return true;
    } catch (error) {
        tid.pass("correctly caught illegal JSON");
        return false;
    }
})
.map(function(x) { return JSON.parse(x) })
.map(function(j) { return j.name })
.timeout(10000, "observer timed out")
.subscribe (
    function(name) {
        tid.equal(name, "AssetMgr", "verified name field is AssetMgr");
        i++;
        if (i >= lines) {
            handle.onCompleted();   // XXX ILLEGAL
        }
    },
    function(err) {  
        console.error(err)
        tid.fail("err leaked through to subscriber");
    },
    function() {
        tid.end();
        console.log("Completed");
    }
);

})