Pourquoi les sujets ne sont-ils pas recommandés dans les extensions réactives .NET?

Je suis actuellement en train de me familiariser avec le framework Reactive Extensions pour .NET et je suis en train de parcourir les différentes ressources d'introduction que j'ai trouvées (principalement http://www.introtorx.com )

Notre application implique un certain nombre d'interfaces matérielles qui détectent les trames réseau, ce seront mes IObservables, j'ai alors une variété de composants qui consommeront ces trames ou effectueront une certaine manière de transformer les données et de produire un nouveau type de trame. Il y aura également d'autres composants qui doivent afficher chaque nième image par exemple. Je suis convaincu que Rx va être utile pour notre application, mais j'ai du mal avec les détails d'implémentation de l'interface IObserver.

La plupart (sinon la totalité) des ressources que j'ai lues ont dit que je ne devrais pas implémenter l'interface IObservable moi-même mais utiliser l'une des fonctions ou classes fournies. D'après mes recherches, il semble que la création d'un Subject<IBaseFrame>me fournirait ce dont j'ai besoin, j'aurais mon thread unique qui lit les données de l'interface matérielle, puis appelle la fonction OnNext de mon Subject<IBaseFrame>instance. Les différents composants IObserver recevraient alors leurs notifications de ce sujet.

Ma confusion vient des conseils donnés en annexe de ce tutoriel où il est dit:

Évitez d'utiliser les types de sujet. Rx est en fait un paradigme de programmation fonctionnel. Utiliser des sujets signifie que nous gérons maintenant l'état, qui est potentiellement en train de muter. Il est très difficile de gérer à la fois l'état mutant et la programmation asynchrone. De plus, de nombreux opérateurs (méthodes d'extension) ont été soigneusement écrits pour garantir que la durée de vie correcte et cohérente des abonnements et des séquences est maintenue; lorsque vous introduisez des sujets, vous pouvez rompre cela. Les versions futures peuvent également voir une dégradation significative des performances si vous utilisez explicitement des sujets.

Mon application est assez critique pour les performances, je vais évidemment tester les performances de l'utilisation des modèles Rx avant d'entrer dans le code de production; cependant, je crains de faire quelque chose qui va à l'encontre de l'esprit du framework Rx en utilisant la classe Subject et qu'une future version du framework va nuire aux performances.

Existe-t-il une meilleure façon de faire ce que je veux? Le thread d'interrogation matériel va fonctionner en continu, qu'il y ait des observateurs ou non (le tampon matériel sauvegardera autrement), c'est donc une séquence très chaude. Je dois ensuite transmettre les trames reçues à plusieurs observateurs.

Tout avis serait grandement apprécié.

Ok, si nous ignorons mes manières dogmatiques et ignorons "les sujets sont bons / mauvais" tous ensemble. Examinons l'espace des problèmes.

Je parie que vous avez l'un des deux styles de système auxquels vous devez vous féliciter.

1. Le système déclenche un événement ou un rappel lorsqu'un message arrive
2. Vous devez interroger le système pour voir s'il y a un message à traiter

Pour l'option 1, facile, nous l'enveloppons simplement avec la méthode FromEvent appropriée et nous avons terminé. Au bar!

Pour l'option 2, nous devons maintenant examiner comment nous sondons cela et comment le faire efficacement. Aussi, lorsque nous obtenons la valeur, comment la publier?

J'imagine que vous voudriez un fil dédié pour le sondage. Vous ne voudriez pas qu'un autre codeur martèle le ThreadPool / TaskPool et vous laisse dans une situation de famine ThreadPool. Sinon, vous ne voulez pas les tracas de la commutation de contexte (je suppose). Supposons donc que nous ayons notre propre thread, nous aurons probablement une sorte de boucle While / Sleep dans laquelle nous nous asseyons pour interroger. Lorsque le chèque trouve des messages, nous les publions. Eh bien, tout cela semble parfait pour Observable.Create. Maintenant, nous ne pouvons probablement pas utiliser une boucle While car cela ne nous permettra pas de retourner un jetable pour permettre l'annulation. Heureusement, vous avez lu tout le livre, alors soyez avertis avec la planification récursive!

