system.reactive - 处理事件日志中的事件并对特定模式使用react(Rx？)

Question

我需要处理来自 EventLog 的事件。这很容易使用 EventLogWatcher附加到 EventRecordWritten 事件。然而，我感兴趣的查询 (eventid == 299 || eventid == 500) 提供了比我需要的更多的事件。这是流的示例

Event ID   Correlation ID
299        1AD...  (this is actually a guid)  X1
500        1AD...                             X2
500        1AD...
500        1AD...
299        43B...                             Y1
299        EDB...                             Z1
500        43B...                             Y2
500        EDB...                             Z2
500        43B...
500        43B...

我对事件 299 和与 299 事件的相关性 ID 匹配的第一个事件 500 感兴趣。我在上面的流中标记了它们，这是我感兴趣的输出集合:[X1, X2, Y1, Y2, Z1, Z2]这可能是一个使用相关 ID 作为键和 EventRecord 元组的字典作为值(value)

{ 1AD.., <X1, X2> }
{ 43B.., <Y1, Y2> }
{ EDB.., <Z1, Z2> }

一般来说，事件可能按顺序出现(299 然后是 500)，但在高并发情况下，我预见两个 299 事件会聚集在一起，然后是 500 事件，所以我不想依赖事件发生的顺序。相关 id 是关联它们的关键(这是事件 eventRecord.Properties[0] 的第一个属性)

我认为这可以通过状态机来解决，但是看看是否有人提出了一个解决方案，其中 Rx 由对 observable 的查询表示。这会将模式匹配逻辑保留在一个地方。

更新 : 这是解决方案的答案。感谢 Gideon，这正是我需要的起点!

var pairs = events
            .Where(e299 => e299.EventArgs.EventRecord.Id == 299)
            .SelectMany(e299 => events.Where(e500 => e500.EventArgs.EventRecord.Id == 500 &&
                                                    e299.EventArgs.EventRecord.Properties[0].Value.ToString() ==
                                                    e500.EventArgs.EventRecord.Properties[0].Value.ToString())
                                    .Take(1), 
                        (e299, e500) => new { First = e299, Second = e500 });

提前致谢，
马蒂亚斯

Answer 1

如果 299 个事件总是在 500 个事件之前，SelectMany和 Where应该够了。

var events; //declared somewhere

var pairs = from e299 in events
            where e299.eventid == 299
            from e500 in events
            where e500.eventid == 500 &&
                  e299.Correlation == e500.Correlation
            select new with {Correlation = e299.Correlation,
                             First = e299,
                             Second = e500}

如果您的源对于每个相关的 299 事件有多个 500 事件，您可能需要切换到 lambda 语法并添加 Take(1)到第二次订阅。