扫描、groupjoin 或 groupbyuntil？

Question

假设有如下图所示的一些问题和答案。答案总是对应于最近的问题。

QUESTIONS: A     B      C        D
ANSWERS:     1 2         3 4 5    6

打印出相应问题和答案的最佳方式是什么？下面是我使用 Scan 得到的解决方案。但是我想知道是否有更直观的方法使用 GroupJoin 来完成此操作。我尝试了 GroupJoin 但无法正常工作。我也很好奇 GroupBy 或 GroupByUntil 是否是合适的选项（但也无法使它们起作用）。最后，请问这种情况需要使用Publish吗？

static void Main(string[] args) {
    var questions = new Subject<string>();
    var answers = new Subject<int>();
    var results = Observable
        .CombineLatest(
            questions,
            answers.StartWith(-1),
            (q, a) => new { Question = q, Answer = a })
        .Scan((a, b) =>
            (a.Question == b.Question)
            ? b
            : new { Question = b.Question, Answer = -1 });

    results.Subscribe(i => Console.WriteLine((i.Answer == -1) ? i.Question : $"  {i.Answer}"));

    questions.OnNext("A");
    answers.OnNext(1);
    answers.OnNext(2);
    questions.OnNext("B");
    questions.OnNext("C");
    answers.OnNext(3);
    answers.OnNext(4);
    answers.OnNext(5);
    questions.OnNext("D");
    answers.OnNext(6);

    Console.ReadLine();
}

预期输出是这样的：

我越想，这个特性就越像数据库世界中的 LeftJoin 行为。我还能够使用类似于您在下面看到的代码使其工作。出于某种原因，它在控制台应用程序中不起作用。当我用我自己的 Versioned 函数（分配递增的 long 值）替换 Timestamp 函数时，它工作得很好。

var results = Observable
    .CombineLatest(questions.Timestamp(), answers.StartWith(-1).Timestamp(), (q, a) => new { Question = q, Answer = a })
    .Select(i => (i.Answer.Timestamp > i.Question.Timestamp)
        ? new { Question = i.Question.Value, Answer = i.Answer.Value }
        : new { Question = i.Question.Value, Answer = -1 });

Answer 1

以下是使用 GroupJoin

的方法

var results = questions.GroupJoin(answers,
    s => questions,
    s => Observable.Empty<int>(),
    (q, a) => new { Question = q, Answers = a });

results2
    .Subscribe(i => 
    {
        Console.WriteLine(string.Format("{0}", i.Question));
        i.Answers.Subscribe(a => Console.WriteLine("  {0}", a));
    });

这会产生与您的代码相同的输出。

Answer 2

也许更愚蠢。我认为最简单的答案是只使用合并并将两个序列投影到一个通用类型。只需查看预期输出，您就会发现它只是一个合并序列。

var questions = new Subject<string>();
var answers = new Subject<int>();
var Qs = questions.Select(q => new { Question = q, Answer = -1});
var As = answers.Select(a => new { Question = (string)null, Answer = a});
Observable.Merge(Qs,As)
    .Subscribe(i => Console.WriteLine((i.Answer == -1) ? i.Question : $"  {i.Answer}"));

questions.OnNext("A");
answers.OnNext(1);
answers.OnNext(2);
questions.OnNext("B");
questions.OnNext("C");
answers.OnNext(3);
answers.OnNext(4);
answers.OnNext(5);
questions.OnNext("D");
answers.OnNext(6);

结果

扫描、groupjoin 或 groupbyuntil？

scan, groupjoin, or groupbyuntil?

c#

system.reactive

observable