¿Intervalo de cambio de operadores RX?

Question

Esto podría ser una pregunta estúpida como yo soy un poco nuevo RX :)

estoy muestreo de un evento (RX para .Net 4.0):

eventAsObservable.Sample (TimeSpan.FromSeconds (1)). Timestamp(). Suscribirse (x => Console.WriteLine ("testing:" + x.Value.EventArgs.str));

El problema es que el tiempo de muestreo debe poder cambiar sobre la marcha, supongo que podría hacer alguna propiedad que elimine el controlador existente y cree uno nuevo cuando cambie, pero parece un poco complicado y más vulnerable a los problemas de sincronización. ¿Hay alguna manera de simplemente cambiar el intervalo?

Ejemplo: decir que alguien está escribiendo una cadena de caracteres, cuando se detecta una cierta secuencia que desea cambiar el tiempo de muestreo sin perderse un evento, y preferiblemente por no obtener un evento más de una vez

Answer 1

I no sé de una forma de cambiar el intervalo de muestreo existente, pero podría hacer muestra a la frecuencia más alta que necesitará, y luego filtrar con una cláusula Where que usa una variable que puede cambiar.

Por ejemplo:

static IObservable<T> SampleEvery<T>(this IObservable<T> source, 
    Func<int> multipleProvider) 
{ 
    int counter = 0; 
    Func<T, bool> predicate = ignored => { 
     counter++; 
     if (counter >= multipleProvider()) 
     { 
      counter = 0; 
     } 
     return counter == 0; 
    }; 
    return source.Where(predicate); 
} 

Se podría entonces lo llama así:

// Keep this somewhere you can change it 
int multiple = 1; 

eventAsObservable.Sample(TimeSpan.FromSeconds(1)) 
       .SampleEvery(() => multiple) 
       .Timestamp() 
       .Subscribe(x => Console.WriteLine("testing:" + 
                x.Value.EventArgs.str)); 

Ahora, cambiando el valor de multiple va a cambiar la frecuencia de muestreo efectiva.

Es un truco bastante feo, pero creo que debería funcionar.

Answer 2

¿Por qué no te suscribes dos veces?

Observable.Merge(
    eventAsObservable.Sample(TimeSpan.FromSeconds(1)).Timestamp().SelectMany(x => doLocalLookup(x)), 
    eventAsObservable.Sample(TimeSpan.FromSeconds(10)).Timestamp().SelectMany(x => doRemoteLookup(x)), 
).Subscribe(Console.WriteLine); 

O si las búsquedas solo están activas según algún tipo de prefijo o calificador como '?' De Google Chrome operador:

Observable.Merge(
    eventAsObservable.Sample(TimeSpan.FromSeconds(1)).Where(x => isLocal(x)).SelectMany(x => doLocalLookup(x)), 
    eventAsObservable.Sample(TimeSpan.FromSeconds(10)).Where(x => isARemoteQuery(x).SelectMany(x => doRemoteLookup(x)), 
).Subscribe(Console.WriteLine); 

Answer 3

Sé que esta pregunta ya ha sido contestada, pero pensé que me gustaría añadir otro par de maneras de abordar de una manera Rx.

Usted podría utilizar Switch en una secuencia de TimeSpan 's:

private Subject<TimeSpan> sampleFrequencies = new Subject<TimeSpan>(); 

sampleFrequencies 
    .Select(x => eventAsObservable.Sample(Observable.Interval(x)).Timestamp()) 
    .Switch() 
    .Subscribe(x => .WriteLine("testing:" + x.Value.EventArgs.str)); 

// To change: 
// sampleFrequencies.OnNext(TimeSpan.FromSeconds(5)); 

Alternativamente, también podrían resolverse utilizando Defer, TakeUntil y Repeat (éste es un poco más loco y se incluye como un ejercicio de pensamiento):

private TimeSpan sampleFrequency = TiemSpan.FromSeconds(2); 
private Subject<Unit> frequencyChanged = new Subject<Unit>(); 

(Observable 
    .Defer(() => eventAsObservable 
     .Sample(Observable.Interval(sampleFrequency) 
    ) 
    .Timestamp() 
    .TakeUntil(frequencyChanged) 
).Repeat() 
.Subscribe(x => .WriteLine("testing:" + x.Value.EventArgs.str)); 

// To change: 
// sampleFrequency = TimeSpan.FromSeconds(5); 
// frequencyChanged.OnNext(new Unit()); 

Answer 4

TL; DR: Crear usando ObservableFromIntervalFunctor observable, como se muestra a continuación:

void Main() 
{ 
    // Pick an initial period, it can be changed later. 
    var intervalPeriod = TimeSpan.FromSeconds(1); 

    // Create an observable using a functor that captures the interval period. 
    var o = ObservableFromIntervalFunctor(() => intervalPeriod); 

    // Log every value so we can visualize the observable. 
    o.Subscribe(Console.WriteLine); 

    // Sleep for a while so you can observe the observable. 
    Thread.Sleep(TimeSpan.FromSeconds(5.0)); 

    // Changing the interval period will takes effect on next tick. 
    intervalPeriod = TimeSpan.FromSeconds(0.3); 

} 

IObservable<long> ObservableFromIntervalFunctor(Func<TimeSpan> intervalPeriodFunctor) 
{ 
    return Observable.Generate(0L, s => true, s => s + 1, s => s, s => intervalPeriodFunctor()); 
} 

Explicación: Observable.Generate tiene una sobrecarga que le permite especificar el momento en que el siguiente valor se generará a través de un funtor. Al pasar un funtor que ha capturado una variable de intervalo de tiempo, puede hacer que sea observable.cambio de período de intervalo al cambiar la variable de intervalo de tiempo capturada.

LINQPad fragmento de código here