أول تعثر خطوة مع رد الفعل امتداد

Question

أنا تكافح مع اول بسيطة "مرحبا العالم" RX التطبيق.أنا باستخدام VS2010 RC, بالإضافة إلى أحدث RX تحميل.

وفيما يلي بسيطة وحدة التحكم التطبيق ؛

    class Program
    {
        static void Main(string[] args)
        {

            var channel = new MessageChannel()
                .Where(m => m.process)
                .Subscribe((MyMessage m) => Console.WriteLine(m.subject));

            //channel.GenerateMsgs();
        }
    }

    public class MyMessage
    {
        public string subject;
        public bool process;
    }

    public class MessageChannel: IObservable<MyMessage>
    {
        List<IObserver<MyMessage>> observers = new List<IObserver<MyMessage>>();

        public IDisposable Subscribe(IObserver<MyMessage> observer)
        {
            observers.Add(observer);
            return observer as IDisposable;
        }

        public void GenerateMsgs()
        {
            foreach (IObserver<MyMessage> observer in observers)
            {
                observer.OnNext(new MyMessage() {subject = "Hello!", process = true});
            }
        }
    }

أحصل على ArgumentNullException في جملة Where.وهنا كومة ؛

System.ArgumentNullException was unhandled
  Message=Value cannot be null.
Parameter name: disposable
  Source=System.Reactive
  ParamName=disposable
  StackTrace:
       at System.Collections.Generic.AnonymousObservable`1.Disposable.Set(IDisposable disposable)
       at System.Collections.Generic.AnonymousObservable`1.<>c__DisplayClass1.<Subscribe>b__0()
       at System.Threading.Scheduler.NowScheduler.Schedule(Action action)
       at System.Collections.Generic.AnonymousObservable`1.Subscribe(IObserver`1 observer)
       at ConsoleApplication1.Program.Main(String[] args) in C:\Users\Jason\documents\visual studio 2010\Projects\ConsoleApplication1\ConsoleApplication1\Program.cs:line 18
       at System.AppDomain._nExecuteAssembly(RuntimeAssembly assembly, String[] args)
       at System.AppDomain.ExecuteAssembly(String assemblyFile, Evidence assemblySecurity, String[] args)
       at Microsoft.VisualStudio.HostingProcess.HostProc.RunUsersAssembly()
       at System.Threading.ThreadHelper.ThreadStart_Context(Object state)
       at System.Threading.ExecutionContext.Run(ExecutionContext executionContext, ContextCallback callback, Object state, Boolean ignoreSyncCtx)
       at System.Threading.ExecutionContext.Run(ExecutionContext executionContext, ContextCallback callback, Object state)
       at System.Threading.ThreadHelper.ThreadStart()
  InnerException: 

Answer 1

وهذا الخط يبدو أن يسبب ضجة:

return observer as IDisposable;

وليس من المفترض أن لتولي المراقب غير القابل للتصرف، من المفترض أن يعود كائن المتاح أن يعرف عن "إلغاء الاشتراك".

<اقتباس فقرة>   

والأسلوب بإرجاع مرجع ل   واجهة IDisposable. وهذا يتيح   المراقب لإلغاء الاشتراك (أي،   لإيقاف تلقي الإخطارات)   قبل انتهاء مزود   إرسالها، ودعا   طريقة OnCompleted المشترك.

ويمكنك جعله يعمل عن طريق القيام بشيء مثل:

public class MessageChannel: IObservable<MyMessage>
{
    class Subscription : IDisposable {
        MessageChannel _c;
        IObservable<MyMessage> _obs;
        public Subscription(MessageChannel c, IObservable<MyMessage> obs) { 
            _c = c; _obs = obs;
        }
        public void Dispose() {
            _c.Unsubscribe(_obs);
        }
    }

    public IDisposable Subscribe(IObserver<MyMessage> observer)
    {
        observers.Add(observer);
        return new Subscription(this, observer);
    }

    void Unsubscribe(IObservable<MyMessage> obs) {
        observers.Remove(obs);
    }
}

Answer 2

!!العلم الأحمر!!

أود أن أقترح بشدة أن لا تنفذ IObserver<T> أو IObservable<T> نفسك.يفضل استخدام Observable.Create<T> أو كملاذ أخير استخدام Subject أنواع.هناك الكثير من الأشياء التي تحتاج إلى النظر بشكل صحيح تنفيذ هذه الواجهات التي يتم التعامل معها بالنسبة لك الصحيح Rx أنواع المشغلين.

في هذا المثال أود أن أحثكم على إسقاط MessageChannel نوع مبادلتها

class Program
{
    static void Main(string[] args)
    {
        var channel = GenerateMsgs()
            .Where(m => m.process)
            .Subscribe((MyMessage m) => Console.WriteLine(m.subject));
    }

    public IObservable<MyMessage> GenerateMsgs()
    {
        return Observable.Create<MyMessage>(observer=>
        {
            observer.OnNext(new MyMessage() {subject = "Hello!", process = true});
        });
    }
}

public class MyMessage
{
    public string subject;
    public bool process;
}

على مزيد من التفتيش من تصميم النظام قد يكون نوعا من الخدمة التي تكشف "قنوات" ملحوظة متواليات.

public interface OrderService
{
    IObservable<OrderRequest> OrderRequests();
    IObservable<Order> ProcessedOrders();
    IObservable<OrderRejection> OrdersRejections();
}

مما يلغي الحاجة إلى هذه مخصصة تطبيقات IObserver<T> أو IObservable<T>.