여러 관찰 가능 항목의 순서 동기화

Question

여러 개의 (2 ~ 10) 뜨거운 관찰 가능 항목이 있는데, 각각 주어진 시간 창에서 하나의 이벤트가 발생하지만 순서는 다릅니다.

내가 3 개 과목이 있고, 다음과 같은 임의의 순서로 그들이 불을 설명하기 위해 말할 수 있습니다 :

Subjects: sA, sB, sC 

Order of time window 1: 
    sB, then sA, then sC 
Order of time window 2: 
    sA, then sC, then sB 
Order of time window 3: 
    sA, then sB, then sC 
etc. 

를 그래서 나는 다음과 같은 스트림 얻을 것이다 병합 할 것입니다 경우 : SB, SA 사우스 캐롤라이나, SA를, sC, sB, sA, sB, sC 등

자, 이제 제 질문은 다음과 같습니다. 각 시간 창 내에서 이벤트 순서를 강제하고 싶습니다. 병합 된 스트림 다음 초래

1. sA 
2. sB 
3. sC 

: SA, SB, 사우스 캐롤라이나 SA, SB, 사우스 캐롤라이나 SA 등

SB, 사우스 캐롤라이나 윈도우 시간 자체는 알려져 있지 않지만 각 이후 이벤트는 각 윈도우에서 정확히 한 번 실행됩니다. 모든 피사체가 해고되면 윈도우가 닫히고 새로운 창이 열립니다.

아이디어 우아하게하는 방법?

확장 질문 (중요하지 않음) : 관찰 가능 항목은 서로 독립적이므로 각 관찰 가능 항목의 순서를 동기화 할 수있는 공통 정적 방법을 소개해야합니다.

public static IObservable<T> SynchronizeOrder<T>(IObservable<T> source, int order) 
{ 
    //return synchronized source 
} 

업데이트 : 뭔가 내 원래의 질문에 분명 아니었다 관찰 가능한 다른 이벤트 유형의 수 (나는 내 ​​예제에서 특정 이벤트 데이터 형식을 사용하지 않은 그 이유는) 결과적 수 있었다 나는 명령 된 관찰 대상을 병합하고 싶지 않다. 독립적 인 관측치가 순서대로 동기화되도록 보장 할 수있는 메커니즘이 필요합니다.

myEventStream 
.ObserveOn(TaskFactory) 
.DoSomeExpensiveComputation() 
.Order(2) //doesn't have to be an extension method 
.Subscibe(computationResult=> 
      sharedRessourceWhereOrderMatters.Update(computationResult)) 

Answer 1

이것은 당신이 관찰 사용하여 원하는 것을 조인 :

이

using System.Reactive; 
    using System.Reactive.Subjects; 
    using System.Reactive.Linq; 

    var a = new Subject<int>(); 
    var b = new Subject<int>(); 
    var c = new Subject<int>(); 

    var test = 
     Observable.When(
      a.And(b).And(c).Then((a1, b1, c1) => new int[] { a1, b1, c1 }) 
     ).SelectMany(arr => arr.ToObservable()); 

    var sub = test.Subscribe(val => Console.Write("{0} ", val)); 

    a.OnNext(1); 
    b.OnNext(2); 
    c.OnNext(3); 

    a.OnNext(1); 
    c.OnNext(3); 
    b.OnNext(2); 

    c.OnNext(3); 
    a.OnNext(1); 
    b.OnNext(2); 

    Console.ReadKey(); 
    sub.Dispose(); 

출력은 다음과 같습니다

1 2 3 1 2 3 1 2 3 

주의 그러나 그 :

1. 이 제공하지 않습니다 확장 메소드를 사용하면 관찰 가능한 특정 순서로 주사하십시오. 당신은 And (단편적 일 수있다)을 사용하여 계획을 세우고, 결국 그것을 실행할 필요가있다. '빌드 업'순서가 출력 순서입니다.

2. 실제 창 만들기는 표시되지 않습니다. 그러나 어딘가에서 관찰 가능한 체인에 주입 할 수없는 이유는 없습니다.

3. 이 솔루션은 세 관찰 변수가 모두 값을 가질 때마다 번을 번 실행합니다. 원하는 동작을 얻으려면 설정하고 올바른 지점에서 찢어 야 할 수도 있습니다.

Answer 2

yamens 답은 동일한 범위에있는 동일한 유형의 관찰 가능 항목에 대해 큽니다. 내 경우에는 (확장 질문) 관측 대상이 다른 유형이 될 수 있으며 실행 순서에 관계없이 순서를 동기화 할 수 있습니다.그래서 여기

내 제안 간단한 해결책 :

private readonly object syncRoot = new object(); 
private int sourceCount; 
private readonly SortedDictionary<int, Action> buffer = new SortedDictionary<int, Action>(); 

public IObservable<T> Order<T>(IObservable<T> source, int order) 
{ 
    return Observable.Create<T>(observer => 
    { 
     lock (syncRoot) 
      sourceCount++; 

     source.Synchronize(syncRoot).Subscribe(item => 
     { 
      buffer[order] =() => observer.OnNext(item); 
      if(buffer.Count == sourceCount) 
      { 
       foreach (var action in buffer.Values) 
        action(); 
       buffer.Clear(); 
      } 
     }, observer.OnError, observer.OnCompleted); 

     return() => 
     { 
      lock (syncRoot) 
       sourceCount--; 
     }; 
    }); 
}//end order