TransactionalCommandProcessor.cs

using System;
using System.Collections.Generic;
using System.Linq;
using System.Transactions;
using Domain.Contracts;
using ElmahExtensions;
using Microsoft.Practices.ServiceLocation;

namespace Domain.Common
{
    /// <summary>
    /// Transactional POC Saga
    /// This should be applied when IEventStore is not a subscriber itself.
    /// </summary>
    public class TransactionalCommandProcessor : ICommandProcessor
    {
        private readonly IEventStore _eventStore;

        public TransactionalCommandProcessor(IEventStore eventStore)
        {
            _eventStore = eventStore;
            DomainEvents = new List<IDomainEvent>();
        }

        public List<IDomainEvent> DomainEvents { get; private set; }
        
        public void Process<TCommand>(TCommand command) where TCommand : CommandBase
        {
            StoreEventsFor(TransactionalExecution(EventedHandler(command.ProcessId, ExecuteCommand(command))));
        }

        private Func<List<IDomainEvent>> EventedHandler(Guid processId, Action commandExecution)
        {
            Func<List<IDomainEvent>> x = () => 
            {
                var subscriber = new DomainEventSubscriberForProcess(processId);
                DomainEventPublisher.Instance.Subscribe(subscriber);

                try
                {
                    commandExecution();
                }
                catch (Exception ex)
                {
                    CustomErrorSignal.Handle(ex);
                }
                finally
                {
                    DomainEventPublisher.Instance.UnSubscribe(subscriber);
                }

                return subscriber.DomainEvents;
            };

            return x;
        }

        private Action ExecuteCommand<TCommand>(TCommand command) where TCommand : CommandBase
        {
            var aggregateService = ServiceLocator.Current.GetInstance<ICommandHandler<TCommand>>();

            if (aggregateService == null)
                throw new ArgumentException(typeof(TCommand).FullName);

            if(command.ProcessId == Guid.Empty)
                throw new ArgumentException("ProcessId cannot be empty.");


            return () => aggregateService.When(command);
        }

        private void StoreEventsFor(List<IDomainEvent> resultedEvents)
        {
            DomainEvents.AddRange(resultedEvents);

            foreach (var domainEvent in resultedEvents)
            {
                _eventStore.Append(domainEvent);
            }
        }

        private List<IDomainEvent> TransactionalExecution(Func<List<IDomainEvent>> e)
        {
            List<IDomainEvent> domainEvents;

            using (var transactionScope = new TransactionScope(TransactionScopeOption.RequiresNew))
            {
               domainEvents = e();

               if(!domainEvents.Any(a => a is IDomainEventError))
                   transactionScope.Complete();
            }

            return domainEvents;
        }
    }
}