Support custom aggregator

src/Marten.Testing/Events/Projections/AggregatorTests.cs

@@ -69,7 +69,7 @@ public void build_a_series_of_events()
                 .Add(new MembersJoined {Members = new string[] {"Gandalf", "Boromir", "Gimli", "Legolas"}})
                 .Add(new MembersDeparted() {Members = new string[] {"Frodo", "Sam"}});
 
-            var party = theAggregator.Build(stream.Events);
+            var party = theAggregator.Build(stream.Events, null);
 
             party.Name.ShouldBe("Destroy the Ring");
 

src/Marten/Events/AggregationQueryHandler.cs

@@ -11,13 +11,15 @@ namespace Marten.Events
 {
     internal class AggregationQueryHandler<T> : IQueryHandler<T> where T : class, new()
     {
-        private readonly Aggregator<T> _aggregator;
+        private readonly IAggregator<T> _aggregator;
         private readonly EventQueryHandler _inner;
+        private readonly IDocumentSession _session;
 
-        public AggregationQueryHandler(Aggregator<T> aggregator, EventQueryHandler inner)
+        public AggregationQueryHandler(IAggregator<T> aggregator, EventQueryHandler inner, IDocumentSession session = null)
         {
             _aggregator = aggregator;
             _inner = inner;
+            _session = session;
         }
 
         public void ConfigureCommand(NpgsqlCommand command)
@@ -31,14 +33,14 @@ public T Handle(DbDataReader reader, IIdentityMap map)
         {
             var @events = _inner.Handle(reader, map);
 
-            return _aggregator.Build(@events);
+            return _aggregator.Build(@events, _session);
         }
 
         public async Task<T> HandleAsync(DbDataReader reader, IIdentityMap map, CancellationToken token)
         {
             var @events = await _inner.HandleAsync(reader, map, token).ConfigureAwait(false);
 
-            return _aggregator.Build(@events);
+            return _aggregator.Build(@events, _session);
         }
     }
 }

src/Marten/Events/Event.cs

@@ -9,7 +9,7 @@ public interface IEvent
         int Version { get; set; }
         object Data { get; }
 
-        void Apply<TAggregate>(TAggregate state, Aggregator<TAggregate> aggregator)
+        void Apply<TAggregate>(TAggregate state, IAggregator<TAggregate> aggregator)
             where TAggregate : class, new();
     }
 
@@ -26,7 +26,7 @@ public Event(T data)
 
         object IEvent.Data => Data;
 
-        public virtual void Apply<TAggregate>(TAggregate state, Aggregator<TAggregate> aggregator)
+        public virtual void Apply<TAggregate>(TAggregate state, IAggregator<TAggregate> aggregator)
             where TAggregate : class, new()
         {
             aggregator.AggregatorFor<T>()?.Apply(state, Data);

src/Marten/Events/EventGraph.cs

@@ -208,11 +208,16 @@ IncludeJoin<TOther> IDocumentMapping.JoinToInclude<TOther>(JoinType joinType, ID
             throw new NotSupportedException();
         }
 
-        public Aggregator<T> AggregateFor<T>() where T : class, new()
+        public void AggregateFor<T>(IAggregator<T> aggregator) where T : class, new()
+        {
+            _aggregates.AddOrUpdate(typeof (T), aggregator, (type, previous) => aggregator);
+        }
+
+        public IAggregator<T> AggregateFor<T>() where T : class, new()
         {
             return _aggregates
                 .GetOrAdd(typeof (T), type => new Aggregator<T>())
-                .As<Aggregator<T>>();
+                .As<IAggregator<T>>();
         }
 
 
@@ -224,7 +229,7 @@ public Type AggregateTypeFor(string aggregateTypeName)
             }).AggregateType;
         }
 
