RxJS Drosselverhalten; erhalten sofort erste Wert

Question

Beispiel Plunkr: https://plnkr.co/edit/NZwb3ol8CbZFtSc6Q9zm?p=preview

ich weiß, dass diese drei Drossel Methoden für rxjs sind (5,0 beta.4):

auditTime(), throttleTime() und debounceTime()

Das Verhalten Ich bin Suche nach ist der eine lodash standardmäßig auf Gas:

• 1) Gib mir den ersten Wert sofort!
• 2) an aufeinander folgenden Werten, halten Werte für gegebene Verzögerung, dann emittieren zuletzt aufgetretenen Wert
• 3), wenn Drosselklappenverzögerungs abgelaufen ist, in den Zustand zurück (1)

Theoretisch sollte dies wie folgt aussehen:

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

Aber

• throttleTime nie gibt mir den letzten Wert, wenn in der emittierten Drossel Timeout
• debounceTime löst nicht sofort
• auditTime nicht sofort

auslöst Kann ich eine der rxjs Methoden kombinieren, um das beschriebene Verhalten zu erreichen?

Answer 1

Ich nahm den AuditTime-Operator und änderte 2 Zeilen, um das gewünschte Verhalten zu erreichen.

New Plunker: https://plnkr.co/edit/4NkXsOeJOSrLUP9WEtp0?p=preview

Original:

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

Änderungen:

aus (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); 
    } 
} 

bis (auditTimeImmediate):

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

So beginne ich die Zeitspanne, nach dem Wert next ed war.

Verbrauch:

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

Answer 2

Für ältere RxJs schrieb ich einen concatLatest Operator, der das meiste von dem macht, was Sie wollen. Mit diesem Code können Sie Ihr Drosselungsverhalten mit diesem Code erhalten:

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

Hier ist der Operator. Ich nahm einen Stich es die Aktualisierung mit RxJS5 zu arbeiten:

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

Und hier ist ein aktualisierter plunkr: https://plnkr.co/edit/DSVmSPRijJwj9msefjRi?p=preview

Hinweis I Ihre lodash Version auf die neueste Version aktualisiert. In lodash 4.7 habe ich die Gas-/Debounce-Operatoren neu geschrieben, um einige Edge-Case-Bugs zu beheben. Sie haben 4.6.1 verwendet, das noch einige dieser Fehler enthielt, obwohl ich glaube, dass sie Ihren Test nicht beeinflusst haben.