J'imagine que quelque chose comme ça pourrait fonctionner. #Pas testé

public class MessageListener
{
    private readonly IObservable<IMessage> _messages;
    private readonly IScheduler _scheduler;

    public MessageListener()
    {
        _scheduler = new EventLoopScheduler();

        var messages = ListenToMessages()
                                    .SubscribeOn(_scheduler)
                                    .Publish();

        _messages = messages;
        messages.Connect();
    }

    public IObservable<IMessage> Messages
    {
        get {return _messages;}
    }

    private IObservable<IMessage> ListenToMessages()
    {
        return Observable.Create<IMessage>(o=>
        {
                return _scheduler.Schedule(recurse=>
                {
                    try
                    {           
                        var messages = GetMessages();
                        foreach (var msg in messages)
                        {
                            o.OnNext(msg);
                        }   
                        recurse();
                    }
                    catch (Exception ex)
                    {
                        o.OnError(ex);
                    }                   
                });
        });
    }

    private IEnumerable<IMessage> GetMessages()
    {
         //Do some work here that gets messages from a queue, 
         // file system, database or other system that cant push 
         // new data at us.
         // 
         //This may return an empty result when no new data is found.
    }
}

La raison pour laquelle je n'aime vraiment pas les sujets, c'est que le développeur n'a généralement pas une conception claire du problème. Piratez un sujet, piquez-le ici là-bas et partout, puis laissez le pauvre développeur de soutien deviner que WTF se passait. Lorsque vous utilisez les méthodes Créer / Générer, etc., vous localisez les effets sur la séquence. Vous pouvez tout voir dans une seule méthode et vous savez que personne d'autre ne lance un effet secondaire désagréable. Si je vois un champ de sujet, je dois maintenant chercher tous les endroits dans une classe où il est utilisé. Si un MFer en expose un publiquement, alors tous les paris sont ouverts, qui sait comment cette séquence est utilisée! Async / Concurrency / Rx est difficile. Vous n'avez pas besoin de rendre les choses plus difficiles en permettant aux effets secondaires et à la programmation de causalité de vous faire encore plus tourner la tête.

En général, vous devriez éviter d'utiliser Subject, mais pour ce que vous faites ici, je pense qu'ils fonctionnent très bien. J'ai posé une question similaire quand je suis tombé sur le message "éviter les sujets" dans les tutoriels Rx.

Pour citer Dave Sexton (de Rxx)

«Les sujets sont les composants avec état de Rx. Ils sont utiles lorsque vous devez créer une observable semblable à un événement en tant que champ ou variable locale.»

J'ai tendance à les utiliser comme point d'entrée dans Rx. Donc, si j'ai un code qui a besoin de dire «quelque chose s'est passé» (comme vous l'avez fait), j'utiliserais un Subjectand call OnNext. Ensuite, exposez-le IObservablepour que les autres puissent s'y abonner (vous pouvez l'utiliser AsObservable()sur votre sujet pour vous assurer que personne ne peut lancer un casting sur un sujet et gâcher les choses).

Vous pouvez également y parvenir avec un événement .NET et une utilisation FromEventPattern, mais si je ne fais que transformer l'événement en un événement de IObservabletoute façon, je ne vois pas l'avantage d'avoir un événement au lieu d'un Subject(ce qui pourrait signifier que je suis absent quelque chose ici)

Cependant, ce que vous devriez éviter assez fortement est de souscrire à un IObservableavec a Subject, c'est-à-dire ne pas passer a Subjectdans la IObservable.Subscribeméthode.

Souvent, lorsque vous gérez un sujet, vous ne faites que réimplémenter des fonctionnalités déjà dans Rx, et probablement d'une manière pas aussi robuste, simple et extensible.

