Séquence observable complète après un «long» intervalle

vous pouvez envisager d'utiliser Timeout Operator . Le seul inconvénient est qu'il se termine par un signal d'erreur. Vous devrez peut-être gérer une fausse erreur

L'opérateur Timeout vous permet d'abandonner un Observable avec une terminaison onError si cet Observable ne parvient pas à émettre des éléments pendant un laps de temps spécifié.

Si vous utilisez l'approche ci-dessous, vous pouvez surpasser l'erreur

characteristic.WhenNotificationReceived()
    .Timeout(TimeSpan.FromSeconds(1))
    .Subscribe(
        onNext: result =>
        {
            ....
            rfidPlayer.OnNext(....);
        },
        onError: error =>
        {
            rfidPlayer.OnCompleted();
        });

Une autre variante vous permet d'indiquer à timeout de basculer vers une sauvegarde Observable que vous spécifiez, plutôt que de se terminer avec une erreur, si la condition de timeout est déclenchée.

 .Timeout(200, Promise.resolve(42));

Je trouve désagréable d'utiliser Timeout cause des exceptions.

Je préfère injecter une valeur dans la séquence que je peux utiliser pour terminer la séquence. Si ma séquence, par exemple, produit des nombres non négatifs, alors si j'injecte un -1 je sais pour terminer la séquence.

Voici un exemple:

Commencez par cette observable qui génère des puissances de 2 commençant par 1 et elle retarde également la production de chaque valeur du nombre de millisecondes de la valeur en attente.

characteristic
    .WhenNotificationReceived()
    .Select(result => BitConverter.ToString(result.Data).Replace("-", ""))
    .Select(x => Observable.Timer(TimeSpan.FromSeconds(1.0)).Select(y => (string)null).StartWith(x))
    .Switch()
    .TakeWhile(x => x != null)
    .Subscribe(rfidPlayer);

Donc 1, 2, 4, 8, etc., de plus en plus lentement.

Maintenant, je veux arrêter cette séquence s'il n'y a pas de valeurs pendant 3.0 secondes, je peux le faire:

1 
2 
4 
8 
16 
32 
64 
128 
256 
512 
1024 
2048 

Si j'exécute cette séquence, j'obtiens cette sortie:

    .Select(x => Observable.Timer(TimeSpan.FromSeconds(3.0)).Select(y => -1).StartWith(x))
    .Switch()
    .TakeWhile(x => x >= 0)

La séquence est sur le point de produire 4096 mais elle attend d'abord 4096 millisecondes pour produire cette valeur - en attendant, Observable.Timer(TimeSpan.FromSeconds(3.0)) déclenche et génère un -1 , arrêtant ainsi la séquence.

L'élément clé de cette requête est l'utilisation de Switch . Il prend un IObservable<IObservable<T>> et produit un IObservable<T> en s'abonnant uniquement à la dernière observable externe et en se désabonnant de la précédente.

Ainsi, dans ma requête, chaque nouvelle valeur produite par la séquence s'arrête et redémarre le Timer .

Dans votre cas, votre observable ressemblerait à ceci:

Observable
    .Generate(1, x => true, x => 2 * x, x => x, x => TimeSpan.FromMilliseconds(x))

Voici un opérateur personnalisé TakeUntilTimeout vous pouvez utiliser, qui est une fine couche au-dessus de l'opérateur Timeout intégré.

/// <summary>
/// Applies a timeout policy for each element in the observable sequence.
/// If the next element isn't received within the specified timeout duration
/// starting from its predecessor, the sequence terminates.
/// </summary>
public static IObservable<T> TakeUntilTimeout<T>(
    this IObservable<T> source,
    TimeSpan timeout)
{
    return source.Timeout(timeout, Observable.Empty<T>());
}