将 Observable 实现到持久队列库中

Question

目前正在编写一个小的持久队列库，它将读取/写入文本文件中的行。这里是add方法，例如：

Queue.prototype.add = function(line, cb){

    getLock(this, err => {
        if(err){
            this.emit('error', err);
            releaseLock(err, cb);
        }
        else{
            fs.appendFile(this.filepath, line, err => {
               err && this.emit('error', err);
               releaseLock(err, cb);
            });
        }
    });
};

我觉得很尴尬的是支持事件发射器和回调（或事件发射器和承诺）。

换句话说，对于队列中的每个方法（添加、查看、删除），我需要返回/回调一个特定于每个调用的结果。使用事件发射器仅意味着调用者可能会针对并非特定于他们刚刚进行的调用的结果采取行动。所以这里的回调或承诺似乎势在必行——你不能只使用事件发射器。

我想知道的是 - observables 能否以某种方式解决必须将回调与事件发射器或承诺与事件发射器配对的问题？

我正在寻找一种方法来实现这个事件/异步队列，只有一种类型的异步回调机制。也许 observables 不是这里的答案，但我仍然在寻找一个好的设计模式。

Answer 1

我不太清楚你为什么在这里需要事件发射器....如果你使用 observables，每个订阅者都会从他们自己的调用中得到结果/错误。

我会这样重写你的方法：

function appendFileObs(filePath, line){
    return Rx.Observable.create((obs) => {
        fs.appendFile(filePath, line, (err, result) => {
            if(err) obs.onError(err);
            else {
                obs.onNext(result);
                obs.onCompleted();
            }
        });
    });
});
// Similar for getLock and releaseLock


Queue.prototype.add = function(line){
    return getLockObs(this)
        .flatMap(() => appendFileObs(this.filePath, line))
        .flatMap(result => releaseLockObs(undefined).map(() => result))
        .catch((err) => {
            return releaseLockObs(err);
        });
};

在这个解决方案中，我对流内部有副作用并不感到自豪，它可能是可以改进的，但你明白了。

这样，当有人调用 .add(line).subscribe() 时，它会得到结果和他调用时发生的错误。

如果您需要广播发生的错误，您可以使用 BehaviourSubject，它同时是一个观察者和可观察者（有用的东西！）