Rx 및 작업 - 새 작업이 생성 될 때 실행중인 작업을 취소 하시겠습니까?

Question

Rx로 처리하고 싶은 사용자 상호 작용 시나리오가 있습니다.

• :하지만 나는 또한 필요 - "사용자가 어떤 작업을 수행, 입력을 멈춘 때"

  시나리오 (1) (일반적으로, 사용자가 지금까지 입력 한 것을 검색) 정식 유사하다 (2) 만

(3) 새로운 작업 단위를 시작할 때, 이가 진행중인 작업을 취소 (내 경우는 CPU
• 의 (아래 참조) "할 몇 가지 작업"단위의 결과의 최신를 얻을 수 집중 형)

(1) 사용자 이벤트에 IObservable을 사용하고 이벤트 사이의 일시 중지시에만 트리거하는 .Throttle()으로 조절합니다 ("사용자 입력 중지").

여기에서 i .Select(_ => CreateMyTask(...).ToObservable()).

이것은 내측 관찰 대상 각각이 단일 작업을 래핑하는 IObservable<IObservable<T>>을 제공합니다.

(2) 마지막으로 .Switch()을 적용하여 최신 작업 단위에서만 결과를 얻습니다.

보류중인 작업 취소 (3)은 어떻게됩니까? Dispose()에 원인 이전 (들)에서

새로운 내부 IObservable<T>있을 때마다 만약 내가 제대로 이해하고

,의 .Switch() 방법은 구독 및 구독 취소.
어쩌면 취소 할 작업을 실행하는 데 유선이 될 수 있습니까?

Answer 1

작업을 수행해야합니까?

당신이 Observables와 함께 일하게되어 기쁘다면, 당신은 자신을 멋지게 할 수 있습니다.

같은 일을보십시오 :

var query = 
    Observable.Create<int>(o => 
    { 
     var cancelling = false; 
     var cancel = Disposable.Create(() => 
     { 
      cancelling = true; 
     }); 
     var subscription = Observable.Start(() => 
     { 
      for (var i = 0; i < 100; i++) 
      { 
       Thread.Sleep(10); //1000 ms in total 
       if (cancelling) 
       { 
        Console.WriteLine("Cancelled on {0}", i); 
        return -1; 
       } 
      } 
      Console.WriteLine("Done"); 
      return 42; 
     }).Subscribe(o); 
     return new CompositeDisposable(cancel, subscription); 
    }); 

이 관찰은 Thread.Sleep(10);와 루프에 대한 몇 가지 노력을하고 있지만, 관찰이 배치 될 때 루프가 종료되고 집중적 인 CPU 작업이 중단. 그런 다음 Switch과 함께 표준 Rx Dispose을 사용하여 진행중인 작업을 취소 할 수 있습니다.

public static IObservable<T> Start<T>(Func<Func<bool>, T> work) 
{ 
    return Observable.Create<T>(o => 
    { 
     var cancelling = false; 
     var cancel = Disposable 
      .Create(() => cancelling = true); 
     var subscription = Observable 
      .Start(() => work(() => cancelling)) 
      .Subscribe(o); 
     return new CompositeDisposable(cancel, subscription); 
    }); 
} 

을 그리고 다음과 같은 기능을 호출 : 당신이 방법에 번들 것을 싶은 경우

,이 시도

Func<Func<bool>, int> work = cancelling => 
{ 
    for (var i = 0; i < 100; i++) 
    { 
     Thread.Sleep(10); //1000 ms in total 
     if (cancelling()) 
     { 
      Console.WriteLine("Cancelled on {0}", i); 
      return -1; 
     } 
    } 
    Console.WriteLine("Done"); 
    return 42; 
}; 

여기이 증명 내 코드입니다 근무 :

var disposable = 
    ObservableEx 
     .Start(work) 
     .Subscribe(x => Console.WriteLine(x)); 

Thread.Sleep(500); 
disposable.Dispose(); 

출력물에 "Cancelled on 50"(언젠가는 "Cancelled on 51")이 출력됩니다.

input.Throttle(...) 
    .Select(_ => Observable.FromAsync(token => CreateMyTask(..., token))) 
    .Switch() 
    .Subscribe(...); 

이 각 작업 단위에 대한 새로운 토큰을 생성하고 새로운 하나마다 그것을 Switch 스위치를 취소합니다

Answer 2

당신은 관찰자 unsubcribes 때 취소 토큰을 생성하는 Observable.FromAsync을 사용할 수 있습니다 .