3

我正在为我的第一个简单的“hello world”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});
            }
        }
    }

我在 Where 子句中得到一个 ArgumentNullException。这是堆栈；

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:
4

2 回答 2

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);
    }
}
于 2010-02-13T19:09:23.100 回答
1

！！红色的标志！！

我强烈建议您不要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>or的这些自定义实现的需求IObservable<T>。

于 2013-04-24T15:03:06.413 回答