Rx：移动中的分组事件计数 Window

Question

我已经开始考虑将 Reactive Extensions 与 EventStore 结合使用。作为概念证明，我想看看我是否可以让 Rx 使用事件流并在 window 一秒内输出按类型分组的事件计数。

所以，假设我正在使用名称为 "orders" 的流，我希望在控制台中看到如下内容：

OrderCreated 201
OrderUpdated 111

（一秒过去了..）

OrderCreated 123
OrderUpdated 132

等等。

到目前为止，我已经能够得到每秒 all 个事件计数的输出。但似乎无法按事件类型对它们进行分组。

我使用的代码基于 James Nugent 的gist：

internal class EventStoreRxSubscription
{
    public Subject<ResolvedEvent> ResolvedEvents { get; }

    public Subject<SubscriptionDropReason>  DroppedReasons { get; }
    public EventStoreSubscription Subscription { get; }

    public EventStoreRxSubscription(EventStoreSubscription subscription, Subject<ResolvedEvent> resolvedEvent, Subject<SubscriptionDropReason> droppedReasons)
    {
        Subscription = subscription;
        ResolvedEvents = resolvedEvent;
        DroppedReasons = droppedReasons;
    }
}

static class EventStoreConnectionExtensions
{
    public static Task<EventStoreRxSubscription> SubscribeTo(this IEventStoreConnection connection, string streamName, bool resolveLinkTos)
    {
        return Task<EventStoreRxSubscription>.Factory.StartNew(() => {

            var resolvedEvents = new Subject<ResolvedEvent>();
            var droppedReasons = new Subject<SubscriptionDropReason>();

            var subscriptionTask = connection.SubscribeToStreamAsync(streamName, resolveLinkTos, 
                                                                    (subscription, @event) => resolvedEvents.OnNext(@event), 
                                                                    (subscription, dropReason, arg3) => droppedReasons.OnNext(dropReason));
            subscriptionTask.Wait();

            return new EventStoreRxSubscription(subscriptionTask.Result, resolvedEvents, droppedReasons);
        });
    }
}

class Program
{
    static void Main(string[] args)
    {
         var connection = EventStoreConnection.Create(new IPEndPoint(IPAddress.Loopback, 1113));
         connection.ConnectAsync();

         var subscriptionTask = connection.SubscribeTo("orders", true);
         subscriptionTask.Wait();

         var events = subscriptionTask.Result.ResolvedEvents;

         var query = events.Timestamp()
                .Buffer(TimeSpan.FromSeconds(1))
                .Select(e => e.Count);

         query.Subscribe(Console.WriteLine);

         Console.ReadLine();
    }
 }

Answer 1

我之前做过类似的事情，我使用 Throttle 将所有事件分组在一个设定的频率内，但是你可以使用 Buffer 来获得每个周期的 count/collection .

下面的示例提供了我如何实现此目的的抽象示例，其中 AggregateType 和 AggregateFunction 将替换为您自己的类型和聚合。

GroupByUntil允许您在设定的时间段内按类型分组。

subscription = observable
    .GroupByUntil(e => e.Key, e => e.Buffer(TimeSpan.FromSeconds(1)))
    .SelectMany(e => e.Aggregate(new AggregateType(), (a, e) => AggregateFunction(a, e))
    .Subscribe(onNext, onError, onCompleted);

编辑

下面是我编写的一个简单示例，用于说明如何完成

public class EventType
{
    public string Type { get; set; }
}

public class AggregatedType
{
    public string EventType { get; set; }
    public int Count { get; set; }
}

class Program
{
    public delegate void ExampleEventHandler(EventType e);

    public static event ExampleEventHandler Event;

    static void Main(string[] args)
    {
        var subscription = Observable.FromEvent<ExampleEventHandler, EventType>(e => Event += e, e => Event -= e)
            .GroupByUntil(e => e.Type, e => e.Buffer(TimeSpan.FromSeconds(1)))
            .SelectMany(e => e
                .Select(ev => new AggregatedType {  EventType = ev.Type })
                .Aggregate(new AggregatedType(), (a, ev) => new AggregatedType { EventType = ev.EventType, Count = a.Count + 1 }))
            .Subscribe(OnAggregaredEvent, OnException, OnCompleted);

        Event(new EventType { Type = "A" });
        Event(new EventType { Type = "A" });
        Event(new EventType { Type = "B" });
        Event(new EventType { Type = "B" });

        SpinWait.SpinUntil(()=> false, TimeSpan.FromSeconds(2));

        Event(new EventType { Type = "A" });
        Event(new EventType { Type = "A" });
        Event(new EventType { Type = "B" });
        Event(new EventType { Type = "B" });

        Console.ReadLine();
    }

    static void OnAggregaredEvent(AggregatedType aggregated)
    {
        Console.WriteLine("Type: {0}, Count: {1}", aggregated.EventType, aggregated.Count);
    }

    static void OnException(Exception ex)
    {
    }

    static void OnCompleted()
    {
    }
}

Rx：移动中的分组事件计数 Window

Rx: Count of Grouped Events in Moving Window

c#

system.reactive

rx.net

eventstoredb