RxJSスロットルの動作。最初の値にすぐに

Question

例Plunkrを得る：​​3210

を私があることを認識していますrxjsのためにこれらの3つのスロットル方式（5.0 beta.4）：

auditTime()、throttleTime()とdebounceTime()

私は行動探しているのは、デフォルトでスロットルに1つのロダッシュがあります：

• 1）すぐに私に最初の価値を与えてください！連続した値で
• 2）は、与えられた遅延の値は、その後、スロットル遅延の期限が切れたときに
• 3）は、（1）

理論的には、これはのようになります状態に戻って最後の発生した値を発する保持します：中に放出された場合

inputObservable 
    .do(() => cancelPreviousRequest()) 
    .throttleTime(500) 
    .subscribe((value) => doNextRequest(value)) 

しかし

• throttleTimeは、私の最後の値を与えることはありませんスロットルタイムアウト
• debounceTimeはすぐ
• auditTimeをトリガしません、私が説明した動作を実現するためにrxjs方法のいずれかを組み合わせることができ、すぐに

をトリガしないのですか？

Answer 1

私はauditTime演算子を使用し、2行を変更して目的の動作を実現しました。

新plunker：オリジナルhttps://plnkr.co/edit/4NkXsOeJOSrLUP9WEtp0?p=preview

：

• https://github.com/ReactiveX/rxjs/blob/master/src/operator/auditTime.ts

の変更：から

（auditTime）：

protected _next(value: T): void { 
    this.value = value; 
    this.hasValue = true; 
    if (!this.throttled) { 
    this.add(this.throttled = this.scheduler.schedule(dispatchNext, this.duration, this)); 
    } 
} 

clearThrottle(): void { 
    const { value, hasValue, throttled } = this; 
    if (throttled) { 
    this.remove(throttled); 
    this.throttled = null; 
    throttled.unsubscribe(); 
    } 
    if (hasValue) { 
    this.value = null; 
    this.hasValue = false; 
    this.destination.next(value); 
    } 
} 

（auditTimeImmediate）へ

：値はnext EDた後

protected _next(value: T): void { 
    this.value = value; 
    this.hasValue = true; 
    if (!this.throttled) { 
     // change 1: 
     this.clearThrottle(); 
    } 
} 

clearThrottle(): void { 
    const { value, hasValue, throttled } = this; 
    if (throttled) { 
     this.remove(throttled); 
     this.throttled = null; 
     throttled.unsubscribe(); 
    } 
    if (hasValue) { 
     this.value = null; 
     this.hasValue = false; 
     this.destination.next(value); 
     // change 2: 
     this.add(this.throttled = this.scheduler.schedule(dispatchNext, this.duration, this)); 
    } 
} 

だから私はタイムアウトを開始します。

使用法：

inputObservable 
    .do(() => cancelPreviousRequest()) 
    .auditTimeImmediate(500) 
    .subscribe((value) => doNextRequest(value)) 

Answer 2

古いRxJの場合、私はあなたが望むもののほとんどを行うconcatLatest演算子を書いていました。このコードを使用すると、このコードで調整動作を得ることができます。

const delay = Rx.Observable.empty().delay(500); 
inputObservable 
    .map(value => Rx.Observable.of(value).concat(delay)) 
    .concatLatest() 
    .subscribe(...); 

ここに演算子があります。

Rx.Observable.prototype.concatLatest = function() { 
    /// <summary> 
    /// Concatenates an observable sequence of observable sequences, skipping sequences that arrive while the current sequence is being observed. 
    /// If N new observables arrive while the current observable is being observed, the first N-1 new observables will be thrown 
    /// away and only the Nth will be observed. 
    /// </summary> 
    /// <returns type="Rx.Observable"></returns> 
    var source = this; 

    return Rx.Observable.create(function (observer) { 
     var latest, 
      isStopped, 
      isBusy, 
      outerSubscription, 
      innerSubscription, 
      subscriptions = new Rx.Subscription(function() { 
       if (outerSubscription) { 
       outerSubscription.unsubscribe(); 
       } 
       if (innerSubscription) { 
       innerSubscription.unsubscribe(); 
       } 
      }), 
      onError = observer.error.bind(observer), 
      onNext = observer.next.bind(observer), 
      innerOnComplete = function() { 
       var inner = latest; 
       if (inner) { 
        latest = undefined; 
        if (innerSubscription) { 
         innerSubscription.unsubscribe(); 
        } 
        innerSubscription = inner.subscribe(onNext, onError, innerOnComplete); 
       } 
       else { 
        isBusy = false; 
        if (isStopped) { 
         observer.complete(); 
        } 
       } 
      }; 

     outerSubscription = source.subscribe(function (newInner) { 
      if (isBusy) { 
       latest = newInner; 
      } 
      else { 
       isBusy = true; 
       if (innerSubscription) { 
        innerSubscription.unsubscribe(); 
       } 
       innerSubscription = newInner.subscribe(onNext, onError, innerOnComplete); 
      } 
     }, onError, function() { 
      isStopped = true; 
      if (!isBusy) { 
       observer.complete(); 
      } 
     }); 

     return subscriptions; 
    }); 
}; 

そして、ここで更新plunkrです：私はRxJS5で動作するようにそれを更新で刺しを取った私は、最新バージョンにlodashのバージョンを更新https://plnkr.co/edit/DSVmSPRijJwj9msefjRi?p=preview

注意。 lodash 4.7では、いくつかのエッジケースのバグを修正するためにスロットル/デバウンス演算子を書き直しました。 4.6.1を使用していましたが、テストに影響しているとは思われませんでしたが、これらのバグのいくつかが残っていました。