Почему в реактивных расширениях .NET не рекомендуются темы?

Хорошо, если мы проигнорируем мои догматические взгляды и проигнорируем все вместе «предметы хорошие / плохие». Давайте посмотрим на проблемное пространство.

Бьюсь об заклад, у вас либо есть 1 из 2 стилей системы, к которым нужно относиться снисходительно.

1. Система вызывает событие или обратный звонок при поступлении сообщения
2. Вам нужно опросить систему, чтобы увидеть, есть ли какие-либо сообщения для обработки

Для варианта 1, просто, мы просто оборачиваем его соответствующим методом FromEvent, и все готово. В паб!

Что касается варианта 2, теперь нам нужно подумать о том, как мы проводим опрос и как это сделать эффективно. Также, когда мы получаем значение, как мы его публикуем?

Я предполагаю, что вам понадобится выделенная ветка для опроса. Вы бы не хотели, чтобы какой-то другой программист забил ThreadPool / TaskPool и оставил вас в ситуации голодания ThreadPool. В качестве альтернативы вы не хотите хлопот с переключением контекста (я думаю). Итак, предположим, что у нас есть собственный поток, у нас, вероятно, будет какой-то цикл While / Sleep, который мы будем опрашивать. Когда проверка обнаруживает сообщения, мы их публикуем. Что ж, все это звучит идеально для Observable.Create. Теперь мы, вероятно, не можем использовать цикл While, поскольку он не позволит нам когда-либо вернуть Disposable, чтобы разрешить отмену. К счастью, вы прочитали всю книгу, так что разбираетесь в рекурсивном планировании!

Думаю, что-то подобное могло сработать. #Не испытано

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.
    }
}

Причина, по которой мне действительно не нравятся Subjects, заключается в том, что обычно разработчик не имеет четкого представления о проблеме. Взломайте тему, ткните ее здесь, там и везде, а затем позвольте бедному разработчику поддержки догадаться, что происходит WTF. Когда вы используете методы Create / Generate и т. Д., Вы локализуете эффекты в последовательности. Вы можете увидеть все это в одном методе, и вы знаете, что никто другой не вызывает неприятных побочных эффектов. Если я вижу поля темы, мне нужно искать все места в классе, в которых они используются. Если какой-то MFer выставит один публично, тогда все ставки отключены, кто знает, как эта последовательность используется! Async / Concurrency / Rx - это сложно. Вам не нужно усложнять задачу, позволяя побочным эффектам и программированию причинно-следственной связи еще больше вскружить вам голову.

В общем, вам следует избегать использования Subject, однако для того, что вы здесь делаете, я думаю, они работают достаточно хорошо. Когда я пришел, я задал аналогичный вопрос через сообщение «Избегайте тем» в учебниках Rx.

Чтобы процитировать Дэйва Секстона (из Rxx)

«Субъекты - это компоненты Rx с отслеживанием состояния. Они полезны, когда вам нужно создать подобный событию наблюдаемый объект в виде поля или локальной переменной».

Я обычно использую их как точку входа в Rx. Поэтому, если у меня есть код, который должен сказать «что-то случилось» (как у вас), я бы использовал Subject и вызвал OnNext. Затем выставьте это как IObservable, чтобы другие могли подписаться (вы можете использовать AsObservable() в своей теме, чтобы убедиться, что никто не может применить к теме и все испортить).

Вы также можете добиться этого с помощью события .NET и использовать FromEventPattern, но если я все равно собираюсь превратить событие в IObservable, я не вижу преимущества наличия события вместо Subject (что может означать, что я здесь что-то не хватает)

Однако чего вам следует избегать, так это подписки на IObservable с Subject, т.е. не передавать Subject в метод IObservable.Subscribe.

Часто, когда вы управляете предметом, вы на самом деле просто повторно реализуете функции, уже существующие в Rx, и, вероятно, не таким надежным, простым и расширяемым способом.

Когда вы пытаетесь адаптировать некоторый асинхронный поток данных к Rx (или создать асинхронный поток данных из потока, который в настоящее время не является асинхронным), наиболее распространенными случаями обычно являются:

• Источником данных является событие. Как говорит Ли, это простейший случай: используйте FromEvent и отправляйтесь в паб.

• Источником данных является синхронная операция, и вы хотите получать обновления по опросу (например, веб-сервис или вызов базы данных): в этом случае вы можете использовать подход, предложенный Ли, или для простых случаев вы можете использовать что-то как Observable.Interval.Select(_ => <db fetch>). Вы можете использовать DistinctUntilChanged (), чтобы предотвратить публикацию обновлений, когда в исходных данных ничего не изменилось.

