Rx.NET的可观察介体

问题描述

class SomeMediator
{
  public string[] Inputs { get; private set; }
  public ISubject<string> Response { get; private set; }
  
  public SomeMediator(string[] inputs)
  {
    Inputs = inputs;
    Response = new Subject<string>();
  }
}

class SomeProcessor
{
  public SomeProcessor(SomeMediator mediator)
  {
    foreach (var input in mediator.Inputs) 
    {
      var clock = new Timer();
      
      clock.Interval = 1000;
      clock.Enabled = true;
      clock.Elapsed += new ElapsedEventHandler(
        (object source,ElapsedEventArgs e) => 
          mediator.Response.OnNext(string.Empty));
    }
  }
}

class SomeRenderer 
{
  public SomeRenderer(SomeMediator mediator)
  {
    mediator.Subscribe(o => Console.WriteLine(o));
  }
}

用法

var mediator = new SomeMediator(new[] { "Input#1","Input#2" });
var subscriber = SomeRenderer(mediator);
var processor = SomeProcessor(mediator);

上面的代码将带有2个输入参数的SomeMediator注入SomeProcessor中，并将SomeRenderer订阅到所有响应中。 SomeProcessor基于2个输入，创建2个Timer线程，并开始向SomeMediator发送通知。每个Timer都异步调用SomeMediator.Response.OnNext，下面我收到警告

Warning: 0 : OnNext called while another OnNext call was in progress on the same Observer

结果，第二条OnNext消息丢失了，因此SomeMediator仅收到SomeProcessor的1条响应，而不是2条。

问题

是否可以通过可观察的方式实现调解器交互，并防止由不同线程引起的OnNext冲突？

更新-示例

比方说，有一个股票交易应用程序具有一些抽象的通用Core功能，可以被其他应用程序重用。该核心可以包括一些市场数据格式化程序（中介程序），以将各种格式的API响应转换为统一模型（响应）并对其进行可视化（渲染）。应该从不同的网关（处理器）接收到响应，这些网关将具有特定的实现方式，并将取决于经纪公司提供的API。

换句话说，我们有多个处理器，例如SomeProcessor1，SomeProcessor2，SomeProcessor3，它们将充当API包装器。它们接受HTTP请求（输入）的一些输入参数并返回响应。输入由API处理器使用。中介器仅获取结果并将其发送到可视化器。中介器具有Subject，该变量传递给每个Processor，并允许Renderer订阅这些更改。每次某个处理器发送更新时，渲染器都应通过调解器接收此通知，并将其显示在UI中。

Processor1      Processor2     Processor3        (API)
            \       |        /
                 Mediator                        (Aggregate and Format)
               /         \
         Logger           Renderer               (Subscribed tasks - Log and display)

解决方法

暂无找到可以解决该程序问题的有效方法，小编努力寻找整理中！

如果你已经找到好的解决方法，欢迎将解决方案带上本链接一起发送给小编。

小编邮箱:dio#foxmail.com (将#修改为@）

c#c#rx.net

Rx.NET的可观察介体

问题描述

解决方法

相关问答