RxJs5をAngular2で使用しているRESTクライアントのタイムベースキャッシュ

Question

ReactiveX/RxJsを新しく使いました。私の使用事例がRxJでスムーズに実行できるかどうか、できれば組み込み演算子を組み合わせて使用​​するのがよいでしょうか。ここに私が達成したいものがあります：

私は、REST APIと通信するAngular2アプリケーションを持っています。アプリケーションのさまざまな部分が異なる時間に同じ情報にアクセスする必要があります。同じリクエストを何度も繰り返してサーバーを叩くのを避けるために、私はクライアント側のキャッシュを追加したいと思います。キャッシングは、ネットワークコールが実際に行われるサービスレイヤーで行われます。このサービス層はObservableを手渡すだけです。キャッシングは、アプリケーションの残りの部分に対して透過的でなければなりません。キャッシングではなく、Observableを認識する必要があります。

最初に、60秒以内にサービスからこの情報を要求するコンポーネントが12個あっても、REST APIの特定の情報は60秒間に1回しか取得されません。各サブスクライバはサブスクリプション時にObservableから（単一の）最後の値を与えられなければなりません。

現在、私はこのようなアプローチでまさにそれを達成するために管理：

この例では

public getInformation(): Observable<Information> { 
    if (!this.information) { 
    this.information = this.restService.get('/information/') 
     .cache(1, 60000); 
    } 
    return this.information; 
} 

は、restService.get(...)は、実際のネットワーク呼び出しを実行し、多くの角度のHTTPサービスのように、Observableを返します。

このアプローチの問題点は、キャッシュをリフレッシュすることです。ネットワークコールが1回だけ実行され、キャッシュされた値が60秒後に新しいサブスクライバにプッシュされなくなるまで、キャッシュは再実行されませんキャッシュが期限切れになった後の最初の要求したがって、60秒のキャッシュの後に発生するサブスクリプションには、Observableからの値は与えられません。

キャッシュがタイムアウトした後に新しいサブスクリプションが発生し、新しい値を60秒間再度キャッシュすると、最初のリクエストを再実行できますか？

ボーナスとして：既存のサブスクリプション（たとえば、最初のネットワークコールを開始した人）が、新しいサブスクリプションによってフェッチされたリフレッシュされた値を取得すると、情報がリフレッシュされるようにObservable-awareアプリケーション全体にすぐに渡されます。

Answer 1

私が探していたものを正確に達成するための解決策を見つけました。それはReactiveX命名法とベストプラクティスに反するかもしれませんが、技術的には、私が望むものとまったく同じです。つまり、組み込み演算子でも同じことを達成する方法が見つかった場合、より良い答えを受け入れることができます。

基本的に、サブスクリプション（ポーリング、タイマーなし）でネットワークコールを再トリガする方法が必要なので、ReplaySubjectがどのように実装されているかを見て、それをベースクラスとして使用しました。私はその後、コールバックベースのクラスRefreshingReplaySubjectを作成しました（名前の改善が歓迎です！）。ここでは、次のとおりです。

export class RefreshingReplaySubject<T> extends ReplaySubject<T> { 

    private providerCallback:() => Observable<T>; 
    private lastProviderTrigger: number; 
    private windowTime; 

    constructor(providerCallback:() => Observable<T>, windowTime?: number) { 
    // Cache exactly 1 item forever in the ReplaySubject 
    super(1); 
    this.windowTime = windowTime || 60000; 
    this.lastProviderTrigger = 0; 
    this.providerCallback = providerCallback; 
    } 

    protected _subscribe(subscriber: Subscriber<T>): Subscription { 
    // Hook into the subscribe method to trigger refreshing 
    this._triggerProviderIfRequired(); 
    return super._subscribe(subscriber); 
    } 

    protected _triggerProviderIfRequired() { 
    let now = this._getNow(); 
    if ((now - this.lastProviderTrigger) > this.windowTime) { 
     // Data considered stale, provider triggering required... 
     this.lastProviderTrigger = now; 
     this.providerCallback().first().subscribe((t: T) => this.next(t)); 
    } 
    } 
} 

そしてここでは、得られた用法である：あなたの答え、ボグダンため

public getInformation(): Observable<Information> { 
    if (!this.information) { 
    this.information = new RefreshingReplaySubject(
    () => this.restService.get('/information/'), 
     60000 
    ); 
    } 
    return this.information; 
} 

Answer 2

これを実装するには、あなたがsubscribtion上でカスタム・ロジックを使用して独自に観察を作成する必要があります。

function createTimedCache(doRequest, expireTime) { 
    let lastCallTime = 0; 
    let lastResult = null; 

    const result$ = new Rx.Subject(); 

    return Rx.Observable.create(observer => { 
     const time = Date.now(); 
     if (time - lastCallTime < expireTime) { 
      return (lastResult 
       // when result already received 
       ? result$.startWith(lastResult) 
       // still waiting for result 
       : result$ 
      ).subscribe(observer); 
     } 
     const disposable = result$.subscribe(observer); 
     lastCallTime = time; 
     lastResult = null; 
     doRequest() 
      .do(result => { 
       lastResult = result; 
      }) 
      .subscribe(v => result$.next(v), e => result$.error(e)); 
     return disposable; 
    }); 
} 

と使用方法を結果としては、次のようになります。

this.information = createTimedCache(
() => this.restService.get('/information/'), 
    60000 
); 

使用例：https://jsbin.com/hutikesoqa/edit?js,console