Merge pull request #126 from TransactionProcessing/task/#125_getlatestversionfromlastevent

Added GetLatestVersionFromLastEvent

Author: StuartFerguson

Shared.EventStore/Aggregate/AggregateRepository.cs

@@ -64,6 +64,29 @@ public async Task<TAggregate> GetLatestVersion(Guid aggregateId,
             return this.ProcessEvents(aggregate, resolvedEvents);
         }
 
+        /// <summary>
+        /// Gets the latest version from last event.
+        /// </summary>
+        /// <param name="aggregateId">The aggregate identifier.</param>
+        /// <param name="cancellationToken">The cancellation token.</param>
+        /// <returns></returns>
+        public async Task<TAggregate> GetLatestVersionFromLastEvent(Guid aggregateId,
+                                                                    CancellationToken cancellationToken)
+        {
+            TAggregate aggregate = new()
+                                   {
+                                       AggregateId = aggregateId
+                                   };
+
+            String streamName = AggregateRepository<TAggregate, TDomainEvent>.GetStreamName(aggregate.AggregateId);
+
+            IList<ResolvedEvent> events = await this.EventStoreContext.GetEventsBackwardAsync(streamName, 1, cancellationToken);
+
+            aggregate = this.ProcessEvents(aggregate, events);
+
+            return aggregate;
+        }
+
         /// <summary>
         /// Gets the name of the stream.
         /// </summary>

Shared.EventStore/Aggregate/IAggregateRepository.cs

@@ -28,6 +28,13 @@ Task<TAggregate> GetLatestVersion(Guid aggregateId,
         Task SaveChanges(TAggregate aggregate,
                          CancellationToken cancellationToken);
 
+        /// <summary>
+        /// Gets the latest version from last event.
+        /// </summary>
+        /// <param name="aggregateId">The aggregate identifier.</param>
+        /// <returns></returns>
+        Task<TAggregate> GetLatestVersionFromLastEvent(Guid aggregateId, CancellationToken cancellationToken);
+
         #endregion
     }
 }

Shared.EventStore/EventStore/EventStoreContext.cs

@@ -80,6 +80,38 @@ public async Task<String> GetPartitionResultFromProjection(String projectionName
 
         }
 
+        /// <summary>
+        /// Gets the events backwards asynchronous.
+        /// </summary>
+        /// <param name="streamName">Name of the stream.</param>
+        /// <param name="maxNumberOfEventsToRetrieve">The maximum number of events to retrieve.</param>
+        /// <param name="cancellationToken">The cancellation token.</param>
+        /// <returns></returns>
+        public async Task<IList<ResolvedEvent>> GetEventsBackwardAsync(String streamName,
+                                                                       Int32 maxNumberOfEventsToRetrieve,
+                                                                       CancellationToken cancellationToken)
+        {
+            List<ResolvedEvent> resolvedEvents = new();
+
+            EventStoreClient.ReadStreamResult response = this.EventStoreClient.ReadStreamAsync(Direction.Backwards,
+                                                                                              streamName,
+                                                                                              StreamPosition.End,
+                                                                                              maxNumberOfEventsToRetrieve,
+                                                                                              resolveLinkTos: true,
+                                                                                              cancellationToken: cancellationToken);
+
+            if (await response.ReadState == ReadState.StreamNotFound)
+            {
+                return resolvedEvents;
+            }
+
+            List<ResolvedEvent> events = await response.ToListAsync(cancellationToken);
+
+            resolvedEvents.AddRange(events);
+
+            return resolvedEvents;
+        }
+
         /// <summary>
         /// Gets the partition state from projection.
         /// </summary>
@@ -162,7 +194,7 @@ public async Task InsertEvents(String streamName,
             this.LogInformation($"About to append {aggregateEvents.Count} to Stream {streamName}");
             await this.EventStoreClient.AppendToStreamAsync(streamName, StreamRevision.FromInt64(expectedVersion), aggregateEvents.AsEnumerable(), cancellationToken: cancellationToken);
         }
-
+        
         /// <summary>
         /// Reads the events.
         /// </summary>

Shared.EventStore/EventStore/IEventStoreContext.cs

@@ -4,7 +4,6 @@
     using System.Collections.Generic;
     using System.Threading;
     using System.Threading.Tasks;
-    using DomainDrivenDesign.EventSourcing;
     using global::EventStore.Client;
 
     public interface IEventStoreContext
@@ -20,6 +19,17 @@ public interface IEventStoreContext
 
         #region Methods
 
+        /// <summary>
+        /// Gets the events backwards asynchronous.
+        /// </summary>
+        /// <param name="streamName">Name of the stream.</param>
+        /// <param name="maxNumberOfEventsToRetrieve">The maximum number of events to retrieve.</param>
+        /// <param name="cancellationToken">The cancellation token.</param>
+        /// <returns></returns>
+        Task<IList<ResolvedEvent>> GetEventsBackwardAsync(String streamName,
+                                                          Int32 maxNumberOfEventsToRetrieve,
+                                                          CancellationToken cancellationToken);
+
         /// <summary>
         /// Gets the partition result from projection.
         /// </summary>
@@ -96,8 +106,8 @@ Task InsertEvents(String streamName,
         /// <param name="cancellationToken">The cancellation token.</param>
         /// <returns></returns>
         Task<List<ResolvedEvent>> ReadEvents(String streamName,
-                                           Int64 fromVersion,
-                                           CancellationToken cancellationToken);
+                                             Int64 fromVersion,
+                                             CancellationToken cancellationToken);
 
         #endregion
     }