如何观察响应式扩展(Rx)中的依赖事件

Question

处理依赖事件的最佳方式是什么?
有一个对象，我需要测试连接是否成功或失败。但是对象首先需要通过我测试成功或失败的初始化步骤，然后继续连接步骤。

• 如果初始化失败返回连接失败。
• 如果初始化成功，返回连接步骤的结果。

我的代码如下。是否有更好的方法来处理这些依赖事件，因为我在初始化订阅中订阅了连接?

如果我有更多像这样的依赖事件，我将继续嵌套订阅吗?

public static void Test()
{
    const int maxValue = 501;
    var random = new Random(BitConverter.ToInt32(Guid.NewGuid().ToByteArray(), 0));
    var initOk = Observable.Interval(TimeSpan.FromMilliseconds(random.Next(maxValue))).Select(i => true);
    var initKo = Observable.Interval(TimeSpan.FromMilliseconds(random.Next(maxValue))).Select(i => false);
    var connectOk = Observable.Interval(TimeSpan.FromMilliseconds(random.Next(maxValue))).Select(i => true);
    var connectKo = Observable.Interval(TimeSpan.FromMilliseconds(random.Next(maxValue))).Select(i => false);
    var initResult = initOk.Amb(initKo).Take(1);
    var connectResult = connectOk.Amb(connectKo).Take(1);
    var id =
        initResult.Subscribe(ir =>
                                 {
                                     if (ir)
                                     {
                                         var cd =
                                             connectResult.Subscribe(cr =>
                                                                         {
                                                                             Console.WriteLine(cr
                                                                                                   ? "Connection succeeded."
                                                                                                   : "Connection failed.");
                                                                         });
                                     }
                                     else
                                     {
                                         Console.WriteLine("Initialization failed thus connection failed.");
                                     }
                                 });
}

Answer 1

通常可以通过使用各种rx操作符来链式调用来避免嵌套。

你的例子可以使用:

initResult.SelectMany(ir =>
       {   
           if (ir != null)
           {
             return connectResult;
           }
           Console.WriteLine("Initialization failed thus connection failed.");
           return Observable.Throw(new Exception("Some Exception"));
       })
       .Subscribe(cr =>
           {
              Console.WriteLine(cr != null
                 ? "Connection succeeded." 
                 : "Connection failed.");
           })

Answer 2

你可以这样写:

var finalResult =
    initResult
        .Select(ir =>
            Observable.If(() => ir, connectResult, Observable.Return(false)))
        .Merge();

你可以这样修改你的信息:

var initResultText =
    initResult
        .Select(ir =>
            ir ? (string)null : "Initialization failed thus connection failed.");
var connectResultText =
    connectResult
        .Select(cr => 
            String.Format("Connection {0}.", cr ? "succeeded" : "failed"));
var finalResult =
    initResultText
        .Select(irt =>
            Observable.If(() =>
                irt == null, connectResultText, Observable.Return(irt)))
        .Merge();

如果您需要在此基础上进一步嵌套，您应该考虑使用一个扩展方法来封装复杂性，从而使组合更容易。