根据开始/结束事件对反应流中的事件进行分组

Question

我观察到一个反应流Observable<Event>，目前它直接发出它的事件。基于 BEGIN/END 事件，我想在内部组的支持下对这个事件流进行分组。

输入流

我有如下事件流：

Event(type = Data, groupId = 1)
Event(type = BeginGroup, groupId = 2)   // outer group begins
Event(type = Data, groupId = 2)
Event(type = BeginGroup, groupId = 3)   // inner group begins
Event(type = Data, groupId = 3)
Event(type = EndGroup, groupId = 3)     // inner group ends
Event(type = EndGroup, groupId = 2)     // outer group ends
Event(type = Data, groupId = 4)
Event(type = Data, groupId = 5)

编辑 - 附加前提条件：

我在示例数据中添加了 ID，但通常我不需要 ID。流将注意满足以下要求：

• 每个BeginGroup 事件都会在某个时间跟随相应的EndGroup 事件
• 组内的事件（在我的示例中具有相同的组理念）将始终在流内的开始/结束事件内，因此可以保证顺序（如上例中）李>

期望的输出流

所以我确保每个事件要么是它上面的组的一部分，要么如果它不是真实组的一部分，则具有唯一的 ID。我想将上述 9 个事件流分组为以下 4 个事件流：

Event(type = Data, groupId = 1)
GroupEvent(groupId = 2, data = <LIST of Events and/or sub groups>) with following data:
    data = [
        Event(type = BeginGroup, groupId = 2)
        Event(type = Data, groupId = 2)
        GroupEvent(groupId = 3, data = <LIST of Events and/or sub groups>) with following data:
            data = [
                Event(type = BeginGroup, groupId = 3)
                Event(type = Data, groupId = 3)
                Event(type = EndGroup, groupId = 3)
            ]
        Event(type = EndGroup, groupId = 2)
    ]
Event(type = Data, groupId = 4)
Event(type = Data, groupId = 5)

我想要什么 - 逻辑

我想在BeginGroup 类型的事件发生后立即开始分组，直到发生正确的EndGroup 事件，并将这两个事件之间的所有事件（包括最终嵌套的组事件）分组。开始/结束组事件之外的元素仅作为单个事件传递。

这是我目前尝试的方法

在问这个问题之前我尝试了一些东西，但我来自 java，我什至很少使用窗口/缓冲区操作符，而且对它们的经验也很少。我看到 C# 中有类似 GroupByUntil 的操作符，所以我尝试使用它，但在我的示例中它从不发出任何内容。

var eventObservable: Observable<Event> = ...

// 1) make the observable hot so it can be resued inside the groupbyuntil operator
eventObservable = observable.Publish().RefCount();
var res = eventObservable
    .GroupByUntil(
        e => e.GroupId, // selector for groups => the group id can be used here
        grp => eventObservable.Where(e => e.GroupId != grp.Key) // stop a group as soon as the group id changes
    )
    .SelectMany(data => data.ToList()) // flatten the observable
    .Select(data => {
        // Convert the list of Events to GroupEvent if it contains more than 1 event
        var list = data.ToList();
        if (list.Count == 1)
            return list[0];
        return new GroupEvent(list);
    })

这种方法不起作用，它根本不会发出任何东西（eventObservable 确实会正确发出它的项目）。此外，它缺少对嵌套组的支持（理论上是否可行）。

有人可以向我解释如何解决我的问题吗？

Answer 1

先代码转储，再解释。

我写的数据类：

public enum EventType
{
    Data,
    BeginGroup,
    EndGroup,
    Group
}

public class Event<T>
{
    public Event(int id, EventType type, T data)
    {
        this.Id = id;
        this.Type = type;
        this.Data = data;
    }

    public int Id { get; set; }
    public EventType Type { get; set; }
    public T Data { get; set;}
}

public class GroupEvent<T> : Event<T> {
    public GroupEvent(int id, IEnumerable<Event<T>> events)
        : base(id, EventType.Group, default(T))
    {
        this.ChildData = events;
    }

    public IEnumerable<Event<T>> ChildData { get; set; }
}

这是逻辑扩展方法（使用 Nuget 包System.Collections.Immutable）：

public static class X
{
    public static IObservable<Event<T>> GroupEvents<T>(this IObservable<Event<T>> source)
    {
        return source
            .Scan((groupId: 1, stack: ImmutableStack<ImmutableList<Event<T>>>.Empty, output: (Event<T>)null), (state, inEvent) =>
            {
                if(inEvent.Type == EventType.Data)
                {
                    if (state.stack.IsEmpty)
                        return (state.groupId + 1, state.stack, new Event<T>(state.groupId, EventType.Data, inEvent.Data));
                    else
                    {
                        var newEvent = new Event<T>(state.stack.Peek()[0].Id, EventType.Data, inEvent.Data);
                        var newList = state.stack.Peek().Add(newEvent);
                        var newStack = state.stack.Pop().Push(newList);
                        return (state.groupId, newStack, null);
                    }
                }

                if(inEvent.Type == EventType.BeginGroup)
                {
                    var newEvent = new Event<T>(state.groupId, EventType.BeginGroup, inEvent.Data);
                    return (state.groupId + 1, state.stack.Push(ImmutableList<Event<T>>.Empty.Add(newEvent)), null);
                }

                if (inEvent.Type == EventType.EndGroup)
                {
                    var newEvent = new Event<T>(state.stack.Peek()[0].Id, EventType.EndGroup, inEvent.Data);
                    var newList = state.stack.Peek().Add(newEvent);
                    var newStack = state.stack.Pop();
                    var toEmit = new GroupEvent<T>(newList[0].Id, newList);
                    if(newStack.IsEmpty)
                        return (state.groupId, newStack, toEmit);
                    else
                    {
                        var parentList = newStack.Peek().Add(toEmit);
                        newStack = newStack.Pop().Push(parentList);
                        return (state.groupId, newStack, null);
                    }
                }

                throw new NotImplementedException();
            })
            .Where(t => t.output != null)
            .Select(t => t.output);
    }
}

这是跑步者代码：

var s = new Subject<Event<int>>();
var o = s.GroupEvents();

s.OnNext(new Event<int>(-1, EventType.Data, 1));
s.OnNext(new Event<int>(-1, EventType.BeginGroup, 2));
s.OnNext(new Event<int>(-1, EventType.Data, 3));
s.OnNext(new Event<int>(-1, EventType.BeginGroup, 4));
s.OnNext(new Event<int>(-1, EventType.Data, 5));
s.OnNext(new Event<int>(-1, EventType.EndGroup, 6));
s.OnNext(new Event<int>(-1, EventType.EndGroup, 7));
s.OnNext(new Event<int>(-1, EventType.Data, 8));
s.OnNext(new Event<int>(-1, EventType.Data, 9));

结果看起来像您在问题中所期望的那样。

---

解释：

我采用了状态机方法，这通常意味着使用.Scan 方法。我们这里的状态是正在运行的groupId 计数，以及一堆消息列表。堆栈的顶部表示我们当前要向哪个组添加消息。由于Scan 不允许您区分状态和输出，因此我们对状态的第三个值是输出变量。

我正在使用不可变集合，因为它们与 Rx 配合得最好。如果你使用可变集合，性能可以提高，但你必须小心多重订阅的影响。

至于使用内置运算符（GroupBy、Buffer、Window、Join 等），我认为其中任何一个都不适用于树状递归你想要的结构。如果您对更扁平的结构感到满意，那么Window 可能会起作用，但它需要一些工作。