Lorsque vous essayez d'adapter un flux de données asynchrone dans Rx (ou de créer un flux de données asynchrone à partir d'un flux qui n'est pas actuellement asynchrone), les cas les plus courants sont généralement:

• La source des données est un événement : comme le dit Lee, c'est le cas le plus simple: utilisez FromEvent et dirigez-vous vers le pub.

• La source des données provient d'une opération synchrone et vous voulez des mises à jour interrogées (par exemple, un service Web ou un appel de base de données): Dans ce cas, vous pouvez utiliser l'approche suggérée par Lee, ou pour des cas simples, vous pouvez utiliser quelque chose comme Observable.Interval.Select(_ => <db fetch>). Vous souhaiterez peut-être utiliser DistinctUntilChanged () pour empêcher la publication de mises à jour lorsque rien n'a changé dans les données source.

• La source des données est une sorte d'API asynchrone qui appelle votre rappel : dans ce cas, utilisez Observable.Create pour connecter votre rappel afin d'appeler OnNext / OnError / OnComplete sur l'observateur.

• La source des données est un appel qui bloque jusqu'à ce que de nouvelles données soient disponibles (par exemple, certaines opérations de lecture de socket synchrone): Dans ce cas, vous pouvez utiliser Observable.Create pour envelopper le code impératif qui lit depuis le socket et publie sur l'Observer. lorsque les données sont lues. Cela peut être similaire à ce que vous faites avec le sujet.

Utiliser Observable.Create vs créer une classe qui gère un Subject équivaut assez à utiliser le mot-clé yield vs créer une classe entière qui implémente IEnumerator. Bien sûr, vous pouvez écrire un IEnumerator pour être aussi propre et aussi bon citoyen que le code de rendement, mais lequel est le mieux encapsulé et offre un design plus soigné? La même chose est vraie pour Observable.Create vs la gestion des sujets.

Observable.Create vous donne un modèle propre pour une configuration paresseuse et un démontage propre. Comment y parvenir avec une classe enveloppant un sujet? Vous avez besoin d'une sorte de méthode Start ... comment savoir quand l'appeler? Ou le démarrez-vous toujours, même si personne n'écoute? Et quand vous avez terminé, comment arrêtez-vous de lire à partir du socket / d'interroger la base de données, etc.? Vous devez avoir une sorte de méthode Stop, et vous devez toujours avoir accès non seulement à l'IObservable auquel vous êtes abonné, mais à la classe qui a créé le sujet en premier lieu.

Avec Observable.Create, tout est regroupé au même endroit. Le corps de Observable.Create n'est pas exécuté tant que quelqu'un ne s'est pas abonné, donc si personne ne s'abonne, vous n'utilisez jamais votre ressource. Et Observable.Create renvoie un jetable qui peut arrêter proprement votre ressource / vos rappels, etc. - cela est appelé lorsque l'observateur se désabonne. La durée de vie des ressources que vous utilisez pour générer l'observable est parfaitement liée à la durée de vie de l'observable lui-même.

Le texte de bloc cité explique à peu près pourquoi vous ne devriez pas utiliser Subject<T>, mais pour le dire plus simple, vous combinez les fonctions d'observateur et d'observable, tout en injectant une sorte d'état entre les deux (que vous encapsuliez ou étendiez).

C'est là que vous rencontrez des problèmes; ces responsabilités devraient être séparées et distinctes les unes des autres.

Cela dit, dans votre cas particulier , je vous recommande de diviser vos préoccupations en parties plus petites.

Tout d'abord, votre thread est chaud et surveille toujours le matériel pour les signaux à émettre des notifications. Comment feriez-vous cela normalement? Événements . Alors commençons par ça.

Définissons le EventArgsdéclenchement de votre événement.

