最初の値にrxjsタイムアウト

Question

私が理解したからthis questionここで私は理解していますtimeout観察可能な時間の任意の値を放出しない場合演算子エラー...私の問題は、この時間のウィンドウがリセットされますそれぞれが放射した後に、最初の値がウィンドウ内に出てくる場合にのみ興味がある場合は、シーケンスを完了する必要があります...

「最初のタイムアウト」を持つ良い方法はありますか？ .take(1).timeout(1000)以外？

Answer 1

あり最初までしか私は承知しているし、ここでそれを行うための方法だとして、のためのタイムアウトません：
（Plunkr利用可能here）

const { Observable } = Rx; 

// this function can be use with the let operator 
// ex: someObs$.let(timeoutFirstOnly(100)).subscribe(...) 
// it'll emit an error if the first value arrives later 
// than timeout parameter value 
const timeoutFirstOnly = timeout => observable$ => { 
    const obs$ = observable$.share(); 

    const [firstValue$, others$] = obs$.partition((value, index) => index === 0); 

    return Observable.merge(firstValue$.timeout(timeout).first(), others$); 
}; 

const created = Rx.Observable.create(observer => { 
    // here, this is working 
    // try to emit at 150ms for example and it'll throw an error 
    setTimeout(() => observer.next(42), 150); 
}); 

created 
    .let(timeoutFirstOnly(100)) 
    .subscribe(
    val => console.log(`received ${val}`), 
    error => console.log(`threw ${error}`) 
); 

は、しかし、私はので、私別の方法があるかもしれないと感じましたGitterに尋ねたと@Dorusは私に良い結果が得られた：ここ

created.publish(src => 
    Rx.Observable.merge(
    src.take(1).timeout(100), 
    src.skip(1) 
) 
); 

説明を、すべてのクレジットは、このいずれかに@Dorusに行く：

publishは2つの味を持っています

1. .publish() : ConnectableObservable<T>
2. .publish(selector : (src : Observable<T>) => Observable<R>): Observable<R>

.publish()はConnectableObservableを返して、その上にまたはそれに自分自身をconnectを呼び出すことによってrefCountを呼び出すことによっても使用することができます。 .publish(selector)セレクタでマルチキャストされたソースを使用させ、結果を出力します。 ところで、.publish()は.multicast(() => new Subject())と同じで、.publish(selector)は.multicast(() => new Subject(), selector)と同じです。

.publish().refCount()も.share()とほぼ同じです。 partitionの前にshareに電話するのは間違いありません。それは問題を部分的に解決します。私はまだ.publish(selector)を好んでいます。.publish(selector)は、ソースオブザーバブルにサブスクライブする前に、セレクタとすべてのソースにsubscribeを実行します。共有すると、最初のサブスクリプションがソースをアクティブにし、何かの同期が2回目のサブスクリプションで失われます。これは、合計同期ソースを使用する場合と同じくらい悪くなり、2回目のサブスクリプションがソースを完全に再実行します。

最後の追加：.publish(selector)は.let(selector)と同じシグネチャを持ちますが、letは送信元をマルチキャストしません。それは違いです。

Answer 2

@ Maximeの回答に加えて、raceを使用して、観測値の最初の値とタイムアウトの間のレースを作成することができます。 neverとtimeoutを組み合わせてタイムアウトを作成します。

このようにして、ソースオブザーバブルと、決して値を出力しないオブザーバブルとの間のレースが終了しますが、時間が経過するとエラーが発生します。ソースオブザーバブルが最初の値を生成した場合、raceはタイムアウトの監視を停止します。

const timed = source.race(Observable.never().timeout(1000));