5

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

因此，假设我正在使用名为“orders”的流，我希望在控制台中看到类似以下内容：

OrderCreated 201
OrderUpdated 111

（第二次通过..）

OrderCreated 123
OrderUpdated 132

等等。

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

我使用的代码是基于 James Nugent 的一个要点：

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();
    }
 }
4

1 回答 1

3

我之前做过类似的事情，我曾经Throttle将所有事件分组在一个设定的频率内，但是你可以使用它Buffer来获取每个时期的计数/收集。

下面的示例提供了一个抽象示例，说明了我是如何实现这一点的，在哪里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()
    {
    }
}
于 2016-02-29T11:22:46.570 回答