c# - Rx .net Subscribe() 和 EventPattern 奇怪

Question

我将 Rx 与使用 EventPattern 的第三方 API 结合使用。在此 API 中，您在对象上注册事件处理程序，然后在对象上调用方法 StartWatching()，以启动事件以开始触发。我正在使用 Observable.FromEventPattern 桥接 Rx 世界中的 API，但我遇到了非常奇怪的问题，订阅只有在调用 StartWatching()。以下是我所看到的简化案例。

这个有效:

foreach (var iq in interactionQueues)
        {
            Observable.FromEventPattern(iq, "TheEvent")
                .Subscribe(e => Log.Info("I got called!"), 
                       e => Log.Info("Error!", e),
                       () => Console.WriteLine("Seq completed!"));

            iq.StartWatching();
        }

如果我在不同的循环中调用 Subscribe() 和 StartWatching() 它会停止工作:

foreach (var iq in interactionQueues)
            Observable.FromEventPattern(iq, "TheEvent")
                .Subscribe(e => Log.Info("I got called!"), 
                       e => Log.Info("Error!", e),
                       () => Console.WriteLine("Seq completed!"));
foreach (var iq in interactionQueues)
           iq.StartWatching();

关于为什么会发生这种情况，我唯一的想法是观察或订阅发生在错误的线程上。我尝试将 Scheduler.CurrentThread 和 Scheduler.Immediate 与 SubscribeOn 和 ObserveOn 一起使用，但这没有帮助。还有其他想法吗？我应该尝试不同的 Scheduler 还是转移注意力？

Answer 1

让我们用一种更友好的方法包装它:

public static TheEventArgs WatchEvent(this InteractionQueue this)
{
    var ret = Observable.Create<TheEventArgs>(subj => {
        // This entire block gets called every time someone calls Subscribe
        var disp = new CompositeDisposable();

        // Subscribe to the event
        disp.Add(Observable.FromEventPattern(iq, "TheEvent").Subscribe(subj));

        // Stop watching when we're done
        disp.Add(Disposable.Create(() => iq.StopWatching());

        iq.StartWatching();

        // This is what to Dispose on Unsubscribe
        return disp;
    });

    // When > 1 person Subscribes, only call the block above (i.e. StartWatching) once
    return ret.Multicast(new Subject<TheEventArgs>()).RefCount();
}