none
Is it safe to subscribe to events using the same handler for multiple sources?

    Question

  • I'd like to be able to observer multiple FileSystemWatcher's and the solution I came up with, although it appears to work correctly on my machine, something tells me it's not quite right.

    Is it safe to add event handlers using ForEach() ?

    List<FileSystemWatcher> _fileSystemWatchers = new List<FileSystemWatcher>();
    
     var fileSystemEventStream =
                    
                    Observable.FromEventPattern<FileSystemEventHandler, FileSystemEventArgs>
                        (
                            _ => _fileSystemWatchers.ForEach(watcher => watcher.Changed += _),
                            _ => _fileSystemWatchers.ForEach(watcher => watcher.Changed -= _)
                        )
                        .ObserveOn(ThreadPoolScheduler.Instance)
                        .SubscribeOn(ThreadPoolScheduler.Instance)                  
                        .GroupBy(ep => ep.EventArgs.FullPath, ep => ep.EventArgs.FullPath)
                        ;


    The universe is mostly hydrogen and ignorance.

    Saturday, August 03, 2013 3:21 AM

Answers

  • Hi, 

    Typically, it's a bad idea.  I assume that multiple FileSystemWatcher objects may raise events concurrently.  That behavior conflicts with Rx's serialized notifications contract (See §§4.2, 6.7; Rx Design Guidelines).

    To solve this problem, it seems that ObserveOn may actually work due to its internal concurrent queueing mechanism; however, it uses a lot of lock-free code and although I'm no expert it does seem like there may be some holes.  But I'd defer to the lock-free experts on that.

    Regardless, it's introducing additional concurrency that seems unnecessary anyway.  Perhaps it's better to use the Synchronize method instead.

    Furthermore, I see no need to use the SubscribeOn method at all.

    Observable.FromEventPattern<FileSystemEventHandler, FileSystemEventArgs>
    (
    	_ => _fileSystemWatchers.ForEach(watcher => watcher.Changed += _),
    	_ => _fileSystemWatchers.ForEach(watcher => watcher.Changed -= _)
    )
    .Synchronize()
    .GroupBy(ep => ep.EventArgs.FullPath, ep => ep.EventArgs.FullPath)
    ...;

    Though maybe it's best to just stick to a compositional approach instead.  This would be my preference:

    _fileSystemWatchers
    	.Select(watcher => Observable.FromEventPattern<FileSystemEventHandler, FileSystemEventArgs>(
    		eh => watcher.Changed += eh,
    		eh => watcher.Changed -= eh))
    	.Merge();
    	.GroupBy(ep => ep.EventArgs.FullPath, ep => ep.EventArgs.FullPath)
    	...;

    - Dave


    http://davesexton.com/blog

    • Marked as answer by Vince Panuccio Sunday, August 04, 2013 9:39 AM
    • Edited by Dave Sexton Sunday, August 04, 2013 1:49 PM Removed invalid nesting of lambdas in second example
    Saturday, August 03, 2013 7:52 AM

All replies

  • Hi, 

    Typically, it's a bad idea.  I assume that multiple FileSystemWatcher objects may raise events concurrently.  That behavior conflicts with Rx's serialized notifications contract (See §§4.2, 6.7; Rx Design Guidelines).

    To solve this problem, it seems that ObserveOn may actually work due to its internal concurrent queueing mechanism; however, it uses a lot of lock-free code and although I'm no expert it does seem like there may be some holes.  But I'd defer to the lock-free experts on that.

    Regardless, it's introducing additional concurrency that seems unnecessary anyway.  Perhaps it's better to use the Synchronize method instead.

    Furthermore, I see no need to use the SubscribeOn method at all.

    Observable.FromEventPattern<FileSystemEventHandler, FileSystemEventArgs>
    (
    	_ => _fileSystemWatchers.ForEach(watcher => watcher.Changed += _),
    	_ => _fileSystemWatchers.ForEach(watcher => watcher.Changed -= _)
    )
    .Synchronize()
    .GroupBy(ep => ep.EventArgs.FullPath, ep => ep.EventArgs.FullPath)
    ...;

    Though maybe it's best to just stick to a compositional approach instead.  This would be my preference:

    _fileSystemWatchers
    	.Select(watcher => Observable.FromEventPattern<FileSystemEventHandler, FileSystemEventArgs>(
    		eh => watcher.Changed += eh,
    		eh => watcher.Changed -= eh))
    	.Merge();
    	.GroupBy(ep => ep.EventArgs.FullPath, ep => ep.EventArgs.FullPath)
    	...;

    - Dave


    http://davesexton.com/blog

    • Marked as answer by Vince Panuccio Sunday, August 04, 2013 9:39 AM
    • Edited by Dave Sexton Sunday, August 04, 2013 1:49 PM Removed invalid nesting of lambdas in second example
    Saturday, August 03, 2013 7:52 AM
  • Thanks for your prompt response.

    The universe is mostly hydrogen and ignorance.

    Sunday, August 04, 2013 9:39 AM