c# - Rx 订阅 FromEvents 合并所有

Question

假设我有一组股票代码:

StockTicker stockTicker = 
  new StockTicker("MSFT", "APPL", "YHOO", "GOOG");
stockTicker.OnPriceChanged += (sender, args ) => 
{
  Console.WriteLine(
    "{0}: Price - {1}  Volume - {2}", 
    args.StockSymbol, args.Price, args.Volume);
};

我可以订阅事件并将其作为 hot IObservable 获取:

IObservable<PriceChangedEventArgs> priceChangedObservable = 
  Observable.FromEventPattern<PriceChangedEventArgs>(
    eventHandler => stockTicker.OnPriceChanged += eventHandler,
    eventHandler => stockTicker.OnPriceChanged -= eventHandler )
      .Select( eventPattern => eventPattern.EventArgs );

priceChangedObservable.Subscribe(args => Console.WriteLine( 
  "{0}: Price - {1}  Volume - {2}", 
    args.StockSymbol, args.Price, args.Volume ) );

这会在新报价进入时打印每个交易品种的报价:按顺序，每个事件都是一个单行 YHOO 25.33、MSFT 127、AAPL 175、GOOG 1126 等

如何修改上面的代码，使每个“报价单”都是每个单独报价单的所有当前报价单的组合？ (YHOO 25.33、MSFT 127、GOOG 1126、AAPL 175)。 “报价”现在是从所有这些中看到的最后一条报价的状态，作为一个“报价”。

我看到 Rx 有一个 Zip 运算符，但它的语义似乎需要 n IObservables 来压缩。这里有多个引号来自同一个 IObservable?

因此 Console.WriteLine 会将订阅的所有 n 引号打印为单个点事件(并且仅在它们都具有值时才打印)

Answer 1

如果这些是以任何顺序到达的随机事件，而您随时只对最新值感兴趣，那么您可能想要一个使用 Scan 保持更新的字典。运算符(operator):

source
    /* keep track of the latest value for each symbol */
    .Scan(new Dictionary<string, decimal>(), (a, b) => a[b.StockSymbol] = b.Price)

    /* example logic to wait until they all have values */
    .Where(dict => stockTicker.StockSymbols.All(dict.ContainsKey))

    .Subscribe(dict => { ... });

---

如果您要缓冲的事件数保持不变并且您知道源在事件之间循环，则 Buffer运营商可能适合你:

source
    /* burst values periodically */
    .Buffer(count)

    .Subscribe(list => { ... })