1

The behaviours is really strange, it looks like a massive bug in the framework itself,

I have a FileSystemWatcher to detect new files in a folder, and everytime a new file is detected, that file is parsed and deleted before an Rx .OnNext notification is sent:

private Subject<MyObject> objectNotification = new Subject<MyObject>();
private FileSystemWatcher watcher;

private void MyClassConstructor(string pathToWatch)
{
    watcher= new FileSystemWatcher();
    watcher.Filter = scraper.Ext;
    watcher.Created += new FileSystemEventHandler(parseMethod);
    watcher.Path = pathToWatch;
    watcher.EnableRaisingEvents = true;
}

private void parseMethod(object sender, FileSystemEventArgs e)
{
    MyObject parsedFile = new MyObject(e.FullPath);
    File.Delete(e.FullPath);
    var syncedSubject = Subject.Synchronize(objectNotification);
    syncedSubject.OnNext(parsedFile);
}

On the other side, somewhere in the GUI code of the main application there's a receiving code:

private IDisposable myObjectSubscription;
.
.
.
private void initSubscription()
{
    this.myObjectSubscription = sendingClass.Subscribe(parsedObject =>
    {
      this.AddObject(parsedObject);
    });
}

private void addObject(MyObject parsedObject)
{
    //Sometimes the same object is being received many times, but has been sent only once!
    //BUG!
}

ideally I should receive a notification within the observer for each value sent by the sender class, unfortunately I get more, a lot more!

3
  • Are you sure it's not FileSystemWatcher that is notifying you multiple times? Jan 9, 2014 at 8:55
  • This shouldn't be possible "in theory" since we are talking about ~200bytes file... but if you make me think the issue appears when watching files from a Samba network share, connected via WiFi, in that case it could be that the FileSystemWatcher is behaving like watching big files needing to be read/written in more than one chunk of bytes and raising an event every chunk? could it be? (it's anyway pretty hard to reproduce)
    – Aytharn
    Jan 12, 2014 at 9:02
  • Please see stackoverflow.com/questions/172060/… Jan 12, 2014 at 12:42

1 Answer 1

2

Observable synchronization is the issue at the heart of this question.

Subject or Observable synchronization methods exist to prevent concurrent OnXXX calls being made.

Because synchronization introduces overhead when it is not necessary (which is often enough) and because of the alternative approaches described below, it is a requirement of the user of the Rx API to decide when and how to synchronize.

In your case every time an event is raised and your event handler is called, parseMethod creates a new synchronized subject just for that event. This isn't what you want, and doesn't prevent concurrent OnNext() calls. You are therefore seeing a race condition bug due to concurrent calls in the observer OnNext() handler.

Instead of creating a synchronized subject in parseMethod, do it in the constructor OR synchronize on the subscription and it should work.

e.g. :

Either do this:

private ISubject<MyObject,MyObject> objectNotification =
    Subject.Synchronized(new Subject<MyObject>());

...

private void parseMethod(object sender, FileSystemEventArgs e)
{
    MyObject parsedFile = new MyObject(e.FullPath);
    File.Delete(e.FullPath);
    objectNotification.OnNext(parsedFile);
}

Or do this:

private void initSubscription()
{
    this.myObjectSubscription = sendingClass.Synchronize().Subscribe(parsedObject =>
    {
      this.AddObject(parsedObject);
    });
}

Note these alternatives have different behaviour in the presence of multiple subscribers!

In the first case you are ensuring there is at most one OnNext() in flight across all observers. The first event goes to the first subscriber, then to the second... until all subscribers have received that event, then the second event is sent etc.

In the second case you introduce some (usually safe) concurrency - now you are ensuring that there is at most one OnNext() in flight for a particular event across all observers. In other words, concurrent OnNext() calls can be made for different events in different observers, but only one OnNext() call for a given event will be in flight at any time, and no concurrent calls are made to the same observer.

Which you prefer is up to you, the latter allows observers to run in parallel but will still prevent the issue you are seeing; which approach is better depends on implementation details not provided above.

1
  • Added some detail explaining the variation in behaviour of the Synchronize alternatives. Jan 9, 2014 at 13:29

Your Answer

Reminder: Answers generated by Artificial Intelligence tools are not allowed on Stack Overflow. Learn more

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Not the answer you're looking for? Browse other questions tagged or ask your own question.