例外のRxによって提起された例外の底に取得しようとしてRxの

Question

に参加を使用しているとき、私は次のようREPROを思い付いた：指定された行で

[Fact] 
public void REPRO() 
{ 
    // when true, window is open to accept requests 
    var isWindowOpen = new BehaviorSubject<bool>(true); 

    // when this ticks and the window is open, we want to execute the logic 
    var request = new Subject<Unit>(); 

    // determines when the execution window opens 
    var openWindow = isWindowOpen 
     .Where(ce => ce); 

    // determines when the execution window closes 
    var closeWindow = isWindowOpen 
     .Where(ce => !ce); 

    var executionCount = 0; 

    var result = Observable 
     .Join(
      openWindow, 
      request, 
      _ => closeWindow, 
      _ => Observable.Empty<Unit>(), 
      (l, r) => Unit.Default) 
     .Do(_ => ++executionCount) 
     // changing isWindowOpen from false to true here is causing Rx to throw an exception 
     .SelectMany(_ => Observable.Return(Unit.Default).Do(__ => isWindowOpen.OnNext(false)).Do(__ => isWindowOpen.OnNext(true))) 
     .Subscribe(); 

    // this line causes Rx to throw an exception: System.InvalidOperationException: Collection was modified; enumeration operation may not execute. 
    request.OnNext(Unit.Default); 

    Assert.Equal(1, executionCount); 
} 

、次の例外がスローされます。

System.InvalidOperationException: Collection was modified; enumeration operation may not execute. 
    at System.ThrowHelper.ThrowInvalidOperationException(ExceptionResource resource) 
    at System.Collections.Generic.Dictionary`2.ValueCollection.Enumerator.MoveNext() 
    at System.Reactive.Linq.ObservableImpl.Join`5._.RightObserver.OnNext(TRight value) 
    at System.Reactive.Subjects.Subject`1.OnNext(T value) 
    at RxJoinTest.Class1.InvokeCommandInvokesTheCommand() in c:\users\kent\documents\visual studio 2015\Projects\RxJoinTest\RxJoinTest\Class1.cs:line 57 

お知らせ例外を説明するコメント行は、私が参加した項目のためのパイプライン中にisWindowOpen、観察を変更する場合にのみ発生します。

誰でもこの（一見バグのある）動作を説明できますか？あるいは、誰もがこの問題を回避しながら私の目標を達成するための別の手段を提供することができますか？

私の実際のシナリオはこれです：ウィンドウの開閉は、コマンドが利用可能になり、実行に使用できなくなります。要求はコマンドを実行したいという欲求を表し、実行するロジック（SelectManyの内部）はコマンドロジック自体です。内部ロジックの一環として、コマンドは自身を無効にしてから再び有効にします。つまり、isWindowOpenが実行中にどのように変更されるかです。

Answer 1

あなたJoinコードでSelectManyは少し単純である、これまで低減することができます。

var result = Observable.Join(
     openWindow, 
     request, 
     _ => closeWindow, 
     _ => Observable.Empty<Unit>(), 
     (l, r) => Unit.Default 
    ) 
    .Do(_ => ++executionCount) 
    // The two lines from the SelectMany. You don't even need both; either one of these lines triggers the error 
    .Do(__ => isWindowOpen.OnNext(false)) 
    .Do(__ => isWindowOpen.OnNext(true)) 
    .Subscribe(); 

短い答えは右の窓の開口部は、左のウィンドウで変更をトリガすることができないということで、左の窓の開口部は、右の窓の変化を引き起こすことができない。

内部的には、左のウィンドウが開くたびに、Rxは開いているすべての右ウィンドウを反復処理し、結合セレクタを実行します。 .DoはJoinセレクタの直後に同期的に実行されるため、RightWindowIteratorがまだ開いている間に発生し、コレクションRightWindowIteratorの変更がトリガされてエラーが発生します。これを確認するには、source codeをご覧ください。

この制限は意味があります。 RightWindowClose関数をObservable.Empty<Unit>()からObservable.Never<Unit>()に変更し、Rxにこの制限がない場合は、無限ループになります。

あなたの目標を達成するのに役立つ限り、あなたは本当にそれらを記述していません。

var result = openWindow.Join(
     request, 
     _ => closeWindow, 
     _ => Observable.Empty<Unit>(), 
     (l, r) => Unit.Default 
    ) 
    .Do(_ => ++executionCount) 
    .Delay(TimeSpan.FromMilliseconds(1)) 
    .Do(__ => isWindowOpen.OnNext(false)) 
    .Do(__ => isWindowOpen.OnNext(true)) 
    .Subscribe(); 

...しかし、それは信頼性の高い堅牢なコードにはほど遠いです：.Do年代の前に遅延を追加すると、コードの実行を作ることになる、イテレータはオープン開催されていないので、彼らは、非同期で実行するようになります。できるだけ科目を取り除くことをお勧めします。特に、参加者の周りの循環論理を取り除くことをお勧めします。

---

EDIT：

あなたは、おそらく次のように所望の効果を達成することができます

var result = request 
    .WithLatestFrom(isWindowOpen, (_, windowOpen) => windowOpen) 
    .Where(windowOpen => windowOpen) 
    .Do(_ => ++executionCount) 
    // changing isWindowOpen from false to true here is causing Rx to throw an exception 
    .SelectMany(_ => Observable.Return(Unit.Default).Do(__ => isWindowOpen.OnNext(false)).Do(__ => isWindowOpen.OnNext(true))) 
    .Subscribe();