观察同步和异步结果

Question

使用 Rx，我想观察一个公开方法 GetItems 和事件 NewItem 的遗留对象。

当调用 GetItems 时，它将同步返回缓存中所有项目的列表。它还会生成项目的异步获取，这些项目将在收到时通过 NewItem 事件发布。

如何通过制定 LINQ 查询来捕获这两个结果集，以一致的方式观察这两个源（同步 + 异步）？生产订单并不重要。

Answer 1

让我看看我是否理解了您的遗留对象。我假设它是一个如下所示的泛型类型：

public class LegacyObject<T>
{
    public IEnumerable<T> GetItems();
    public event EventHandler<NewItemEventArgs<T>> NewItem;
}

使用如下所示的新项目事件参数：

public class NewItemEventArgs<T> : System.EventArgs
{
    public T NewItem { get; private set; }
    public NewItemEventArgs(T newItem)
    {
        this.NewItem = newItem;
    }
}

现在，我为 LegacyObject创建了一个 .ToObservable() 扩展方法;：

public static IObservable<T> ToObservable<T>(
    this LegacyObject<T> @this)
{
    return Observable.Create<T>(o =>
    {
        var gate = new object();
        lock (gate)
        {
            var list = new List<T>();
            var subject = new Subject<T>();
            var newItems = Observable
                .FromEventPattern<NewItemEventArgs<T>>(
                    h => @this.NewItem += h,
                    h => @this.NewItem -= h)
                .Select(ep => ep.EventArgs.NewItem);
            var inner = newItems.Subscribe(ni =>
            {
                lock (gate)
                {
                    if (!list.Contains(ni))
                    {
                        list.Add(ni);
                        subject.OnNext(ni);
                    }
                }
            });
            list.AddRange(@this.GetItems());
            var outer = list.ToArray().ToObservable()
                .Concat(subject).Subscribe(o);
            return new CompositeDisposable(inner, outer);
        }
    });
}

此方法为每个订阅者创建一个新的可观察对象 - 在编写这样的扩展方法时，这是正确的做法。

它创建一个gate对象来锁定对内部列表的访问。

因为您说调用 GetItems 的行为会生成异步函数来获取新项目，所以我确保在调用 GetItems 之前创建 NewItem 订阅。

inner 订阅会检查新项目是否在列表中，并且仅在主题不在列表中时才调用该主题的 OnNext 。

调用 GetItems 并通过 AddRange 将值添加到内部列表中。

在 NewItem 事件开始在另一个线程上触发之前，项目虽然不太可能但有可能不会添加到列表中。这就是为什么对列表的访问有锁的原因。 inner 订阅将等待，直到它能够获得锁定，然后再尝试将项目添加到列表中，这将在初始项目添加到列表后发生。

最后，内部列表变成一个可观察的，与主题连接起来，并且o观察者订阅这个可观察的。

使用 CompositeDisposable 将这两个订阅作为单个 IDisposable 返回。

这就是 ToObservable 方法。

现在，我通过在遗留对象上创建一个构造函数来测试这一点，该构造函数允许我传递可枚举值和可观察值。当调用 GetItems 时返回可枚举值，并且可观察值驱动 NewItem 事件。

因此，我的测试代码如下所示：

var tester = new Subject<int>();
var legacy = new LegacyObject<int>(new [] { 1, 2, 3, }, tester);

var values = legacy.ToObservable();

values.Subscribe(v => Console.WriteLine(v));

tester.OnNext(3);
tester.OnNext(4);
tester.OnNext(4);
tester.OnNext(5);

写入控制台的值是：

1
2
3
4
5

让我知道这是否满足您的需求。

Answer 2

如果我正确理解你的问题，可以使用这样的方法来完成：

legacyObject.GetItems().ToObservable()
    .Merge(
        Observable.FromEventPattern<IntEventArgs>(legacyObject, "NewItem")
            .Select(e => e.EventArgs.Value));

编辑：Enigmativity 在上面的解决方案中发现了竞争条件（请参阅下面的评论）。希望这个可以解决这个问题：

Observable.FromEventPattern<IntEventArgs>(legacyObject, "NewItem")
   .Select(e => e.EventArgs.Value)
   .Merge(Observable.Defer(() => legacyObject.GetItems().ToObservable()));

如果有帮助的话，这是我的其余测试代码。我不知道我是否准确地模拟了你的遗留类：

class IntEventArgs : EventArgs
{
    public IntEventArgs(int value)
    {
        Value = value;
    }

    public int Value { get; private set; }
}

class Legacy
{
    public event EventHandler<IntEventArgs> NewItem;

    public IList<int> GetItems()
    {
        GenerateNewItemAsync();
        return new List<int> { 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15 };
    }

    private void GenerateNewItemAsync()
    {
        Task.Factory.StartNew(() =>
        {
            for (var i = 0; i < 100000; ++i)
            {
                var handler = NewItem;
                if (handler != null) handler(this, new IntEventArgs(i));
                Thread.Sleep(500);
            }
        });
    }
}

class Example
{
    static void Main()
    {
        var legacyObject = new Legacy();

        legacyObject.GetItems().ToObservable()
            .Merge(
                Observable.FromEventPattern<IntEventArgs>(legacyObject, "NewItem")
                    .Select(e => e.EventArgs.Value))
            .Subscribe(Console.WriteLine);

        Console.ReadLine();
    }
}