[c#] .NET Reactive Extensions에서 주제가 권장되지 않는 이유는 무엇입니까?

좋아, 우리가 내 독단적 인 방식을 무시하고 “주체가 좋다 / 나쁘다”를 모두 무시한다면. 문제 공간을 살펴 보겠습니다.

나는 당신이 이해해야 할 시스템의 두 가지 스타일 중 하나를 가지고 있다고 확신합니다.

1. 메시지가 도착하면 시스템이 이벤트 또는 콜백을 발생시킵니다.
2. 처리 할 메시지가 있는지 확인하려면 시스템을 폴링해야합니다.

옵션 1의 경우 간단합니다. 적절한 FromEvent 메서드로 래핑하면됩니다. 술집에!

옵션 2의 경우 이제이를 폴링하는 방법과이를 효율적으로 수행하는 방법을 고려해야합니다. 또한 가치를 얻었을 때 어떻게 게시합니까?

폴링 전용 스레드가 필요하다고 생각합니다. 다른 코더가 ThreadPool / TaskPool을 망치고 ThreadPool 고갈 상황에 빠지는 것을 원하지 않을 것입니다. 또는 컨텍스트 전환의 번거 로움을 원하지 않습니다. 따라서 자체 스레드가 있다고 가정하면 폴링을 위해 앉아있는 While / Sleep 루프가있을 것입니다. 수표에서 일부 메시지를 찾으면이를 게시합니다. 이 모든 것이 Observable.Create에 완벽하게 들립니다. 이제 취소를 허용하기 위해 Disposable을 반환 할 수 없기 때문에 While 루프를 사용할 수 없습니다. 다행히도 전체 책을 읽었으므로 재귀 스케줄링에 능숙합니다!

나는 이와 같은 것이 효과가있을 것이라고 상상한다. #검증되지 않은

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

제가 주제를 정말 좋아하지 않는 이유는 대개 개발자가 문제에 대한 명확한 디자인을 가지고 있지 않은 경우입니다. 주제를 해킹하고 여기저기서 찔러 넣은 다음 WTF에서 지원 개발자가 추측 한 내용이 계속 진행되도록합니다. Create / Generate etc 메서드를 사용하면 시퀀스에 대한 효과를 지역화하는 것입니다. 하나의 방법으로 모든 것을 볼 수 있으며 다른 누구도 불쾌한 부작용을 일으키지 않는다는 것을 알 수 있습니다. 주제 필드가 보이면 이제 사용중인 클래스의 모든 장소를 찾아야합니다. 일부 MFer가 공개적으로 공개하면 모든 베팅이 해제됩니다. 누가이 시퀀스가 ​​사용되는지 알고 있습니다! Async / Concurrency / Rx는 어렵습니다. 부작용과 인과 관계 프로그래밍이 머리를 더 많이 돌리도록 허용하여 더 어렵게 만들 필요가 없습니다.

답변

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

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

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