Rx ve görevler - yeni görev oluşturulduğunda çalışma görevini iptal eder misiniz?

Question

Kullanıcı etkileşimi senaryolarım var Rx ile kullanmak istiyorum.

• : ama ben de gerek - "kullanıcı bazı işler yapmak, yazmayı bırakır zaman"

  senaryo (1) (genellikle, kullanıcının o ana kadar yazılan ne aramak) kanonik benzer (2) sadece

(3) yeni bir iş birimi başladığında, devam eden çalışmaları iptal (benim durumumda bu işlemci
• var (aşağıya bakınız) "do bazı iş" birimlerinin sonuçlarının son olsun yoğun)

(1) Kullanıcı olayları için IObservable kullanıyorum, sadece olaylar arasındaki duraklamaları tetiklemek için .Throttle() ile kısıtlıyorum ("kullanıcı yazmayı durdur").

Bundan i .Select(_ => CreateMyTask(...).ToObservable()).

Bu, iç gözlenebilirlerin her birinin tek bir görevi tamamladığı bir IObservable<IObservable<T>> verir.

Almak için (2) Sonunda sadece en yeni iş biriminden sonuçlar almak için .Switch()'u uygularım.

Peki ya (3) - bekleyen görevleri iptal et? Dispose() onları neden öncekine (ler) den

yeni iç IObservable<T> geldiği zaman Eğer doğru anlamak

, .Switch() yöntem ona abone ve abonelikten.
Belki de bir şey iptal etme görevini tetiklemek için kablolanabilir?

Answer 1

Görevler ile çalışmak zorunda mısınız?

Sadece Gözlemcilerle çalışmaktan memnunsanız, bunu kendiniz yapabilirsiniz.

böyle bir şey yaptığını Dene:

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); 
    }); 

Bu gözlemlenebilir Thread.Sleep(10); ile döngü bazı zor işi yapıyor, ancak gözlemlenebilir tanzim edildiğinde döngüden çıkılır ve yoğun CPU iş olmaktan çıkar. Ardından, devam eden çalışmayı iptal etmek için standart Rx Dispose'u Switch ile birlikte kullanabilirsiniz.

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); 
    }); 
} 

Ve sonra böyle bir işlevle diyoruz: Eğer bir yöntemde bohçalanmış o istiyorsanız

, o zaman bu deneyin

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; 
}; 

İşte bu kanıtladı kod bu işlendi: çıktı olarak "50'de İptal Edildi" (bazen "51'de İptal Edildi") özelliğini aldım.

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

Bu işin her birim için yeni bir kod oluşturmanız ve yenisine her seferinde o Switch anahtarları iptal edecektir:

Answer 2

sadece gözlemci unsubcribes zaman iptal edilir belirteçleri üretecektir hangi Observable.FromAsync kullanabilirsiniz .