压缩分组的 IObservable

Question

public delegate void QuoteChangeEvent(IQuote q);
var priceChangedObservable = Observable.FromEvent<QuoteChangeEvent, IQuote> 
    (handler =>
    {
        QuoteChangeEvent qHandler = (e) =>
        {
            handler(e);
        };
        return qHandler;
    },
    qHandler => api.MAPI.OnQuoteChange += qHandler,
    qHandler => api.MAPI.OnQuoteChange -= qHandler);
    var grouped = priceChangedObservable
    .GroupBy(instrument => instrument.Symbol);

所以grouped是type IObservable<IGroupedObservable<string, IQuote>>

两个问题。

1(我试图

grouped.SortBy(instrument => instrument.Symbol);

但SortBy似乎不存在？

2( 假设有两个符号进来，GOOG 和 AAPL 进入grouped。 如何使用Zip运算符，以便Subscribe时得到的是Tuple<IQuote, IQuote>？

我无法完全获得正确的语法。像这样：

Observable.Zip(?, ?, (a, b) => Tuple.Create(a, b))
    .Subscribe(tup => Console.WriteLine("Quotes: {0} {1}", tup.item1, tup.item2));

编辑 1

我差点明白了：

    var first = grouped.Where(group => group.Key == "GOOG").FirstAsync();
    var second = grouped.Where(group => group.Key == "AAPL").FirstAsync();
    Observable.Zip(first, second, (a, b) => Tuple.Create(a, b))
    .Subscribe(tup => Console.WriteLine("Quotes: {0} {1}", tup.Item1, tup.Item2));

问题是tup不是类型 <IQuote, IQuote> ，而是类型：

Tuple<IGroupedObservable<string, IQuote>, IGroupedObservable<string, IQuote>>

Answer 1

这对你来说是什么样子的？

IObservable<(IQuote, IQuote)> results =
    grouped
        .Publish(gs =>
            gs
                .Where(g => g.Key == "GOOG")
                .SelectMany(g => g)
                .Zip(
                    gs
                        .Where(g => g.Key == "AAPL")
                        .SelectMany(g => g),
                    (x, y) => (x, y)));