如何在完成之前从 ReplaySubject<T> 获取最新值

Question

我需要一种方法来获取添加到符合特定条件的 ReplaySubject 的最新项目。下面的示例代码完成了我需要它做的事情，但感觉不是正确的方法：

static void Main(string[] args)
{
    var o = new ReplaySubject<string>();

    o.OnNext("blueberry");
    o.OnNext("chimpanzee");
    o.OnNext("abacus");
    o.OnNext("banana");
    o.OnNext("apple");
    o.OnNext("cheese");

    var latest = o.Where(i => i.StartsWith("b"))
        .Latest().First();

    Console.WriteLine(latest);

    Console.WriteLine("Press any key to exit");
    Console.ReadKey();
}

输出：

banana
Press any key to exit

最初，我尝试使用.Where().TakeLast(1)；但是，我现在从以前的question 知道，您必须在TakeLast() 将返回任何内容之前就该主题调用OnComplete()。调用 OnComplete() 对我来说不是一个选项，因为我需要保持这个流打开。

谁能验证这是否是我想要完成的最有效的方法？谢谢！

编辑

请注意，我正在使用响应式扩展，IEnumerable 代码示例将无法正常工作。

更新

我倾向于使用以下代码，因为我相信它是非阻塞的，除非有人能以不同的方式告诉我：

var latest = o.Where(i => i.StartsWith("b")).Replay(1);

using (latest.Connect())
     latest.Subscribe(Console.WriteLine);

Answer 1

您可以考虑使用BehaviorSubject<string>。缺点是您必须在一开始就订阅，但这可能就是您想要做的。这应该为您提供所需的隔离。

var o = new ReplaySubject<string>();
var bs = new BehaviorSubject<string>(default(string));
o.Where(i => i.StartsWith("b")).Subscribe(bs);

o.OnNext("blueberry"); Console.WriteLine(bs.First());
o.OnNext("chimpanzee"); Console.WriteLine(bs.First());
o.OnNext("abacus"); Console.WriteLine(bs.First());
o.OnNext("banana"); Console.WriteLine(bs.First());
o.OnNext("apple"); Console.WriteLine(bs.First());
o.OnNext("cheese"); Console.WriteLine(bs.First());

输出：

blueberry
blueberry
blueberry
banana
banana
banana

Answer 2

只要您喜欢使用这些阻塞运算符（看起来像您），我会考虑使用 MostRecent 运算符。

static void Main(string[] args)
{
    var o = new ReplaySubject<string>();

    o.OnNext("blueberry");
    o.OnNext("chimpanzee");
    o.OnNext("abacus");
    o.OnNext("banana");
    o.OnNext("apple");
    o.OnNext("cheese");

    var latest = o.Where(i => i.StartsWith("b"))
        .MostRecent("SomeDefaultValue")
        .First();

    Console.WriteLine(latest);

    Console.WriteLine("Press any key to exit");
    Console.ReadKey();
}