Modifier l'intervalle des opérateurs RX?

Question

Cela pourrait être une question stupide que je suis un peu nouveau pour RX :)

J'échantillonnant un événement (RX pour .NET 4.0):

eventAsObservable.Sample (TimeSpan.FromSeconds (1)). Timestamp(). Subscribe (x => Console.WriteLine ("testing:" + x.Value.EventArgs.str)); Le problème est que le temps d'échantillonnage doit être capable de changer à la volée, je suppose que je pourrais faire une propriété qui supprime le gestionnaire existant et en crée un nouveau quand il change, mais il semble un peu brouillon et plus vulnérable aux problèmes de synchronisation. Y a-t-il un moyen de simplement changer l'intervalle?

Exemple: Dire que quelqu'un est en train d'écrire une chaîne de caractères, lorsqu'une certaine séquence est détecté que vous voulez changer le temps d'échantillonnage sans manquer un événement, et de préférence par ne pas obtenir un événement plus d'une fois

Answer 1

I Je ne sais pas d'une façon de changer l'intervalle d'échantillonnage existant, mais ce que pourrait faire est d'échantillonner à la fréquence la plus élevée dont vous aurez besoin, puis filtrer avec une clause Where qui utilise une variable peut changer.

Par exemple:

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); 
} 

Vous souhaitez alors appeler comme ça:

// 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)); 

Maintenant, en changeant la valeur de multiple va changer la fréquence d'échantillonnage efficace.

Il est un hack assez laid, mais je pense que cela devrait fonctionner.

Answer 2

Pourquoi ne pas vous abonner deux fois? Ou si les recherches ne sont actives que sur la base d'un préfixe ou d'un qualificatif tel que "?" De Google Chrome, par exemple. opérateur:

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

Je sais que cette question a déjà été répondue, mais j'ai pensé que j'ajouterais quelques autres façons de l'aborder de manière Rx.

Vous pouvez utiliser Switch sur une séquence 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)); 

Alternativement, il pourrait également être résolu en utilisant Defer, TakeUntil et Repeat (celui-ci est un peu plus fou et est inclus comme un exercice de pensée):

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: Créer une observable en utilisant ObservableFromIntervalFunctor, comme cela est indiqué ci-dessous:

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()); 
} 

Explication: Observable.Generate a une surcharge qui vous permet de spécifier le moment où la valeur suivante sera générée par un foncteur. En passant un foncteur qui a capturé une variable de timespan, vous pouvez rendre l'observable.changement de période d'intervalle en changeant la variable de laps de temps capturée.

LINQPad snippet here