Extensiones reactivas frente a FileSystemWatcher

Question

Una de las cosas que me ha molestado mucho acerca de FileSystemWatcher es la forma en que dispara varios eventos para un único cambio lógico en un archivo. Sé por qué sucede, pero no quiero tener que preocuparme; solo quiero volver a analizar el archivo una vez, no entre cuatro y seis veces seguidas. Idealmente, habría un evento que solo se dispara cuando un archivo dado se hace cambiando, en lugar de cada paso en el camino.

A lo largo de los años, he encontrado varias soluciones a este problema, con diversos grados de fealdad. Pensé que Reactive Extensions sería la mejor solución, pero hay algo que no estoy haciendo bien, y espero que alguien pueda señalar mi error.

que tienen un método de extensión:

public static IObservable<IEvent<FileSystemEventArgs>> GetChanged(this FileSystemWatcher that) 
{ 
    return Observable.FromEvent<FileSystemEventArgs>(that, "Changed"); 
} 

En última instancia, me gustaría conseguir un evento por nombre de archivo, dentro de un determinado período de tiempo - por lo que cuatro eventos en una fila con un único nombre de archivo se reducen a una evento, pero no pierdo nada si se modifican varios archivos al mismo tiempo. BufferWithTime suena como la solución ideal.

var bufferedChange = watcher.GetChanged() 
    .Select(e => e.EventArgs.FullPath) 
    .BufferWithTime(TimeSpan.FromSeconds(1)) 
    .Where(e => e.Count > 0) 
    .Select(e => e.Distinct()); 

Cuando me suscribo a este observable, un solo cambio en un archivo supervisado desencadena mi método de suscripción cuatro veces seguidas, que más bien en contra del propósito. Si elimino la llamada Distinct(), veo que cada una de las cuatro llamadas contiene dos eventos idénticos, por lo que hay algo de almacenamiento en búfer. Aumentar el TimeSpan pasado al BufferWithTime parece no tener ningún efecto: alcancé los 20 segundos sin ningún cambio en el comportamiento.

Esta es mi primera incursión en Rx, así que probablemente me esté perdiendo algo obvio. ¿Lo estoy haciendo mal? ¿Hay un mejor enfoque? Gracias por cualquier sugerencia ...

Answer 1

Mi error. De alguna manera tengo varios FileSystemWatchers monitoreando las carpetas de los demás. El observable se activaba una vez para cada observador, pero BufferWithTime parece estar funcionando correctamente. Todavía tengo que averiguar por qué mis vigilantes están disparando eventos para carpetas que pensé que estaban configuradas para ignorar, pero eso no tiene nada que ver con Rx o con esta pregunta.

De hecho, tal vez pueda despejar en ese problema, y ​​cambiar al tener un único vigilante de seguimiento de una carpeta principal, el uso de Rx para filtrar los eventos de carpetas que no estoy interesado.

Answer 2

BufferWithTime.Where () .Elija (...) va a hacer el trabajo, pero lo que realmente quiere es Throttle()

Answer 3

Sólo para calentar un viejo tema, ya que estoy trabajando en eso en este momento, también:

por supuesto este tema es insignificante en el contexto de ver un archivo, ya que FileSystemWatcher solo se dispara cada ~ 3 segundos con un evento modificado para un solo archivo cuando realizar el seguimiento de tamaño a través de

_fileSystemWatcher.NotifyFilter = NotifyFilters.Size | .... 

Pero asumamos FileSystemWatcher dispararía muchos eventos en una fila (tal vez muchos archivos se ha cambiado/renombra/creado), y otras personas que lean esto:

Usted no quiere use Throttle o BufferWithTime en este caso: Throttle es un poco engañoso .. prohíbe cualquier disparo hasta que transcurra TimeSpan time sin un evento. Significado: nunca podría disparar cuando se usa algo como Throttle(TimeSpan.FromMilliseconds(200)), y después de cada evento hay una pausa < 200 ms. Entonces, no es realmente el "estrangulamiento" que la gente espera.Es bueno para la entrada del usuario, cuando desea esperar hasta que el usuario haya dejado de escribir algo. Es malo para la regulación de la carga.

BufferWithTime tampoco es lo que desea: solo llena un búfer de tiempo. Es bueno cuando tienes una carga inicial alta por evento, como abrir una conexión a un servicio web. En ese caso, le conviene procesar por lotes los eventos cada "tiempo" segundos. Pero no cuando se equilibra la carga, ya que la cantidad de eventos no cambia.

La solución es el método Sample(TimeSpan time): toma el último evento dentro de un TimeSpan, que es el acelerador "real". Creo que los chicos de Rx realmente estropearon el nombramiento en este caso.

Answer 4

puede usar un grupo para agregar eventos del sistema de archivos por nombre de archivo, y usar el observable resultante con el método de extensiones Throttle. He escrito una pequeña muestra con números enteros, pero la idea básica es la misma.

var obs = from n in Enumerable.Range(1, 40).ToObservable() 
    group n by n/10 into g 
    select new { g.Key, Obs = g.Throttle(TimeSpan.FromMilliseconds(10.0)) } into h 
    from x in h.Obs 
    select x; 
obs.Subscribe(x => Console.WriteLine(x)); 

salidas:

9 
19 
29 
39 
40 

que es para cada grupo (n/10) el último número entero observada.