如何解开嵌套的IObservable

Question

我正在尝试从以下代码创建IObservable<string>，但是我似乎找不到找到正确解开事件处理程序值的方法。

发生的事情是，PasswordBox可能会更改，因此，每当我想要观察它时，都将在其上进行观察，并在引发密码更改事件时提供一个字符串资源。如果我对普通事件进行处理，效果很好，但我对使用System.Reactive如何工作感到好奇。

        var passwordChanged = WhenPropertyChanged
            .Where(name => nameof(PasswordBox) == name)
            .Select(d => PasswordBox)
            .Where(d => d != null)
            .Select(box =>
            {
                return Observable.FromEvent<RoutedEventHandler, RoutedEventArgs>(
                    handler => box.PasswordChanged += handler,
                    handler => box.PasswordChanged -= handler);
            }).Select(d => nameof(Password));

对我来说，似乎必须在Select（box => ...部分内有某种方式，我可以返回一个可以用于正确订阅的对象（IObservable<IObservable<RoutedEventArgs>>除外）。

像下面的作品那样做。但是我认为，如果您端到端使用它，那么React可以更好地避免事件处理程序内存泄漏。

        var passwordHasChanged = new Subject<string>();
        // listen for changes of the password
        WhenPropertyChanged
            .Where(name => nameof(PasswordBox) == name)
            .Select(d => PasswordBox)
            .Where(d => d != null)
            .Subscribe(box =>
            {
                box.PasswordChanged += (sender, args) => passwordHasChanged.OnNext(nameof(Password));
            });

        passwordHasChanged.Subscribe(d => Log.Debug("Password changed"));

Answer 1

假设我有一个班级，该班级有一个我想订阅的事件，并且我想随时都订阅该类的唯一实例。

public class EventSource
{
    public EventSource(long i)
    {
        DoSomething(i);
    }

    private async void DoSomething(long i)
    {
        await Task.Yield();
        while (true)
        {
            await Task.Delay(500);

            Tick?.Invoke(i);
        }
    }

    public event Action<long> Tick;
}

static void Main(string[] args)
{
    var o1 = Observable.Timer(TimeSpan.FromSeconds(2), TimeSpan.FromSeconds(2)).Select(i => new EventSource(i));

    var o2 = o1.SelectMany(i =>
    {
        return Observable.FromEvent<Action<long>, long>(handler => i.Tick += handler, handler => i.Tick -= handler);
    });

    o2.Subscribe(i => Console.WriteLine(i));

    Console.ReadKey();

即我在这里模拟目标对象每两秒钟更改一次的行为，对于每个新实例，我们从事件中产生IObservable。也就是说，o2是可观察的平面。

这里的问题是，我们每隔一个滴答声（这里每两秒钟，您的情况下每个属性都会发生变化），我们需要先取消订阅前一个实例，然后才订阅新实例。 
并且在上面的代码中，这并不仅仅因为在SelectMany内部只有实际实例，而没有上一个实例。即我们需要成对方法，它会产生像（i1，i2），（i2，i3）...
的对
但是，我没有找到执行此操作的现有方法
例如在这里
Trouble Implementing a Sliding Window in Rx
从那里找到的任何解决方案都比您在Subject上的示例更复杂。
所以，在我看来，在这种情况下，您可以写类似的东西
var passwordHasChanged = new Subject<string>();
var handler = new RoutedEventHandler((sender, args) => passwordHasChanged.OnNext(nameof(Password)));
var oldBox = null as PasswordBox;
    // listen for changes of the password
    WhenPropertyChanged
        .Where(name => nameof(PasswordBox) == name)
        .Select(d => PasswordBox)
        .Where(d => d != null)
        .Subscribe(box =>
        {
             if(oldBox != null)
                 oldBox.PasswordChanged -= handler;
              box.PasswordChanged += handler;
              oldBox = box;
        });

    passwordHasChanged.Subscribe(d => Log.Debug("Password changed"));

或者，就像我之前说的，您可以使用上面链接中的滑动窗口；但是最主要的是您需要取消订阅先前的实例。

Answer 2

如果只希望从最后一次输入密码中获取事件，请使用Switch。

[Switch适用于IObservable<IObservable<T>>，并且在获得较新的观测值时取消订阅先前的观测值。

   var passwordChanged = WhenPropertyChanged
            .Where(name => nameof(PasswordBox) == name)
            .Select(d => PasswordBox)
            .Where(d => d != null)
            .Select(box =>                
                Observable.FromEvent<RoutedEventHandler, RoutedEventArgs>(
                    handler => box.PasswordChanged += handler,
                    handler => box.PasswordChanged -= handler);
            )
            .Switch()
            .Select(d => nameof(Password));