Reactive Extensions RX는 최신 값을 얻기 위해 여러 관찰 시간을 샘플로 처리합니다.

Question

제어 관측 값에 의해 정의 된 시간에 여러 관측 값을 샘플링해야하는 상황이 있습니다.

나는 대리석 다이어그램으로 바로 뛰어 올 것이다. 아래

공동 제어용 관찰 (Observe.Interval (TimeSpan.FromSeconds 1)

오 * 다른 종류의 관찰 가능한이

co tick ---------x---------x---------x---------x---------x---------x- 
o1 int --2----1-------4--------3--------1-------2----3-------5------ 
o2 bool -t-----------f---------t---------------------------f--------- 

result ---------Z---------Z---------Z---------Z---------Z---------Z- 

I 그 샘플링 확장 방법을 개발할 필요가이다 각각의 o 관측 가능한 최신 값 제어 채널에 진드기가있는 경우

Z는 o 관측 값의 최신 샘플 값을 selec으로 전달한 결과입니다 토르 기능. 그것은 CombineLatest와 비슷하지만 꽤 아닙니다.

result ---------1---------0---------3---------1---------3---------0- 

먼저 1 O2 때문이다로 나는이 경우에 결과를 싶습니다

(i, b) => {if (b) return i; else return 0; } 

:

즉, 아주 간단한 예를 들어, FUNC이 같다고 말할 수 마지막으로 참이었고 o1은 마지막 1이었습니다. 두 번째 0은 o2가 마지막으로 거짓 이었기 때문입니다.

o2는 항상 모든 샘플 사이에 하나의 값을 생성하지는 않습니다. 나는 아직도 마지막 가치를 움켜 잡을 필요가있다. (. Sample()는 작동하지 않습니다.). 사실, 관련된 함수와 타입은 더 복잡하기 때문에 위의 int와 bool 타입 때문에 가정을하지 마십시오.

또한 은이 필요합니다. 선택기 기능은 틱마다 한 번만 실행해야합니다.

이것은 현재 해결책이지만 CombineLatest 이후 샘플을 샘플링 한 이후 위의 요구 사항을 준수하지 않습니다.

var combined = interval.CombineSampled(
      Status, 
      H1, 
      H2 
      M, 
      T, 
      HMode, 
      TMode, 
      (i, s, hr1, hr2, m, t, hMode, tMode) => 
      { 
       ... (left out) 

       return result; 
      }); 

는 CombineSampled :

public static IObservable<TResult> CombineSampled<TSource1, TSource2, TSource3, TSource4, TSource5, TSource6, TSource7, TSource8, TResult>(this IObservable<TSource1> controllingSource, IObservable<TSource2> source2, IObservable<TSource3> source3, IObservable<TSource4> source4, IObservable<TSource5> source5, IObservable<TSource6> source6, IObservable<TSource7> source7, IObservable<TSource8> source8, Func<TSource1, TSource2, TSource3, TSource4, TSource5, TSource6, TSource7, TSource8, TResult> selector) 
    { 
     return controllingSource.Publish(s => s.CombineLatest(source2, source3, source4, source5, source6, source7, source8, selector).SampleEx(s)); 
    } 


public static IObservable<T> SampleEx<T, S>(this IObservable<T> source, IObservable<S> samples) 
    { 
     // This is different from the Rx version in that source elements will be repeated, and that 
     // we complete when either sequence ends. 
     return Observable.Create((IObserver<T> obs) => 
      { 
       object gate = new object(); 
       bool hasSource = false; 
       var value = default(T); 

       return new CompositeDisposable(
        source.Synchronize(gate).Subscribe(v => { value = v; hasSource = true; }, obs), 
        samples.Synchronize(gate).Subscribe(_ => { if (hasSource) obs.OnNext(value); }, obs) 
       ); 
      }); 
    } 

Answer 1

변화 당신의 CombineSampled이에 :

public static IObservable<TResult> CombineSampled<TSource1, TSource2, TSource3, TSource4, TSource5, TSource6, TSource7, TSource8, TResult>(this IObservable<TSource1> controllingSource, IObservable<TSource2> source2, IObservable<TSource3> source3, IObservable<TSource4> source4, IObservable<TSource5> source5, IObservable<TSource6> source6, IObservable<TSource7> source7, IObservable<TSource8> source8, Func<TSource1, TSource2, TSource3, TSource4, TSource5, TSource6, TSource7, TSource8, TResult> selector) 
{ 
    return controllingSource.Publish(s => s 
     .CombineLatest(source2, source3, source4, source5, source6, source7, source8, 
      Tuple.Create<TSource1, TSource2, TSource3, TSource4, TSource5, TSource6, TSource7, TSource8>) 
     .SampleEx(s) 
     .Select(t => selector(t.Item1, t.Item2, t.Item3, t.Item4, t.Item5, t.Item6, t.Item7, t.Item8)); 
}