Czy ten kod Reactive Extensions wycieka z pamięci?

Question

Napisałem kod, który zmienia zdarzenie FileSystemWatcher w obserwowalną sekwencję.

Moim celem jest podzielenie wszystkich zmian w systemie plików w celu oddzielenia strumieni i ich dławienia.

Na przykład, jeśli mam 10 różnych plików, które zmieniają się 3 razy w ciągu pół sekundy, otrzymam notyfikację tylko raz dla każdego pliku.

Co mnie jednak niepokoi, to operator GroupBy(). Aby to zadziałało (zakładam), musiałoby z czasem budować grupę i zużywać niewielkie ilości pamięci.

Czy spowoduje to "wyciek", a jeśli tak, w jaki sposób mogę temu zapobiec?

FileSystemWatcher _watcher = new FileSystemWatcher("d:\\") { 
    EnableRaisingEvents = true, 
    NotifyFilter = NotifyFilters.LastWrite | NotifyFilters.Size 
}; 

void Main() 
{ 
    var fileSystemEventStream = 
     Observable.FromEventPattern<FileSystemEventHandler, FileSystemEventArgs> 
      (
       _ => _watcher.Changed += _, 
       _ => _watcher.Changed -= _ 
      ) 
      .ObserveOn(ThreadPoolScheduler.Instance) 
      .SubscribeOn(ThreadPoolScheduler.Instance) 
      .GroupBy(ep => ep.EventArgs.FullPath, ep => ep.EventArgs.FullPath) 
      ; 

    var res = 
     from fileGroup in fileSystemEventStream 
     from file in fileGroup.Throttle(TimeSpan.FromSeconds(1)) 
     select file; 

    res.Subscribe(
     ReceiveFsFullPath, 
     exception => { 
      Console.WriteLine ("Something went wrong - " + exception.Message + " " + exception.StackTrace); 
     }); 

    Console.Read(); 
} 

void ReceiveFsFullPath(string s){ 
    Console.WriteLine ("Received file system event on thread " + Thread.CurrentThread.ManagedThreadId); 
    Console.WriteLine(s); 
} 

Answer 1

Tak, dla każdego nowego klucza GroupBy tworzy temat i utrzymuje słownik tych tematów. I subskrybujesz każdy z nich. Jest to więc niewielka porcja pamięci, która z upływem czasu będzie rosnąć, bez uprzedniego wypuszczenia starych wpisów. To, czego naprawdę potrzebujesz, to usunięcie klucza po upływie czasu zegara przepustnicy. Nie mogę wymyślić, jak to zrobić z wbudowanymi operatorami. Potrzebujesz więc niestandardowego operatora. Oto ukłucie w jednym.

public IObservable<T> ThrottleDistinct<T>(this IObservable<T> source, TimeSpan delay) 
{ 
    return Observable.Create(observer => 
    { 
     var notifications = new Subject<IObservable<T>>(); 
     var subscription = notifications.Merge().Subscribe(observer); 
     var d = new Dictionary<T, IObserver<T>>(); 
     var gate = new object(); 
     var sourceSubscription = new SingleAssignmentDisposable(); 
     var subscriptions = new CompositeDisposable(subscription, sourceSubscription); 
     sourceSubscription.Disposable = source.Subscribe(value => 
     { 
      IObserver<T> entry; 
      lock(gate) 
      { 
      if (d.TryGetValue(value, out entry)) 
      { 
       entry.OnNext(value); 
      } 
      else 
      { 
       var s = new Subject<T>(); 
       var o = s.Throttle(delay).FirstAsync().Do(() => 
       { 
       lock(gate) 
       { 
        d.Remove(value); 
       } 
       }); 
       notifications.OnNext(o); 
       d.Add(value, s); 
       s.OnNext(value); 
      } 
      } 
     }, observer.OnError, notifications.OnCompleted); 

     return subscriptions; 
    }); 
} 

... 
Observable.FromEventPattern(...) 
    .Select(e => e.EventArgs.FullPath) 
    .ThrottleDistinct(TimeSpan.FromSeconds(1)) 
    .Subscribe(...); 

Answer 2

Zgodnie z odpowiedzią Brandona tematy będą rosły i nie będą mogły zostać odzyskane *. Moją główną obawą związaną z przeciekającą pamięcią jest to, że nie przechwytujesz subskrypcji! tj

res.Subscribe(... 

należy zastąpić

subscription = res.Subscribe(... 

jeśli nie uchwycić subskrypcję, nigdy nie można wyrzucać z abonamentu, a tym samym nie puścisz obsługi zdarzeń, co masz „wyciekły pamięć ". Oczywiście to nie ma sensu, jeśli w rzeczywistości nie dysponujesz abonamentem.

* Cóż, jeśli skończą, zostaną automatycznie unieszkodliwione, tak by działały. Po zakończeniu zdarzenia FileDeleted możesz wykonać sekwencję?