使用 Rx,我想观察一个同时公开方法 GetItems
和事件 NewItem
的遗留对象。
当 GetItems
被调用时,它会同步返回缓存中所有项目的列表。它还会生成异步获取项目,这些项目将在收到项目时通过 NewItem
事件发布。
如何通过制定 LINQ 查询以捕获两个结果集,以一致的方式观察这两个源(同步 + 异步)?生产顺序不重要。
最佳答案
让我看看我是否理解了您的遗留对象。我假设它是一个看起来像这样的通用类型:
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;
}
}
现在,我创建了一个 .ToObservable()
LegacyObject<T>
的扩展方法:
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
的行为生成异步函数以获取新项目我已确保 NewItem
订阅是在调用 GetItems
之前创建的.
inner
订阅检查新项目是否在列表中并且只调用 OnNext
如果它不在列表中,则在该主题上。
调用GetItems
制作并通过AddRange
将值添加到内部列表中.
在 NewItem
之前,项目不会被添加到列表中的可能性不大,但有可能。事件开始在另一个线程上触发。这就是为什么对列表的访问有一个锁。 inner
订阅将等到它可以在尝试将项目添加到列表之前获得锁定,这将在初始项目添加到列表之后发生。
最后,内部列表变成了一个可观察对象,与主题连接起来,o
observer 订阅这个 observable。
这两个订阅作为单个 IDisposable
返回使用 CompositeDisposable
.
这就是 ToObservable
的内容方法。
现在,我通过在遗留对象上创建一个构造函数来测试它,该构造函数允许我传入可枚举值和可观察值。 GetItems
时返回枚举。被调用并且 observable 驱动 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
让我知道这是否满足您的需求。
关于c# - 观察同步和异步结果,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/7707458/