按特定顺序查找消息

如何解决按特定顺序查找消息

我已经开始研究 ReactiveX，但不知道它是否适合我正在尝试解决的问题，因为要么我现在对 ReactiveX 不够了解，要么它没有我想要的需要。

假设我不断收到可能有 20 种不同类型的消息。所有消息都应首先保存到数据库中。然后我需要一些进一步的分析。我对按该顺序出现的 A、B、C 和 D 类型感兴趣（不必一个接一个）。当消息 A 出现时，应将其视为我需要触发的流程的开始。然后我应该等待消息 B（等待时任何其他消息类型都可以到达）到达并执行流程中的步骤。在消息 B 之后，我等待消息 C 并在过程中执行步骤。然后我等待标志着进程结束的消息 D。然后我需要重新开始并等待启动新进程的消息 A。

我正在使用 .NET，但是来自任何平台的代码可能都可以弄清楚如何（或是否）可以做到这一点。

更新：提供更多背景信息

使用@Enigmativity 示例代码，我将尝试稍微扩展这个问题。消息由设备产生。所以让我们假设在“A1,B2,B1,C1,F3,....”流中的第一个字母是消息类型，数字是设备的 ID。因此消息 A、B、C 和 D 需要属于同一设备才能被视为匹配。服务器总是会收到所有消息，因为设备会重复这些消息直到得到确认。这是单个设备可以产生的（流可以包含来自所有设备的混合消息）：

A1,H1,F1,A1 - 这里设备在完成任何操作之前重启，首先 A1,B1 应该被忽略，我们现在重新开始等待 A、B、C 和 D。

A1,B1 - 这不可能发生。 A1 总是在 B、C 或 D 之前。有时它可能不会到达 D，但它会重新开始。

解决方法

就 Rx.NET 的 API 跟踪 RxJS 而言，这很简单。假设我们已经拥有所有消息的 observable：

const ofType = theType => filter(({type}) => type === theType);

const a$ = messages$.pipe(ofType('a'));
const b$ = messages$.pipe(ofType('b'));
const c$ = messages$.pipe(ofType('c'));
const d$ = messages$.pipe(ofType('d'));

const handleB$ = b$.pipe(take(1),concatMap(bMsg => /* do the b step */));
const handleC$ = c$.pipe(take(1),concatMap(cMsg => /* do the c step */));
const waitForD$ = d$.pipe(take(1));

const process$ = a$.pipe(
  // while we are handling this "a" message,ignore other "a" messages
  exhaustMap((aMsg) => {
    // these will execute sequentially. once complete,we go back to
    // listening for "a" messages
    return concat(handleB$,handleC$,waitForD$);
  })
);

请注意，process$ 是一个 observable，它将发出“b”和“c”步骤的结果，以及“d”消息。如果您愿意，可以忽略或抑制此输出。

---

我发现Rx.NET 可能缺少exhaustMap 实现。 Here's an SO question 解决这个问题。

,

我不确定，根据您的描述，是否有任何保证您将始终获得 A、B、C 和 D 中的每一种消息类型，而不会获得另一组或重叠的值。例如，如果第二个 A 出现在最后一个 D 之前，我有两种方法以防重新启动出现问题。

这是我的基本代码设置：

var subject = new Subject<string>();

IObservable<(string a,string b,string c,string d)> query = ...

query.Subscribe(x => Console.WriteLine($"{x.a} {x.b} {x.c} {x.d}"));

"A1,B1,A2,C1,F1,D1,A3,A4,B2,B3,A5,C2,B4,F2,D2,D3,C3,D3"
    .Split(',')
    .ToObservable()
    .Subscribe(subject);

以下是所有内容都按顺序排列且完美匹配的情况（尽管散布着其他类型的消息：

IObservable<(string a,string d)> query =
    subject
        .Do(x => { /* Save here */ })
        .Publish(ss =>
        {
            var ssa = ss.Where(s => s[0] == 'A');
            var ssb = ss.Where(s => s[0] == 'B');
            var ssc = ss.Where(s => s[0] == 'C');
            var ssd = ss.Where(s => s[0] == 'D');
            return Observable.When(
                ssa
                    .And(ssb)
                    .And(ssc)
                    .And(ssd)
                    .Then((a,b,c,d) => (a: a,b: b,c: c,d: d)));
        });

这个查询使用了 Rx 中非常强大但很少使用的模式/计划查询（又名连接）。

如果您确实遇到需要在消息乱序时重置的问题并且您需要最新的消息，那么我认为这有效：

IObservable<(string a,string d)> query =
    subject
        .Do(x => { /* Save here */ })
        .Publish(ss =>
                ss
                    .Where(s => s[0] == 'A')
                    .Select(sa => ss.Where(s => s[0] == 'B').Select(sb => (a: sa,b: sb)))
                    .Switch()
                    .Select(sab => ss.Where(s => s[0] == 'C').Select(sc => (a: sab.a,b: sab.b,c: sc)))
                    .Switch()
                    .Select(sabc => ss.Where(s => s[0] == 'D').Select(sd => (a: sabc.a,b: sabc.b,c: sabc.c,d: sd)))
                    .Switch());

第一个查询给出了这个：

A1 B1 C1 D1
A2 B2 C2 D2
A3 B3 C3 D3

一切都很好，配对。

第二个给出了这个：

A1 B1 C1 D1
A4 B3 C2 D2
A4 B3 C2 D3
A5 B4 C3 D3