Line data    Source code

       1             : #region Copyright
       2             : // // -----------------------------------------------------------------------
       3             : // // <copyright company="cdmdotnet Limited">
       4             : // //   Copyright cdmdotnet Limited. All rights reserved.
       5             : // // </copyright>
       6             : // // -----------------------------------------------------------------------
       7             : #endregion
       8             : 
       9             : using System;
      10             : using System.Collections.Generic;
      11             : using System.Collections.ObjectModel;
      12             : using System.Linq;
      13             : using Akka.Actor;
      14             : using cdmdotnet.Logging;
      15             : using Cqrs.Authentication;
      16             : using Cqrs.Commands;
      17             : using Cqrs.Domain;
      18             : using Cqrs.Domain.Exceptions;
      19             : using Cqrs.Events;
      20             : using Cqrs.Infrastructure;
      21             : 
      22             : namespace Cqrs.Akka.Domain
      23             : {
      24             :         public abstract class AkkaAggregateRoot<TAuthenticationToken>
      25             :                 : ReceiveActor // PersistentActor 
      26             :                 , IAggregateRoot<TAuthenticationToken>
      27           0 :         {
      28             :                 protected IUnitOfWork<TAuthenticationToken> UnitOfWork { get; set; }
      29             : 
      30             :                 protected IAkkaAggregateRepository<TAuthenticationToken> Repository { get; set; }
      31             : 
      32             :                 protected ILogger Logger { get; set; }
      33             : 
      34             :                 protected ICorrelationIdHelper CorrelationIdHelper { get; set; }
      35             : 
      36             :                 protected IAuthenticationTokenHelper<TAuthenticationToken> AuthenticationTokenHelper { get; set; }
      37             : 
      38             :                 private ICollection<IEvent<TAuthenticationToken>> Changes { get; set; }
      39             : 
      40             :                 public Guid Id { get; protected set; }
      41             : 
      42             :                 public int Version { get; protected set; }
      43             : 
      44           0 :                 protected AkkaAggregateRoot(IUnitOfWork<TAuthenticationToken> unitOfWork, ILogger logger, IAkkaAggregateRepository<TAuthenticationToken> repository, ICorrelationIdHelper correlationIdHelper, IAuthenticationTokenHelper<TAuthenticationToken> authenticationTokenHelper)
      45             :                 {
      46             :                         UnitOfWork = unitOfWork;
      47             :                         Logger = logger;
      48             :                         Repository = repository;
      49             :                         CorrelationIdHelper = correlationIdHelper;
      50             :                         AuthenticationTokenHelper = authenticationTokenHelper;
      51             :                         Changes = new ReadOnlyCollection<IEvent<TAuthenticationToken>>(new List<IEvent<TAuthenticationToken>>());
      52             :                 }
      53             : 
      54             :                 #region Overrides of ActorBase
      55             : 
      56             :                 /// <summary>
      57             :                 /// User overridable callback.
      58             :                 ///                 <p/>
      59             :                 ///                 Is called when an Actor is started.
      60             :                 ///                 Actors are automatically started asynchronously when created.
      61             :                 ///                 Empty default implementation.
      62             :                 /// </summary>
      63           1 :                 protected override void PreStart()
      64             :                 {
      65             :                         base.PreStart();
      66             :                         Repository.LoadAggregateHistory(this, throwExceptionOnNoEvents: false);
      67             :                 }
      68             : 
      69             :                 #endregion
      70             : 
      71           0 :                 protected virtual void Execute<TCommand>(Action<TCommand> action, TCommand command)
      72             :                         where TCommand : ICommand<TAuthenticationToken>
      73             :                 {
      74             :                         UnitOfWork.Add(this);
      75             :                         try
      76             :                         {
      77             :                                 AuthenticationTokenHelper.SetAuthenticationToken(command.AuthenticationToken);
      78             :                                 CorrelationIdHelper.SetCorrelationId(command.CorrelationId);
      79             :                                 action(command);
      80             : 
      81             :                                 UnitOfWork.Commit();
      82             : 
      83             :                                 Sender.Tell(true, Self);
      84             :                         }
      85             :                         catch(Exception exception)
      86             :                         {
      87             :                                 Logger.LogError("Executing an Akka.net request failed.", exception: exception, metaData: new Dictionary<string, object> { { "Type", GetType() }, { "Command", command } });
      88             :                                 Sender.Tell(false, Self);
      89             :                                 throw;
      90             :                         }
      91             :                 }
      92             : 
      93           0 :                 public IEnumerable<IEvent<TAuthenticationToken>> GetUncommittedChanges()
      94             :                 {
      95             :                         return Changes;
      96             :                 }
      97             : 
      98           0 :                 public virtual void MarkChangesAsCommitted()
      99             :                 {
     100             :                         Version = Version + Changes.Count;
     101             :                         Changes = new ReadOnlyCollection<IEvent<TAuthenticationToken>>(new List<IEvent<TAuthenticationToken>>());
     102             :                 }
     103             : 
     104           0 :                 public virtual void LoadFromHistory(IEnumerable<IEvent<TAuthenticationToken>> history)
     105             :                 {
     106             :                         Type aggregateType = GetType();
     107             :                         foreach (IEvent<TAuthenticationToken> @event in history.OrderBy(e =>e.Version))
     108             :                         {
     109             :                                 if (@event.Version != Version + 1)
     110             :                                         throw new EventsOutOfOrderException(@event.Id, aggregateType, Version + 1, @event.Version);
     111             :                                 ApplyChange(@event, true);
     112             :                         }
     113             :                 }
     114             : 
     115           0 :                 protected virtual void ApplyChange(IEvent<TAuthenticationToken> @event)
     116             :                 {
     117             :                         ApplyChange(@event, false);
     118             :                 }
     119             : 
     120             :                 private void ApplyChange(IEvent<TAuthenticationToken> @event, bool isEventReplay)
     121             :                 {
     122             :                         this.AsDynamic().Apply(@event);
     123             :                         if (!isEventReplay)
     124             :                         {
     125             :                                 Changes = new ReadOnlyCollection<IEvent<TAuthenticationToken>>(new []{@event}.Concat(Changes).ToList());
     126             :                         }
     127             :                         else
     128             :                         {
     129             :                                 Id = @event.Id;
     130             :                                 Version++;
     131             :                         }
     132             :                 }
     133             :         }
     134             : }