// The event args that has the information.
public class BaseFrameEventArgs : EventArgs
{
    public BaseFrameEventArgs(IBaseFrame baseFrame)
    {
        // Validate parameters.
        if (baseFrame == null) throw new ArgumentNullException("IBaseFrame");

        // Set values.
        BaseFrame = baseFrame;
    }

    // Poor man's immutability.
    public IBaseFrame BaseFrame { get; private set; }
}

Maintenant, la classe qui déclenchera l'événement. Remarque, cela pourrait être une classe statique (puisque vous avez toujours un fil conducteur suivi de la mémoire tampon du matériel), ou quelque chose que vous appelez à la demande qui est abonnée à ce que . Vous devrez le modifier le cas échéant.

public class BaseFrameMonitor
{
    // You want to make this access thread safe
    public event EventHandler<BaseFrameEventArgs> HardwareEvent;

    public BaseFrameMonitor()
    {
        // Create/subscribe to your thread that
        // drains hardware signals.
    }
}

Alors maintenant, vous avez une classe qui expose un événement. Les observables fonctionnent bien avec les événements. À tel point qu'il existe un support de première classe pour la conversion de flux d'événements (pensez à un flux d'événements comme plusieurs déclenchements d'un événement) en IObservable<T>implémentations si vous suivez le modèle d'événement standard, via la méthode statiqueFromEventPattern sur la Observableclasse .

Avec la source de vos événements et la FromEventPatternméthode, nous pouvons créer IObservable<EventPattern<BaseFrameEventArgs>>facilement un (la EventPattern<TEventArgs>classe incarne ce que vous verriez dans un événement .NET, notamment, une instance dérivée de EventArgset un objet représentant l'expéditeur), comme ceci:

// The event source.
// Or you might not need this if your class is static and exposes
// the event as a static event.
var source = new BaseFrameMonitor();

// Create the observable.  It's going to be hot
// as the events are hot.
IObservable<EventPattern<BaseFrameEventArgs>> observable = Observable.
    FromEventPattern<BaseFrameEventArgs>(
        h => source.HardwareEvent += h,
        h => source.HardwareEvent -= h);

Bien sûr, vous voulez un IObservable<IBaseFrame>, mais c'est facile, en utilisant la Selectméthode d'extension sur la Observableclasse pour créer une projection (comme vous le feriez dans LINQ, et nous pouvons envelopper tout cela dans une méthode facile à utiliser):

public IObservable<IBaseFrame> CreateHardwareObservable()
{
    // The event source.
    // Or you might not need this if your class is static and exposes
    // the event as a static event.
    var source = new BaseFrameMonitor();

    // Create the observable.  It's going to be hot
    // as the events are hot.
    IObservable<EventPattern<BaseFrameEventArgs>> observable = Observable.
        FromEventPattern<BaseFrameEventArgs>(
            h => source.HardwareEvent += h,
            h => source.HardwareEvent -= h);

    // Return the observable, but projected.
    return observable.Select(i => i.EventArgs.BaseFrame);
}

Il est mauvais de généraliser que les sujets ne sont pas bons à utiliser pour une interface publique. S'il est certainement vrai que ce n'est pas à quoi devrait ressembler une approche de programmation réactive, c'est définitivement une bonne option d'amélioration / refactorisation pour votre code classique.

Si vous avez une propriété normale avec un accesseur d'ensemble public et que vous souhaitez notifier les modifications, rien ne s'oppose à son remplacement par un BehaviorSubject. INPC ou d'autres événements supplémentaires ne sont tout simplement pas aussi propres et cela m'épuise personnellement. À cette fin, vous pouvez et devez utiliser BehaviorSubjects comme propriétés publiques au lieu de propriétés normales et abandonner INPC ou d'autres événements.

De plus, l'interface Objet rend les utilisateurs de votre interface plus conscients de la fonctionnalité de vos propriétés et sont plus susceptibles de s'abonner au lieu d'obtenir simplement la valeur.

C'est le meilleur à utiliser si vous souhaitez que d'autres écoutent / s'abonnent aux modifications d'une propriété.