我有一个 rxjs 观察器(实际上是一个 Subject)，它永远跟踪一个文件，就像 tail -f 一样。例如，它非常适合监控日志文件。

这种“永远”的行为对我的应用程序来说很好，但对测试来说却很糟糕。目前我的应用程序可以运行，但我的测试永远挂起。

我想强制观察者更改提前完成，因为我的测试代码知道文件中应该有多少行。我该怎么做？

我尝试在我返回的 Subject 句柄上调用 onCompleted 但此时它基本上被转换为观察者并且你不能强制它关闭，错误是:

Object # has no method '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;
}           

下面是无法弄清楚如何强制永远结束的测试代码(磁带样式测试)。注意“非法”行:

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

})

最佳答案

听起来你解决了你的问题，但对于你原来的问题

I'd like to force an observer change to complete early, because my test code knows how many lines should be in the file. How do I do this?

一般来说，当您有更好的选择时，不鼓励使用 Subject，因为它们往往是人们使用他们熟悉的编程风格的拐杖。与其尝试使用 Subject，我建议您考虑每个事件在 Observable 生命周期中的含义。

包装事件发射器

EventEmitter#on/off 模式的包装器已经存在，其形式为 Observable.fromEvent。它处理清理并仅在有监听器时保持订阅有效。因此 ObserveTail 可以重构为

function ObserveTail(filename) {

  return Rx.Observable.create(function(observer) {
    var lineSep = /[\r]{0,1}\n/;
    tail = new Tail(filename, lineSep, {}, true);
    var line = Rx.Observable.fromEvent(tail, "line");
    var close = Rx.Observable.fromEvent(tail, "close");
    var error = Rx.Observable.fromEvent(tail, "error")
                  .flatMap(function(err) { return Rx.Observable.throw(err); });
    //Only take events until close occurs and wrap in the error for good measure
    //The latter two are terminal events in this case.
    return line.takeUntil(close).merge(error).subscribe(observer);
  });
} 

与普通使用 Subjects 相比，它有几个好处，第一，您现在可以在下游实际看到错误，第二，这将在您完成事件后处理事件清理。

避免*同步方法

然后这可以在不使用 readSync

//If it doesn't exist then we are done here
//You could also throw from the filter if you want an error tracked
var source = Rx.Observable.fromNodeCallback(fs.exists)(filename)
    .filter(function(exists) { return exists; })
    .flatMap(ObserveTail(filename));

接下来，您可以使用 flatMap 来简化过滤器/ map / map 序列。

var result = source.flatMap(function(x) {
  try {
    return Rx.Observable.just(JSON.parse(x));
  } catch (e) {
    return Rx.Observable.empty();
  }
}, 
//This allows you to map the result of the parsed value
function(x, json) {
  return json.name;
})
.timeout(10000, "observer timed out");

不要发出信号，取消订阅

当流仅沿一个方向传播时，如何停止“发出停止信号”。我们实际上很少想让观察者直接与 Observable 通信，因此更好的模式是不实际“发出”停止信号，而是简单地取消订阅 Observable 并将其留给 Observable 的行为从那里确定它应该做什么。

本质上，您的 Observer 真的不应该关心您的 Observable 而不仅仅是说“我在这里完成了”。

为此，您需要声明停止时要达到的条件。

在这种情况下，由于您只是在测试用例中的一组数字后停止，您可以使用 take 取消订阅。因此最终的订阅 block 看起来像:

result
 //After lines is reached this will complete.
 .take(lines)
 .subscribe (
    function(name) {
        tid.equal(name, "AssetMgr", "verified name field is AssetMgr");
    },
    function(err) {  
        console.error(err)
        tid.fail("err leaked through to subscriber");
    },
    function() {
        tid.end();
        console.log("Completed");
    }
);

编辑 1

正如评论中所指出的，在这个特定的 api 的情况下，没有真正的“关闭”事件，因为 Tail 本质上是一个无限操作。从这个意义上说，它与鼠标事件处理程序没有什么不同，当人们停止监听时，我们将停止发送事件。所以你的 block 可能最终看起来像:

function ObserveTail(filename) {

  return Rx.Observable.create(function(observer) {
    var lineSep = /[\r]{0,1}\n/;
    tail = new Tail(filename, lineSep, {}, true);
    var line = Rx.Observable.fromEvent(tail, "line");
    var error = Rx.Observable.fromEvent(tail, "error")
                  .flatMap(function(err) { return Rx.Observable.throw(err); });
    //Only take events until close occurs and wrap in the error for good measure
    //The latter two are terminal events in this case.
    return line
            .finally(function() {  tail.unwatch(); })
            .merge(error).subscribe(observer);
  }).share();
} 

finally 和 share 运算符的添加创建了一个对象，该对象将在新订阅者到达时附加到尾部，并且只要有至少有一个订阅者还在听。然而，一旦所有订阅者都完成了，我们就可以安全地unwatch尾部了。

关于javascript - 强制完成 rxjs 观察者，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/35051883/