Rx : 일정 기간 동안 첫 번째 항목 기다림

Question

레거시 이벤트 기반 방법을 관찰 가능 기반으로 변환하고 싶습니다. 그러나 Rx를 매우 신중하게 사용하고 있으므로 지금 당황합니다.

나는 이벤트 소스를 가지고 있는데, 지금은 관찰 할 수 있습니다. 특정 시점에서 나는 라인의 다음 요소를 리턴함으로써 끝나는 메소드를 시작해야한다.

이벤트 기반의 접근 방식은 다음과 같다 : 당신이 볼 수 있듯이

public async Task<ReaderEvent> WaitForReaderAsync(int PlaceId, TimeSpan waitFor) 
{ 
    ReaderEvent result = null; 
    using (var cts = CancellationTokenSource.CreateLinkedTokenSource(new [] { topLevelToken })) 
    { 
     cts.CancelAfter(waitFor); 

     EventHandler<ReaderEvent> localHandler = (o, e) => 
     { 
      if (e.PlaceId == PlaceId) 
      { 
       result = e; 
       cts.Cancel(); 
      } 
     }; 

     ReaderEventHandler += localHandler; 
     try 
     { 
      await Task.Delay(waitFor, cts.Token).ConfigureAwait(false); 
     } 
     catch (OperationCanceledException) { } 
     catch (Exception ex) 
     { 
      //... 
     } 

     ReaderEventHandler -= localHandler; 
    } 

    return result; 
} 

, 아이디어는 지연이 취소됩니다 내가 대기하고있는 이벤트 또는 토큰 소스의 도착에 의해 중 하나를 취소한다는 것입니다 그 특정 시간이 지나면 구성에 따라. 아주 깨끗합니다. 이제

의 수신 버전 : 나는 너무 제한 시간 연장과 시도

public async Task<ReaderEvent> WaitForReaderAsync(int PlaceId, TimeSpan waitFor) 
{ 
    ReaderEvent result = null; 

    var observable = _OnReaderEvent.FirstAsync(r => r.PlaceId == PlaceId); 

    using (var cts = CancellationTokenSource.CreateLinkedTokenSource(new [] { topLevelToken })) 
    { 
     cts.CancelAfter(waitFor); 
     using (observable.Subscribe(x => { 
      result = x; 
      cts.Cancel(); 
     { 
      try 
      { 
       await Task.Delay(waitFor, cts.Token).ConfigureAwait(false); 
      } 
      catch (OperationCanceledException) { } 
     } 
    } 
    return result; 
} 

너무 깨끗하지 ... 더 악화 ... . 그러나 이것이 원샷 구독이기 때문에 구독을 처리하기 전에 어떻게 든 기다려야합니다. 유일한 차이점은 OnError가 CancelAfter의 기본 제공 메커니즘이 아니라 로컬 토큰을 취소한다는 것입니다.

반죽/더 간결한 (Rx에 더 의존하는) 방법이 있습니까?

감사합니다.

Answer 1

왜 그냥 코드의 간단한 수신 버전으로 이동하지 :

public async Task<ReaderEvent> WaitForReaderAsync(int PlaceId, TimeSpan waitFor) 
{ 
    return await 
     _OnReaderEvent 
      .Where(r => r.PlaceId == PlaceId) 
      .Buffer(waitFor, 1) 
      .Select(xs => xs.FirstOrDefault()) 
      .FirstOrDefaultAsync() 
      .ToTask(); 
} 

Answer 2

당신이 시도 할 수 :

var values = await _OnReaderEvent 
    .Where(r => r.PlaceId == placeId) 
    .Buffer(waitFor, 1) 
    .FirstAsync(); // get list of matching elements during waitFor time 

return values.FirstOrDefault(); // return first element or null if the list is empty