c# - Rx: 移動ウィンドウ内のグループ化されたイベントの数

Question

EventStore で Reactive Extensions を使用することを検討し始めました。概念実証として、Rx でイベント ストリームを消費し、1 秒間のウィンドウでタイプごとにグループ化されたイベントの数を出力できるかどうかを確認したいと思います。

たとえば、「orders」という名前のストリームを使用しているとします。コンソールに次のようなものが表示されることを確認したいと思います。

OrderCreated 201
OrderUpdated 111

(秒が過ぎます..)

OrderCreated 123
OrderUpdated 132

等々。

これまでのところ、 1 秒あたりのすべてのイベント数の出力を取得できました。しかし、イベントの種類ごとにグループ化できないようです。

私が使用しているコードは、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();
    }
 }