Erster ins Stocken geraten Schritt mit Reactive Extensions

Question

Ich kämpfe mit meiner ersten einfachen „Hallo Welt“ RX-Anwendung. Ich verwende VS2010 RC sowie den neuesten RX-Download.

Im Folgenden ist die einfache Konsole app;

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

erhalte ich ein Argument an der Where-Klausel. Hier ist der Stapel;

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

Diese Zeile scheint verursacht die Aufregung zu:

return observer as IDisposable;

Sie sollen nicht die die Beobachter davon ausgehen, Einweg, sollen Sie ein Wegwerf-Objekt zurückzugeben, die über „Abbestellen“ kennt.

  

Die Methode gibt einen Verweis auf eine   IDisposable-Schnittstelle. Das ermöglicht   der Beobachter auf Abmelden (das heißt,   Empfangen von Benachrichtigungen zu stoppen)   bevor der Provider abgeschlossen   Senden sie und rief die   Teilnehmer OnCompleted Verfahren.

Sie können es mit etwas zu tun, wie funktioniert:

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

!! Rote Fahne !!

Ich würde empfehlen, dass Sie nicht implementieren IObserver<T> oder selbst IObservable<T>. Favor die Verwendung von Observable.Create<T> oder als letztes Mittel die Subject Typen verwenden. Es gibt viele Dinge, die Sie beachten müssen, um richtig diese Schnittstellen zu implementieren, die für Sie durch die richtigen Rx Typen und Operatoren behandelt wird.

In diesem Beispiel würde ich Sie bitten, die MessageChannel Art und tauschen es für

fallen

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

Bei der weiteren Überprüfung eines Systemdesigns können Sie irgendeine Art von Dienstleistung, die entlarvt „Kanäle“ als beobachtbaren Sequenzen.

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

So negiert die Notwendigkeit für diese kundenspezifischen Implementierungen von IObserver<T> oder IObservable<T>.