Rx 需要帮助来编写类的层次结构

Question

我在最后一个示例中使用了 101 个 Rx Framework (http://rxwiki.wikidot.com/101samples#toc47) 示例，并创建了一个如下所示的类，并在测试函数中使用了类似的用法。

---

private void Test()
{
    var order = new Order();
    order.ObservableOrder.Subscribe(
        ord => Console.WriteLine("Order progress "),  // subscribe to onnext event
        ex => Console.WriteLine("Paid error " + ex.Message),   // subscribe to error event
        () => Console.WriteLine("Paid oncompleted ")  // subscribe to completed event
      ); // Subscribe

     order.Start();
 }


 public class Order
 {
      private readonly Subject<Order> _subject = new Subject<Order>();
      public IObservable<Order> ObservableOrder { get { return _subject.AsObservable(); } }

      public void RaiseError() {
         _subject.OnError(new Exception("test exception"));
      }
      public void RaiseCompleted() {
          _subject.OnCompleted();
      }
      public void RaiseProgress() {
          _subject.OnNext(this);
      }

      public void Start() {
             for (int i = 0; i < 5; i++)
                   this.RaiseProgress();

              this.RaiseError(); // either error is raised 
              this.RaiseCompleted();  // or completed is raised
      }
    } 

我的要求是如何编写一个基类（或某种工厂）来 使其易于用于 20-30 个做类似事情的类（就像他们需要 发送进度，完成和错误事件）另外，我怎样才能将一些数据发送回 关于进度和 oncompleted 事件（比如我们将异常发送到 onerror 事件）

Answer 1

OnCompleted() 不接受任何参数，显然无法更改。

你有几个选择：

您可以创建一个包含进程和最终有效负载的有效负载类。 class OrderProgress { double Process; OrderResult Result; }

您可以公开另一个主题Progress，除了ObservableOrder，您还可以订阅该主题。

如果您想保持Start 可重入，您可以让它返回一个同时暴露Progress 和OrderProgress 可观察对象的对象。

编辑：包括示例代码（我的第二个选项）：

private void Test() 
{ 
    var order = new Order(); 

    order.ObservableProgress.Subscribe( 
        prog => Console.WriteLine("Order progress " + prog.ToString()
    );

    order.ObservableOrder.Subscribe( 
        ord => Console.WriteLine("Order progress"),
        ex => Console.WriteLine("Paid error " + ex.Message), 
        () => Console.WriteLine("Paid oncompleted")
    );

    order.Start(); 
}


public class Order 
{ 
    private readonly Subject<Order> _subject = new Subject<Order>(); 
    private readonly Subject<double> _progressSubject = new Subject<double>(); 

    public IObservable<Order> ObservableOrder
    {
        get { return _subject.AsObservable(); }
    }

    public IObservable<double> ObservableProgress
    {
        get { return _progressSubject.AsObservable(); }
    }

    public void RaiseError() { 
        _subject.OnError(new Exception("test exception")); 
    } 

    public void RaiseCompleted() {
        _subject.OnNext(this); 
        _subject.OnCompleted(); 

        _progressSubject.OnCompleted();
    } 

    public void RaiseProgress(double progress) { 
        _progressSubject.OnNext(progress); 
    } 

    public void Start() { 
        for (int i = 0; i < 5; i++) 
            this.RaiseProgress((double)i / 5D); 

        this.RaiseError(); // either error is raised  
        this.RaiseCompleted();  // or completed is raised 
    } 
}