単純なユースケースでのRxJS5による再帰の回避

Question

再帰を回避する方法を解明しようとしています（可能な場合）。 RxJSを使用してキュー上にメソッドを作成しています。キューが空でない場合、drainメソッドは再帰的に自身を呼び出します。ドレインメソッドは現在、キューから空になるまで一度に1つの項目を削除するように設計されています。以下はこの目的のために働くと思われますが、再帰的にドレインを呼び出すことを避けることができるかどうかを調べたいと思います。再帰の問題は、再帰的な処理が完了するまで（おそらく私の推測である）再帰的メソッドからのアイテムを「返す」ことはできないということです。

だから私の質問は次のとおりです。

は、私が代わりにメソッドが再帰実行されたときに一度にキューからすべての排水のアイテムを受け取るので、個別キューから各項目を受け取るためにドレイン（）の加入者を求めています。どうすればそれを達成できますか？再帰的な方法でこれを達成できますか？それとも非再帰的な方法でしか達成できないのでしょうか？後者の場合、それを行う方法？このメソッドは、キューを排出し、キューは、我々はそれだけで再帰を使用するのが最も簡単だと再呼び出しそう 、ロックを解除、その後、毎回の項目を削除し、ロックする必要が 空のときにしようと停止します

キューが空でない場合はこの方法をドレイン

Queue.prototype.drain = function (opts) { 

    opts = opts || {}; 
    const delay = opts.delay || 500; 

    return this.init() 
     .flatMap(() => { 
      return acquireLock(this) 
       .flatMap(obj => { 
        return acquireLockRetry(obj) 
       }); 
     }) 
     .flatMap(obj => { 
      return removeOneLine(this) 
       .flatMap(l => { 
        return releaseLock(this, obj.id) 
         .map(obj => l); 
       }); 
     }) 
     .flatMap(() => { 
      return Rx.Observable.timer(delay) 
       .flatMap(() => { 
        return this.drain() /// <<< recurse 
         .takeUntil(this.isEmpty()); /// <<<< until 
       }); 
     }) 
     .catch(e => { 
      const force = !String(e.stack || e).match(/acquire lock timed out/); 
      return releaseLock(this, force); 
     }); 

}; 

キューが空の場合は//チェック

Queue.prototype.isEmpty = function() { 

    return this.init() 
     .flatMap(() => { 
      return acquireLock(this) 
       .flatMap(obj => { 
        return acquireLockRetry(obj) 
       }) 
     }) 
     .flatMap(obj => { 
      return findFirstLine(this) 
       .flatMap(l => { 
        return releaseLock(this, obj.id) 
         .map(obj => l); 
       }); 
     }) 
     .filter(l => { 
      // filter out any lines => only fire event if there is no line 
      return !l; 
     }) 
     .catch(e => { 
      const force = !String(e.stack || e).match(/acquire lock timed out/); 
      return releaseLock(this, force); 
     }); 

}; 

Answer 1

問題へ

一つの可能​​な解決策は、このようなものです：

const obs = new Rx.Subject(); 

    q.drain(obs).subscribe(function (v) { 
     console.log('end result => ', v); 
    }); 

    obs.subscribe(function (v) { 
     console.log('next item that was drained => ', v); 
    }); 

とドレイン方法は、単に次のようになります。

Queue.prototype.drain = function (obs, opts) { 

    opts = opts || {}; 

    const delay = opts.delay || 500; 

    return this.init() 
     .flatMap(() => { 
      return acquireLock(this) 
       .flatMap(obj => { 
        console.log(' drain lock id => ', obj.id); 
        return acquireLockRetry(obj) 
       }); 
     }) 
     .flatMap(obj => { 
      return removeOneLine(this) 
       .flatMap(l => { 
        return releaseLock(this, obj.id) 
         .map(obj => { 
          obs.next(l); 
          return l; 
         }); 
       }); 
     }) 
     .flatMap(() => { 
      return Rx.Observable.timer(500) 
       .flatMap(() => { 
        return this.drain(obs, opts) 
         .takeUntil(this.isEmpty()); 
       }); 
     }) 
     .catch(e => { 
      console.error('\n', ' => isEmpty() error => \n', e.stack || e); 
      const force = !String(e.stack || e).match(/acquire lock timed out/); 
      return releaseLock(this, force); 
     }); 

}; 

このように、毎回の項目が削除され、それが起動します待ち行列が完全に排水されたら最後にイベントを発することを願っています。