• Источник данных - это своего рода асинхронный API, который вызывает ваш обратный вызов: в этом случае используйте Observable.Create, чтобы подключить ваш обратный вызов для вызова OnNext / OnError / OnComplete для наблюдателя.

• Источником данных является вызов, который блокируется до тех пор, пока не станут доступны новые данные (например, некоторые синхронные операции чтения из сокета): в этом случае вы можете использовать Observable.Create, чтобы обернуть императивный код, который читает из сокета. и публикуется в Observer.OnNext при чтении данных. Это может быть похоже на то, что вы делаете с темой.

Использование Observable.Create по сравнению с созданием класса, который управляет Subject, довольно эквивалентно использованию ключевого слова yield по сравнению с созданием всего класса, реализующего IEnumerator. Конечно, вы можете написать IEnumerator, чтобы он был таким же чистым и хорошим гражданином, как код yield, но какой из них лучше инкапсулирован и выглядит более аккуратным? То же самое верно для Observable.Create и управления Subject.

Observable.Create дает вам чистый шаблон для ленивой установки и чистого разрыва. Как этого добиться с помощью класса, обертывающего тему? Вам нужен какой-то метод Start ... как узнать, когда его вызвать? Или вы просто всегда запускаете его, даже когда его никто не слушает? И когда вы закончите, как вы заставите его перестать читать из сокета / опрашивать базу данных и т. Д.? У вас должен быть какой-то метод Stop, и у вас должен быть доступ не только к IObservable, на который вы подписаны, но и к классу, который изначально создал Subject.

С Observable.Create все это собрано в одном месте. Тело Observable.Create не запускается, пока кто-то не подпишется, поэтому, если никто не подписывается, вы никогда не используете свой ресурс. И Observable.Create возвращает Disposable, который может полностью отключить ваш ресурс / обратные вызовы и т.д. - это вызывается, когда Observer отписывается. Время жизни ресурсов, которые вы используете для создания Observable, аккуратно привязано к времени жизни самого Observable.

Цитируемый текст блока в значительной степени объясняет, почему вам не следует использовать Subject<T> , но, проще говоря, вы комбинируете функции наблюдателя и наблюдаемого, вводя какое-то состояние между ними (независимо от того, инкапсулируете ли вы или расширяете).

Здесь вы столкнетесь с проблемой; эти обязанности должны быть отделены друг от друга.

Тем не менее, в вашем конкретном случае я бы порекомендовал вам разбить ваши опасения на более мелкие части.

Во-первых, у вас есть активный поток, и вы всегда отслеживаете оборудование для сигналов, для которых нужно создавать уведомления. Как бы вы это сделали обычно? События. Итак, начнем с этого.

Давайте определим EventArgs, по которому будет запускаться ваше событие.

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

Теперь класс, который запускает событие. Обратите внимание, что это может быть статический класс (поскольку у вас всегда есть поток, выполняющий мониторинг аппаратного буфера) или что-то, что вы вызываете по запросу, которое подписывается на это. Вам придется изменить это соответствующим образом.

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.
    }
}

Итак, теперь у вас есть класс, который предоставляет событие. Наблюдаемые хорошо работают с событиями. Настолько, что существует первоклассная поддержка преобразования потоков событий (представьте поток событий как несколько запусков события) в _ 5_, если вы следуете стандартному шаблону событий через статический FromEventPattern метод в Observable классе .

Используя источник ваших событий и метод FromEventPattern, мы можем легко создать IObservable<EventPattern<BaseFrameEventArgs>> ( EventPattern<TEventArgs> класс воплощает то, что вы бы увидели в событии .NET, в частности, экземпляр, производный от EventArgs, и объект, представляющий отправителя), например:

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

Конечно, вам нужен IObservable<IBaseFrame>, но это просто, используя _ 14_ метод расширения в классе Observable для создания проекции (точно так же, как и в LINQ, и мы можем обернуть все это простым в использовании методом):

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

Плохо обобщать, что Субъекты не подходят для публичного интерфейса. Хотя это, безусловно, правда, что подход реактивного программирования не должен выглядеть, это определенно хороший вариант улучшения / рефакторинга для вашего классического кода.

Если у вас есть обычное свойство с общедоступным набором аксессуаров и вы хотите уведомлять об изменениях, ничто не говорит против его замены на BehaviorSubject. INPC или другие другие мероприятия просто не так чисты, и это лично меня утомляет. Для этой цели вы можете и должны использовать BehaviorSubjects в качестве общедоступных свойств вместо обычных свойств и отказаться от INPC или других событий.

Кроме того, Subject-interface делает пользователей вашего интерфейса более осведомленными о функциональности ваших свойств и с большей вероятностью подпишется, а не просто получит значение.

Лучше всего использовать, если вы хотите, чтобы другие слушали / подписывались на изменения свойства.