-        public Aggregator<T> AggregateStreamsInlineWith<T>() where T : class, new()
+        public IAggregator<T> AggregateStreamsInlineWith<T>() where T : class, new()
         {
             var aggregator = AggregateFor<T>();
             var finder = new AggregateFinder<T>();

src/Marten/Events/EventStore.cs

@@ -83,7 +83,7 @@ public Task<IList<IEvent>> FetchStreamAsync(Guid streamId, int version = 0, Date
         {
             var inner = new EventQueryHandler(_selector, streamId, version, timestamp);
             var aggregator = _schema.Events.AggregateFor<T>();
-            var handler = new AggregationQueryHandler<T>(aggregator, inner);
+            var handler = new AggregationQueryHandler<T>(aggregator, inner, _session);
 
             return _connection.Fetch(handler, null);
         }
@@ -93,7 +93,7 @@ public Task<T> AggregateStreamAsync<T>(Guid streamId, int version = 0, DateTime?
         {
             var inner = new EventQueryHandler(_selector, streamId, version, timestamp);
             var aggregator = _schema.Events.AggregateFor<T>();
-            var handler = new AggregationQueryHandler<T>(aggregator, inner);
+            var handler = new AggregationQueryHandler<T>(aggregator, inner, _session);
 
             return _connection.FetchAsync(handler, null, token);
         }

src/Marten/Events/IEventStoreConfiguration.cs

@@ -17,9 +17,11 @@ public interface IEventStoreConfiguration
         IEnumerable<IAggregator> AllAggregates();
         EventMapping EventMappingFor(string eventType);
         bool IsActive { get; }
-        Aggregator<T> AggregateFor<T>() where T : class, new();
+
+        void AggregateFor<T>(IAggregator<T> aggregator) where T : class, new();
+        IAggregator<T> AggregateFor<T>() where T : class, new();
         Type AggregateTypeFor(string aggregateTypeName);
-        Aggregator<T> AggregateStreamsInlineWith<T>() where T : class, new();
+        IAggregator<T> AggregateStreamsInlineWith<T>() where T : class, new();
 
         void TransformEventsInlineWith<TEvent, TView>(ITransform<TEvent, TView> transform);
         void InlineTransformation(IProjection projection);

src/Marten/Events/Projections/AggregationProjection.cs

@@ -9,9 +9,9 @@ namespace Marten.Events.Projections
     public class AggregationProjection<T> : IProjection where T : class, new()
     {
         private readonly IAggregationFinder<T> _finder;
-        private readonly Aggregator<T> _aggregator;
+        private readonly IAggregator<T> _aggregator;
 
-        public AggregationProjection(IAggregationFinder<T> finder, Aggregator<T> aggregator)
+        public AggregationProjection(IAggregationFinder<T> finder, IAggregator<T> aggregator)
         {
             _finder = finder;
             _aggregator = aggregator;
@@ -52,6 +52,5 @@ public EventStream[] MatchingStreams(IDocumentSession session)
             return session.PendingChanges.AllChangedFor<EventStream>()
                 .Where(_aggregator.AppliesTo).ToArray();
         }
-
     }
 }

src/Marten/Events/Projections/Aggregator.cs

@@ -7,7 +7,7 @@
 
 namespace Marten.Events.Projections
 {
-    public class Aggregator<T> : IAggregator where T : class, new()
+    public class Aggregator<T> : IAggregator<T> where T : class, new()
     {
         public static readonly string ApplyMethod = "Apply";
 
@@ -34,7 +34,7 @@ public Aggregator()
 
         public string Alias { get; }
 
-        public T Build(IEnumerable<IEvent> events)
+        public T Build(IEnumerable<IEvent> events, IDocumentSession session)
         {
             var state = new T();
 

src/Marten/Events/Projections/IAggregator.cs

@@ -1,10 +1,18 @@
 using System;
+using System.Collections.Generic;
 
 namespace Marten.Events.Projections
 {
-    public interface IAggregator 
+    public interface IAggregator
     {
         Type AggregateType { get; }
         string Alias { get; }
+        bool AppliesTo(EventStream stream);
+    }
+
+    public interface IAggregator<T> : IAggregator
+    {
+        IAggregation<T, TEvent> AggregatorFor<TEvent>();
+        T Build(IEnumerable<IEvent> events, IDocumentSession session);
     }
 }

src/Marten/QuerySession.cs

@@ -9,7 +9,6 @@
 using Marten.Schema;
 using Marten.Services;
 using Marten.Services.BatchQuerying;
-using Marten.Util;
 using Npgsql;
 using Remotion.Linq.Parsing.Structure;
 

src/Marten/Services/BatchQuerying/BatchedQuery.cs

@@ -1,7 +1,6 @@
 using System;
 using System.Collections.Generic;
 using System.Linq;
-using System.Linq.Expressions;
 using System.Threading;
 using System.Threading.Tasks;
 using Baseline;