Documentation Coverage Report
Current view: top level - Cqrs/Bus - InProcessBus.cs Hit Total Coverage
Version: 2.2 Artefacts: 25 25 100.0 %
Date: 2017-09-22

          Line data    Source code
       1             : #region Copyright
       2             : // // -----------------------------------------------------------------------
       3             : // // <copyright company="Chinchilla Software Limited">
       4             : // //   Copyright Chinchilla Software Limited. All rights reserved.
       5             : // // </copyright>
       6             : // // -----------------------------------------------------------------------
       7             : #endregion
       8             : 
       9             : using System;
      10             : using System.Collections.Concurrent;
      11             : using System.Collections.Generic;
      12             : using System.Diagnostics;
      13             : using System.Linq;
      14             : using cdmdotnet.Logging;
      15             : using Cqrs.Authentication;
      16             : using Cqrs.Commands;
      17             : using Cqrs.Configuration;
      18             : using Cqrs.Events;
      19             : using Cqrs.Messages;
      20             : using SpinWait = Cqrs.Infrastructure.SpinWait;
      21             : 
      22             : namespace Cqrs.Bus
      23             : {
      24             :         /// <summary>
      25             :         /// An in process command bus 
      26             :         /// (<see cref="ICommandPublisher{TAuthenticationToken}"/> and <see cref="ICommandReceiver{TAuthenticationToken}"/>)
      27             :         /// event bus
      28             :         /// (<see cref="IEventPublisher{TAuthenticationToken}"/> and <see cref="IEventHandler{TAuthenticationToken,TTarget,TEvent}"/>)
      29             :         /// as well as a <see cref="IEventHandlerRegistrar"/> and <see cref="ICommandHandlerRegistrar"/> that requires no networking.
      30             :         /// </summary>
      31             :         /// <typeparam name="TAuthenticationToken">The <see cref="Type"/> of the authentication token.</typeparam>
      32             :         public class InProcessBus<TAuthenticationToken>
      33             :                 : IPublishAndWaitCommandPublisher<TAuthenticationToken>
      34             :                 , IEventPublisher<TAuthenticationToken>
      35             :                 , IEventHandlerRegistrar
      36             :                 , ICommandHandlerRegistrar
      37             :                 , ICommandReceiver<TAuthenticationToken>
      38             :                 , IEventReceiver<TAuthenticationToken>
      39           1 :         {
      40             :                 /// <summary>
      41             :                 /// Gets or sets the Route Manager
      42             :                 /// </summary>
      43             :                 private static RouteManager Routes { get; set; }
      44             : 
      45             :                 /// <summary>
      46             :                 /// Gets or sets the Authentication Token Helper
      47             :                 /// </summary>
      48             :                 protected IAuthenticationTokenHelper<TAuthenticationToken> AuthenticationTokenHelper { get; private set; }
      49             : 
      50             :                 /// <summary>
      51             :                 /// Gets or sets the CorrelationId Helper
      52             :                 /// </summary>
      53             :                 protected ICorrelationIdHelper CorrelationIdHelper { get; private set; }
      54             : 
      55             :                 /// <summary>
      56             :                 /// Gets or sets the Dependency Resolver
      57             :                 /// </summary>
      58             :                 protected IDependencyResolver DependencyResolver { get; private set; }
      59             : 
      60             :                 /// <summary>
      61             :                 /// Gets or sets the Logger
      62             :                 /// </summary>
      63             :                 protected ILogger Logger { get; private set; }
      64             : 
      65             :                 /// <summary>
      66             :                 /// Gets or sets the Configuration Manager
      67             :                 /// </summary>
      68             :                 protected IConfigurationManager ConfigurationManager { get; private set; }
      69             : 
      70             :                 /// <summary>
      71             :                 /// Gets or sets the Bus Helper
      72             :                 /// </summary>
      73             :                 protected IBusHelper BusHelper { get; private set; }
      74             : 
      75             :                 /// <summary>
      76             :                 /// Gets or sets the current list of events waiting to be evaluated for <see cref="PublishAndWait{TCommand,TEvent}(TCommand,Cqrs.Events.IEventReceiver{TAuthenticationToken})"/>
      77             :                 /// </summary>
      78             :                 protected IDictionary<Guid, IList<IEvent<TAuthenticationToken>>> EventWaits { get; private set; }
      79             : 
      80             :                 /// <summary>
      81             :                 /// Gets or sets the Telemetry Helper
      82             :                 /// </summary>
      83             :                 protected ITelemetryHelper TelemetryHelper { get; set; }
      84             : 
      85             :                 static InProcessBus()
      86             :                 {
      87             :                         Routes = new RouteManager();
      88             :                 }
      89             : 
      90             :                 /// <summary>
      91             :                 /// Instantiates a new instance of the <see cref="InProcessBus{TAuthenticationToken}"/> class.
      92             :                 /// </summary>
      93           1 :                 public InProcessBus(IAuthenticationTokenHelper<TAuthenticationToken> authenticationTokenHelper, ICorrelationIdHelper correlationIdHelper, IDependencyResolver dependencyResolver, ILogger logger, IConfigurationManager configurationManager, IBusHelper busHelper)
      94             :                 {
      95             :                         AuthenticationTokenHelper = authenticationTokenHelper;
      96             :                         CorrelationIdHelper = correlationIdHelper;
      97             :                         DependencyResolver = dependencyResolver;
      98             :                         Logger = logger;
      99             :                         ConfigurationManager = configurationManager;
     100             :                         BusHelper = busHelper;
     101             :                         EventWaits = new ConcurrentDictionary<Guid, IList<IEvent<TAuthenticationToken>>>();
     102             :                         TelemetryHelper = configurationManager.CreateTelemetryHelper("Cqrs.InProcessBus.UseApplicationInsightTelemetryHelper", correlationIdHelper);
     103             :                 }
     104             : 
     105             :                 /// <summary>
     106             :                 /// Sets the
     107             :                 /// <see cref="IMessageWithAuthenticationToken{TAuthenticationToken}.AuthenticationToken"/>,
     108             :                 /// <see cref="IMessage.CorrelationId"/>,
     109             :                 /// <see cref="IMessage.OriginatingFramework"/> to "Built-In" and
     110             :                 /// adds a value of "Built-In" to the <see cref="IMessage.Frameworks"/>
     111             :                 /// if not already done so
     112             :                 /// </summary>
     113           1 :                 protected virtual void PrepareCommand<TCommand>(TCommand command)
     114             :                         where TCommand : ICommand<TAuthenticationToken>
     115             :                 {
     116             :                         if (command.AuthenticationToken == null || command.AuthenticationToken.Equals(default(TAuthenticationToken)))
     117             :                                 command.AuthenticationToken = AuthenticationTokenHelper.GetAuthenticationToken();
     118             :                         command.CorrelationId = CorrelationIdHelper.GetCorrelationId();
     119             : 
     120             :                         if (string.IsNullOrWhiteSpace(command.OriginatingFramework))
     121             :                                 command.OriginatingFramework = "Built-In";
     122             : 
     123             :                         var frameworks = new List<string>();
     124             :                         if (command.Frameworks != null)
     125             :                                 frameworks.AddRange(command.Frameworks);
     126             :                         frameworks.Add("Built-In");
     127             :                         command.Frameworks = frameworks;
     128             :                 }
     129             : 
     130             :                 /// <summary>
     131             :                 /// Locates a suitable <see cref="ICommandValidator{TAuthenticationToken,TCommand}"/> to validate the provided <paramref name="command"/> and validates the provided <paramref name="command"/> if one is located
     132             :                 /// Calls <see cref="PrepareCommand{TCommand}"/>
     133             :                 /// Checks if the provided <paramref name="command"/> is required to be processed
     134             :                 /// Locates a single <see cref="RouteHandlerDelegate">command handler</see> for the provided <paramref name="command"/>
     135             :                 /// </summary>
     136             :                 /// <returns>
     137             :                 /// False if a suitable <see cref="ICommandValidator{TAuthenticationToken,TCommand}"/> is located and the provided <paramref name="command"/> fails validation,
     138             :                 /// False if no <see cref="RouteHandlerDelegate">command handler</see> is found but the command isn't required to be handled,
     139             :                 /// True otherwise.
     140             :                 /// </returns>
     141           1 :                 protected virtual bool PrepareAndValidateCommand<TCommand>(TCommand command, out RouteHandlerDelegate commandHandler)
     142             :                         where TCommand : ICommand<TAuthenticationToken>
     143             :                 {
     144             :                         Type commandType = command.GetType();
     145             : 
     146             :                         if (command.Frameworks != null && command.Frameworks.Contains("Built-In"))
     147             :                         {
     148             :                                 Logger.LogInfo("The provided command has already been processed by the Built-In bus.", string.Format("{0}\\PrepareAndValidateEvent({1})", GetType().FullName, commandType.FullName));
     149             :                                 commandHandler = null;
     150             :                                 return false;
     151             :                         }
     152             : 
     153             :                         ICommandValidator<TAuthenticationToken, TCommand> commandValidator = null;
     154             :                         try
     155             :                         {
     156             :                                 commandValidator = DependencyResolver.Resolve<ICommandValidator<TAuthenticationToken, TCommand>>();
     157             :                         }
     158             :                         catch (Exception exception)
     159             :                         {
     160             :                                 Logger.LogDebug("Locating an ICommandValidator failed.", string.Format("{0}\\Handle({1})", GetType().FullName, commandType.FullName), exception);
     161             :                         }
     162             : 
     163             :                         if (commandValidator != null && !commandValidator.IsCommandValid(command))
     164             :                         {
     165             :                                 Logger.LogInfo("The provided command is not valid.", string.Format("{0}\\Handle({1})", GetType().FullName, commandType.FullName));
     166             :                                 commandHandler = null;
     167             :                                 return false;
     168             :                         }
     169             : 
     170             :                         PrepareCommand(command);
     171             : 
     172             :                         bool isRequired = BusHelper.IsEventRequired(commandType);
     173             : 
     174             :                         commandHandler = Routes.GetSingleHandler(command, isRequired);
     175             :                         // This check doesn't require an isRequired check as there will be an exception raised above and handled below.
     176             :                         if (commandHandler == null)
     177             :                         {
     178             :                                 Logger.LogDebug(string.Format("The command handler for '{0}' is not required.", commandType.FullName));
     179             :                                 return false;
     180             :                         }
     181             : 
     182             :                         return true;
     183             :                 }
     184             : 
     185             :                 #region Implementation of ICommandSender<TAuthenticationToken>
     186             : 
     187             :                 /// <summary>
     188             :                 /// Publishes the provided <paramref name="command"/> on the command bus.
     189             :                 /// </summary>
     190             :                 void ICommandPublisher<TAuthenticationToken>.Publish<TCommand>(TCommand command)
     191             :                 {
     192             :                         Send(command);
     193             :                 }
     194             : 
     195             :                 /// <summary>
     196             :                 /// Publishes the provided <paramref name="command"/> on the command bus.
     197             :                 /// </summary>
     198           1 :                 public virtual void Send<TCommand>(TCommand command)
     199             :                         where TCommand : ICommand<TAuthenticationToken>
     200             :                 {
     201             :                         DateTimeOffset startedAt = DateTimeOffset.UtcNow;
     202             :                         Stopwatch mainStopWatch = Stopwatch.StartNew();
     203             :                         string responseCode = "200";
     204             :                         bool wasSuccessfull = false;
     205             : 
     206             :                         IDictionary<string, string> telemetryProperties = new Dictionary<string, string> { { "Type", "InProcessBus" } };
     207             :                         string telemetryName = string.Format("{0}/{1}", command.GetType().FullName, command.Id);
     208             :                         var telemeteredCommand = command as ITelemeteredMessage;
     209             :                         if (telemeteredCommand != null)
     210             :                                 telemetryName = telemeteredCommand.TelemetryName;
     211             :                         telemetryName = string.Format("Command/{0}", telemetryName);
     212             : 
     213             :                         try
     214             :                         {
     215             :                                 RouteHandlerDelegate commandHandler;
     216             :                                 if (!PrepareAndValidateCommand(command, out commandHandler))
     217             :                                         return;
     218             : 
     219             :                                 try
     220             :                                 {
     221             :                                         Action<IMessage> handler = commandHandler.Delegate;
     222             :                                         handler(command);
     223             :                                 }
     224             :                                 catch (Exception exception)
     225             :                                 {
     226             :                                         responseCode = "500";
     227             :                                         Logger.LogError("An issue occurred while trying to publish a command.", exception: exception, metaData: new Dictionary<string, object> {{"Command", command}});
     228             :                                         throw;
     229             :                                 }
     230             : 
     231             :                                 Logger.LogInfo(string.Format("A command was sent of type {0}.", command.GetType().FullName));
     232             :                                 wasSuccessfull = true;
     233             :                         }
     234             :                         finally
     235             :                         {
     236             :                                 mainStopWatch.Stop();
     237             :                                 TelemetryHelper.TrackDependency("InProcessBus/CommandBus", "Command", telemetryName, null, startedAt, mainStopWatch.Elapsed, responseCode, wasSuccessfull, telemetryProperties);
     238             :                         }
     239             :                 }
     240             : 
     241             :                 /// <summary>
     242             :                 /// Publishes the provided <paramref name="commands"/> on the command bus.
     243             :                 /// </summary>
     244             :                 void ICommandPublisher<TAuthenticationToken>.Publish<TCommand>(IEnumerable<TCommand> commands)
     245             :                 {
     246             :                         Send(commands);
     247             :                 }
     248             : 
     249             :                 /// <summary>
     250             :                 /// Publishes the provided <paramref name="commands"/> on the command bus.
     251             :                 /// </summary>
     252           1 :                 public virtual void Send<TCommand>(IEnumerable<TCommand> commands)
     253             :                         where TCommand : ICommand<TAuthenticationToken>
     254             :                 {
     255             :                         IEnumerable<TCommand> sourceCommands = commands.ToList();
     256             : 
     257             :                         DateTimeOffset startedAt = DateTimeOffset.UtcNow;
     258             :                         Stopwatch mainStopWatch = Stopwatch.StartNew();
     259             :                         string responseCode = "500";
     260             :                         bool wasSuccessfull = false;
     261             : 
     262             :                         IDictionary<string, string> telemetryProperties = new Dictionary<string, string> { { "Type", "InProcessBus" } };
     263             :                         string telemetryName = "Commands";
     264             :                         string telemetryNames = string.Empty;
     265             :                         foreach (TCommand command in sourceCommands)
     266             :                         {
     267             :                                 string subTelemetryName = string.Format("{0}/{1}", command.GetType().FullName, command.Id);
     268             :                                 var telemeteredCommand = command as ITelemeteredMessage;
     269             :                                 if (telemeteredCommand != null)
     270             :                                         subTelemetryName = telemeteredCommand.TelemetryName;
     271             :                                 telemetryNames = string.Format("{0}{1},", telemetryNames, subTelemetryName);
     272             :                         }
     273             :                         if (telemetryNames.Length > 0)
     274             :                                 telemetryNames = telemetryNames.Substring(0, telemetryNames.Length - 1);
     275             :                         telemetryProperties.Add("Commands", telemetryNames);
     276             : 
     277             :                         try
     278             :                         {
     279             :                                 foreach (TCommand command in sourceCommands)
     280             :                                         Send(command);
     281             : 
     282             :                                 responseCode = "200";
     283             :                                 wasSuccessfull = true;
     284             :                         }
     285             :                         finally
     286             :                         {
     287             :                                 mainStopWatch.Stop();
     288             :                                 TelemetryHelper.TrackDependency("InProcessBus/CommandBus", "Command", telemetryName, null, startedAt, mainStopWatch.Elapsed, responseCode, wasSuccessfull, telemetryProperties);
     289             :                         }
     290             :                 }
     291             : 
     292             :                 #endregion
     293             : 
     294             :                 #region Implementation of ISendAndWaitCommandSender<TAuthenticationToken>
     295             : 
     296             :                 /// <summary>
     297             :                 /// Sends the provided <paramref name="command"></paramref> and waits for an event of <typeparamref name="TEvent"/>
     298             :                 /// </summary>
     299             :                 /// <param name="command">The <typeparamref name="TCommand"/> to send.</param>
     300             :                 /// <param name="eventReceiver">If provided, is the <see cref="IEventReceiver{TAuthenticationToken}" /> that the event is expected to be returned on.</param>
     301           1 :                 public virtual TEvent SendAndWait<TCommand, TEvent>(TCommand command, IEventReceiver<TAuthenticationToken> eventReceiver = null)
     302             :                         where TCommand : ICommand<TAuthenticationToken>
     303             :                 {
     304             :                         return SendAndWait<TCommand, TEvent>(command, -1, eventReceiver);
     305             :                 }
     306             : 
     307             :                 /// <summary>
     308             :                 /// Sends the provided <paramref name="command"></paramref> and waits for an event of <typeparamref name="TEvent"/> or exits if the specified timeout is expired.
     309             :                 /// </summary>
     310             :                 /// <param name="command">The <typeparamref name="TCommand"/> to send.</param>
     311             :                 /// <param name="millisecondsTimeout">The number of milliseconds to wait, or <see cref="F:System.Threading.Timeout.Infinite"/> (-1) to wait indefinitely.</param>
     312             :                 /// <param name="eventReceiver">If provided, is the <see cref="IEventReceiver{TAuthenticationToken}" /> that the event is expected to be returned on.</param>
     313           1 :                 public virtual TEvent SendAndWait<TCommand, TEvent>(TCommand command, int millisecondsTimeout, IEventReceiver<TAuthenticationToken> eventReceiver = null)
     314             :                         where TCommand : ICommand<TAuthenticationToken>
     315             :                 {
     316             :                         return SendAndWait(command, events => (TEvent)events.SingleOrDefault(@event => @event is TEvent), millisecondsTimeout, eventReceiver);
     317             :                 }
     318             : 
     319             :                 /// <summary>
     320             :                 /// Sends the provided <paramref name="command"></paramref> and waits for an event of <typeparamref name="TEvent"/> or exits if the specified timeout is expired.
     321             :                 /// </summary>
     322             :                 /// <param name="command">The <typeparamref name="TCommand"/> to send.</param>
     323             :                 /// <param name="timeout">A <see cref="T:System.TimeSpan"/> that represents the number of milliseconds to wait, or a TimeSpan that represents -1 milliseconds to wait indefinitely.</param>
     324             :                 /// <param name="eventReceiver">If provided, is the <see cref="IEventReceiver{TAuthenticationToken}" /> that the event is expected to be returned on.</param>
     325           1 :                 public virtual TEvent SendAndWait<TCommand, TEvent>(TCommand command, TimeSpan timeout, IEventReceiver<TAuthenticationToken> eventReceiver = null)
     326             :                         where TCommand : ICommand<TAuthenticationToken>
     327             :                 {
     328             :                         long num = (long)timeout.TotalMilliseconds;
     329             :                         if (num < -1L || num > int.MaxValue)
     330             :                                 throw new ArgumentOutOfRangeException("timeout", timeout, "SpinWait_SpinUntil_TimeoutWrong");
     331             :                         return SendAndWait<TCommand, TEvent>(command, (int)timeout.TotalMilliseconds, eventReceiver);
     332             :                 }
     333             : 
     334             :                 /// <summary>
     335             :                 /// Sends the provided <paramref name="command"></paramref> and waits until the specified condition is satisfied an event of <typeparamref name="TEvent"/>
     336             :                 /// </summary>
     337             :                 /// <param name="command">The <typeparamref name="TCommand"/> to send.</param>
     338             :                 /// <param name="condition">A delegate to be executed over and over until it returns the <typeparamref name="TEvent"/> that is desired, return null to keep trying.</param>
     339             :                 /// <param name="eventReceiver">If provided, is the <see cref="IEventReceiver{TAuthenticationToken}" /> that the event is expected to be returned on.</param>
     340           1 :                 public virtual TEvent SendAndWait<TCommand, TEvent>(TCommand command, Func<IEnumerable<IEvent<TAuthenticationToken>>, TEvent> condition, IEventReceiver<TAuthenticationToken> eventReceiver = null)
     341             :                         where TCommand : ICommand<TAuthenticationToken>
     342             :                 {
     343             :                         return PublishAndWait(command, condition, eventReceiver);
     344             :                 }
     345             : 
     346             :                 /// <summary>
     347             :                 /// Sends the provided <paramref name="command"></paramref> and waits for an event of <typeparamref name="TEvent"/> or exits if the specified timeout is expired.
     348             :                 /// </summary>
     349             :                 /// <param name="command">The <typeparamref name="TCommand"/> to send.</param>
     350             :                 /// <param name="condition">A delegate to be executed over and over until it returns the <typeparamref name="TEvent"/> that is desired, return null to keep trying.</param>
     351             :                 /// <param name="millisecondsTimeout">The number of milliseconds to wait, or <see cref="F:System.Threading.Timeout.Infinite"/> (-1) to wait indefinitely.</param>
     352             :                 /// <param name="eventReceiver">If provided, is the <see cref="IEventReceiver{TAuthenticationToken}" /> that the event is expected to be returned on.</param>
     353           1 :                 public virtual TEvent SendAndWait<TCommand, TEvent>(TCommand command, Func<IEnumerable<IEvent<TAuthenticationToken>>, TEvent> condition, int millisecondsTimeout, IEventReceiver<TAuthenticationToken> eventReceiver = null)
     354             :                         where TCommand : ICommand<TAuthenticationToken>
     355             :                 {
     356             :                         return PublishAndWait(command, condition, millisecondsTimeout, eventReceiver);
     357             :                 }
     358             : 
     359             :                 /// <summary>
     360             :                 /// Sends the provided <paramref name="command"></paramref> and waits for an event of <typeparamref name="TEvent"/> or exits if the specified timeout is expired.
     361             :                 /// </summary>
     362             :                 /// <param name="command">The <typeparamref name="TCommand"/> to send.</param>
     363             :                 /// <param name="condition">A delegate to be executed over and over until it returns the <typeparamref name="TEvent"/> that is desired, return null to keep trying.</param>
     364             :                 /// <param name="timeout">A <see cref="T:System.TimeSpan"/> that represents the number of milliseconds to wait, or a TimeSpan that represents -1 milliseconds to wait indefinitely.</param>
     365             :                 /// <param name="eventReceiver">If provided, is the <see cref="IEventReceiver{TAuthenticationToken}" /> that the event is expected to be returned on.</param>
     366           1 :                 public virtual TEvent SendAndWait<TCommand, TEvent>(TCommand command, Func<IEnumerable<IEvent<TAuthenticationToken>>, TEvent> condition, TimeSpan timeout, IEventReceiver<TAuthenticationToken> eventReceiver = null)
     367             :                         where TCommand : ICommand<TAuthenticationToken>
     368             :                 {
     369             :                         return PublishAndWait(command, condition, timeout, eventReceiver);
     370             :                 }
     371             : 
     372             :                 #endregion
     373             : 
     374             :                 #region Implementation of IEventPublisher<TAuthenticationToken>
     375             : 
     376             :                 /// <summary>
     377             :                 /// Publishes the provided <paramref name="event"/> on the event bus.
     378             :                 /// </summary>
     379           1 :                 public virtual void Publish<TEvent>(TEvent @event)
     380             :                         where TEvent : IEvent<TAuthenticationToken>
     381             :                 {
     382             :                         Type eventType = @event.GetType();
     383             :                         string eventName = eventType.FullName;
     384             :                         ISagaEvent<TAuthenticationToken> sagaEvent = @event as ISagaEvent<TAuthenticationToken>;
     385             :                         if (sagaEvent != null)
     386             :                                 eventName = string.Format("Cqrs.Events.SagaEvent[{0}]", sagaEvent.Event.GetType().FullName);
     387             : 
     388             :                         DateTimeOffset startedAt = DateTimeOffset.UtcNow;
     389             :                         Stopwatch mainStopWatch = Stopwatch.StartNew();
     390             :                         string responseCode = "200";
     391             :                         bool wasSuccessfull = false;
     392             : 
     393             :                         IDictionary<string, string> telemetryProperties = new Dictionary<string, string> { { "Type", "InProcessBus" } };
     394             :                         string telemetryName = string.Format("{0}/{1}", eventName, @event.Id);
     395             :                         var telemeteredEvent = @event as ITelemeteredMessage;
     396             :                         if (telemeteredEvent != null)
     397             :                                 telemetryName = telemeteredEvent.TelemetryName;
     398             :                         telemetryName = string.Format("Event/{0}", telemetryName);
     399             : 
     400             :                         try
     401             :                         {
     402             :                                 if (@event.Frameworks != null && @event.Frameworks.Contains("Built-In"))
     403             :                                 {
     404             :                                         Logger.LogInfo("The provided event has already been processed by the Built-In bus.", string.Format("{0}\\PrepareAndValidateEvent({1})", GetType().FullName, eventType.FullName));
     405             :                                         return;
     406             :                                 }
     407             : 
     408             :                                 if (@event.AuthenticationToken == null || @event.AuthenticationToken.Equals(default(TAuthenticationToken)))
     409             :                                         @event.AuthenticationToken = AuthenticationTokenHelper.GetAuthenticationToken();
     410             :                                 @event.CorrelationId = CorrelationIdHelper.GetCorrelationId();
     411             : 
     412             :                                 if (string.IsNullOrWhiteSpace(@event.OriginatingFramework))
     413             :                                 {
     414             :                                         @event.TimeStamp = DateTimeOffset.UtcNow;
     415             :                                         @event.OriginatingFramework = "Built-In";
     416             :                                 }
     417             : 
     418             :                                 var frameworks = new List<string>();
     419             :                                 if (@event.Frameworks != null)
     420             :                                         frameworks.AddRange(@event.Frameworks);
     421             :                                 frameworks.Add("Built-In");
     422             :                                 @event.Frameworks = frameworks;
     423             : 
     424             :                                 bool isRequired;
     425             :                                 if (!ConfigurationManager.TryGetSetting(string.Format("{0}.IsRequired", eventName), out isRequired))
     426             :                                         isRequired = true;
     427             : 
     428             :                                 IEnumerable<Action<IMessage>> handlers = Routes.GetHandlers(@event, isRequired).Select(x => x.Delegate).ToList();
     429             :                                 // This check doesn't require an isRequired check as there will be an exception raised above and handled below.
     430             :                                 if (!handlers.Any())
     431             :                                         Logger.LogDebug(string.Format("An event handler for '{0}' is not required.", eventName));
     432             : 
     433             :                                 foreach (Action<IMessage> handler in handlers)
     434             :                                 {
     435             :                                         IList<IEvent<TAuthenticationToken>> events;
     436             :                                         if (EventWaits.TryGetValue(@event.CorrelationId, out events))
     437             :                                                 events.Add(@event);
     438             :                                         handler(@event);
     439             :                                 }
     440             : 
     441             :                                 Logger.LogInfo(string.Format("An event was sent of type {0}.", eventName));
     442             :                                 wasSuccessfull = true;
     443             :                         }
     444             :                         catch (Exception exception)
     445             :                         {
     446             :                                 responseCode = "500";
     447             :                                 Logger.LogError("An issue occurred while trying to publish an event.", exception: exception, metaData: new Dictionary<string, object> { { "Event", @event } });
     448             :                                 throw;
     449             :                         }
     450             :                         finally
     451             :                         {
     452             :                                 mainStopWatch.Stop();
     453             :                                 TelemetryHelper.TrackDependency("InProcessBus/EventBus", "Event", telemetryName, null, startedAt, mainStopWatch.Elapsed, responseCode, wasSuccessfull, telemetryProperties);
     454             :                         }
     455             :                 }
     456             : 
     457             :                 /// <summary>
     458             :                 /// Publishes the provided <paramref name="events"/> on the event bus.
     459             :                 /// </summary>
     460           1 :                 public virtual void Publish<TEvent>(IEnumerable<TEvent> events)
     461             :                         where TEvent : IEvent<TAuthenticationToken>
     462             :                 {
     463             :                         IEnumerable<TEvent> sourceEvents = events.ToList();
     464             : 
     465             :                         DateTimeOffset startedAt = DateTimeOffset.UtcNow;
     466             :                         Stopwatch mainStopWatch = Stopwatch.StartNew();
     467             :                         string responseCode = "500";
     468             :                         bool wasSuccessfull = false;
     469             : 
     470             :                         IDictionary<string, string> telemetryProperties = new Dictionary<string, string> { { "Type", "InProcessBus" } };
     471             :                         string telemetryName = "Events";
     472             :                         string telemetryNames = string.Empty;
     473             :                         foreach (TEvent @event in sourceEvents)
     474             :                         {
     475             :                                 string subTelemetryName = string.Format("{0}/{1}", @event.GetType().FullName, @event.Id);
     476             :                                 var telemeteredCommand = @event as ITelemeteredMessage;
     477             :                                 if (telemeteredCommand != null)
     478             :                                         subTelemetryName = telemeteredCommand.TelemetryName;
     479             :                                 telemetryNames = string.Format("{0}{1},", telemetryNames, subTelemetryName);
     480             :                         }
     481             :                         if (telemetryNames.Length > 0)
     482             :                                 telemetryNames = telemetryNames.Substring(0, telemetryNames.Length - 1);
     483             :                         telemetryProperties.Add("Events", telemetryNames);
     484             : 
     485             :                         try
     486             :                         {
     487             :                                 foreach (TEvent @event in sourceEvents)
     488             :                                         Publish(@event);
     489             : 
     490             :                                 responseCode = "200";
     491             :                                 wasSuccessfull = true;
     492             :                         }
     493             :                         finally
     494             :                         {
     495             :                                 mainStopWatch.Stop();
     496             :                                 TelemetryHelper.TrackDependency("InProcessBus/EventBus", "Event", telemetryName, null, startedAt, mainStopWatch.Elapsed, responseCode, wasSuccessfull, telemetryProperties);
     497             :                         }
     498             :                 }
     499             : 
     500             :                 #endregion
     501             : 
     502             :                 #region Implementation of IPublishAndWaitCommandPublisher<TAuthenticationToken>
     503             : 
     504             :                 /// <summary>
     505             :                 /// Publishes the provided <paramref name="command"></paramref> and waits for an event of <typeparamref name="TEvent"/>
     506             :                 /// </summary>
     507             :                 /// <param name="command">The <typeparamref name="TCommand"/> to publish.</param>
     508             :                 /// <param name="eventReceiver">If provided, is the <see cref="IEventReceiver{TAuthenticationToken}" /> that the event is expected to be returned on.</param>
     509           1 :                 public TEvent PublishAndWait<TCommand, TEvent>(TCommand command, IEventReceiver<TAuthenticationToken> eventReceiver = null) where TCommand : ICommand<TAuthenticationToken>
     510             :                 {
     511             :                         return PublishAndWait<TCommand, TEvent>(command, -1, eventReceiver);
     512             :                 }
     513             : 
     514             :                 /// <summary>
     515             :                 /// Publishes the provided <paramref name="command"></paramref> and waits for an event of <typeparamref name="TEvent"/> or exits if the specified timeout is expired.
     516             :                 /// </summary>
     517             :                 /// <param name="command">The <typeparamref name="TCommand"/> to publish.</param>
     518             :                 /// <param name="millisecondsTimeout">The number of milliseconds to wait, or <see cref="F:System.Threading.Timeout.Infinite"/> (-1) to wait indefinitely.</param>
     519             :                 /// <param name="eventReceiver">If provided, is the <see cref="IEventReceiver{TAuthenticationToken}" /> that the event is expected to be returned on.</param>
     520           1 :                 public TEvent PublishAndWait<TCommand, TEvent>(TCommand command, int millisecondsTimeout, IEventReceiver<TAuthenticationToken> eventReceiver = null) where TCommand : ICommand<TAuthenticationToken>
     521             :                 {
     522             :                         return PublishAndWait(command, events => (TEvent)events.SingleOrDefault(@event => @event is TEvent), millisecondsTimeout, eventReceiver);
     523             :                 }
     524             : 
     525             :                 /// <summary>
     526             :                 /// Publishes the provided <paramref name="command"></paramref> and waits for an event of <typeparamref name="TEvent"/> or exits if the specified timeout is expired.
     527             :                 /// </summary>
     528             :                 /// <param name="command">The <typeparamref name="TCommand"/> to publish.</param>
     529             :                 /// <param name="timeout">A <see cref="T:System.TimeSpan"/> that represents the number of milliseconds to wait, or a TimeSpan that represents -1 milliseconds to wait indefinitely.</param>
     530             :                 /// <param name="eventReceiver">If provided, is the <see cref="IEventReceiver{TAuthenticationToken}" /> that the event is expected to be returned on.</param>
     531           1 :                 public TEvent PublishAndWait<TCommand, TEvent>(TCommand command, TimeSpan timeout, IEventReceiver<TAuthenticationToken> eventReceiver = null) where TCommand : ICommand<TAuthenticationToken>
     532             :                 {
     533             :                         long num = (long)timeout.TotalMilliseconds;
     534             :                         if (num < -1L || num > int.MaxValue)
     535             :                                 throw new ArgumentOutOfRangeException("timeout", timeout, "SpinWait_SpinUntil_TimeoutWrong");
     536             :                         return PublishAndWait<TCommand, TEvent>(command, (int)timeout.TotalMilliseconds, eventReceiver);
     537             :                 }
     538             : 
     539             :                 /// <summary>
     540             :                 /// Publishes the provided <paramref name="command"></paramref> and waits until the specified condition is satisfied an event of <typeparamref name="TEvent"/>
     541             :                 /// </summary>
     542             :                 /// <param name="command">The <typeparamref name="TCommand"/> to publish.</param>
     543             :                 /// <param name="condition">A delegate to be executed over and over until it returns the <typeparamref name="TEvent"/> that is desired, return null to keep trying.</param>
     544             :                 /// <param name="eventReceiver">If provided, is the <see cref="IEventReceiver{TAuthenticationToken}" /> that the event is expected to be returned on.</param>
     545           1 :                 public TEvent PublishAndWait<TCommand, TEvent>(TCommand command, Func<IEnumerable<IEvent<TAuthenticationToken>>, TEvent> condition, IEventReceiver<TAuthenticationToken> eventReceiver = null) where TCommand : ICommand<TAuthenticationToken>
     546             :                 {
     547             :                         return PublishAndWait(command, condition, -1, eventReceiver);
     548             :                 }
     549             : 
     550             :                 /// <summary>
     551             :                 /// Publishes the provided <paramref name="command"></paramref> and waits for an event of <typeparamref name="TEvent"/> or exits if the specified timeout is expired.
     552             :                 /// </summary>
     553             :                 /// <param name="command">The <typeparamref name="TCommand"/> to publish.</param>
     554             :                 /// <param name="condition">A delegate to be executed over and over until it returns the <typeparamref name="TEvent"/> that is desired, return null to keep trying.</param>
     555             :                 /// <param name="millisecondsTimeout">The number of milliseconds to wait, or <see cref="F:System.Threading.Timeout.Infinite"/> (-1) to wait indefinitely.</param>
     556             :                 /// <param name="eventReceiver">If provided, is the <see cref="IEventReceiver{TAuthenticationToken}" /> that the event is expected to be returned on.</param>
     557           1 :                 public TEvent PublishAndWait<TCommand, TEvent>(TCommand command, Func<IEnumerable<IEvent<TAuthenticationToken>>, TEvent> condition, int millisecondsTimeout,
     558             :                         IEventReceiver<TAuthenticationToken> eventReceiver = null) where TCommand : ICommand<TAuthenticationToken>
     559             :                 {
     560             :                         DateTimeOffset startedAt = DateTimeOffset.UtcNow;
     561             :                         Stopwatch mainStopWatch = Stopwatch.StartNew();
     562             :                         string responseCode = "200";
     563             :                         bool wasSuccessfull = false;
     564             : 
     565             :                         IDictionary<string, string> telemetryProperties = new Dictionary<string, string> { { "Type", "InProcessBus" } };
     566             :                         string telemetryName = string.Format("{0}/{1}", command.GetType().FullName, command.Id);
     567             :                         var telemeteredCommand = command as ITelemeteredMessage;
     568             :                         if (telemeteredCommand != null)
     569             :                                 telemetryName = telemeteredCommand.TelemetryName;
     570             :                         telemetryName = string.Format("Command/{0}", telemetryName);
     571             : 
     572             :                         TEvent result;
     573             : 
     574             :                         try
     575             :                         {
     576             :                                 if (eventReceiver != null)
     577             :                                         throw new NotSupportedException("Specifying a different event receiver is not yet supported.");
     578             :                                 RouteHandlerDelegate commandHandler;
     579             :                                 if (!PrepareAndValidateCommand(command, out commandHandler))
     580             :                                         return (TEvent)(object)null;
     581             : 
     582             :                                 result = (TEvent)(object)null;
     583             :                                 EventWaits.Add(command.CorrelationId, new List<IEvent<TAuthenticationToken>>());
     584             : 
     585             :                                 Action<IMessage> handler = commandHandler.Delegate;
     586             :                                 handler(command);
     587             :                                 Logger.LogInfo(string.Format("A command was sent of type {0}.", command.GetType().FullName));
     588             :                                 wasSuccessfull = true;
     589             :                         }
     590             :                         finally
     591             :                         {
     592             :                                 mainStopWatch.Stop();
     593             :                                 TelemetryHelper.TrackDependency("InProcessBus/CommandBus", "Command", telemetryName, null, startedAt, mainStopWatch.Elapsed, responseCode, wasSuccessfull, telemetryProperties);
     594             :                         }
     595             : 
     596             :                         SpinWait.SpinUntil(() =>
     597             :                         {
     598             :                                 IList<IEvent<TAuthenticationToken>> events = EventWaits[command.CorrelationId];
     599             : 
     600             :                                 result = condition(events);
     601             : 
     602             :                                 return result != null;
     603             :                         }, millisecondsTimeout, SpinWait.DefaultSleepInMilliseconds);
     604             : 
     605             :                         TelemetryHelper.TrackDependency("InProcessBus/CommandBus", "Command/AndWait", string.Format("Command/AndWait{0}", telemetryName.Substring(7)), null, startedAt, mainStopWatch.Elapsed, responseCode, wasSuccessfull, telemetryProperties);
     606             :                         return result;
     607             :                 }
     608             : 
     609             :                 /// <summary>
     610             :                 /// Publishes the provided <paramref name="command"></paramref> and waits for an event of <typeparamref name="TEvent"/> or exits if the specified timeout is expired.
     611             :                 /// </summary>
     612             :                 /// <param name="command">The <typeparamref name="TCommand"/> to publish.</param>
     613             :                 /// <param name="condition">A delegate to be executed over and over until it returns the <typeparamref name="TEvent"/> that is desired, return null to keep trying.</param>
     614             :                 /// <param name="timeout">A <see cref="T:System.TimeSpan"/> that represents the number of milliseconds to wait, or a TimeSpan that represents -1 milliseconds to wait indefinitely.</param>
     615             :                 /// <param name="eventReceiver">If provided, is the <see cref="IEventReceiver{TAuthenticationToken}" /> that the event is expected to be returned on.</param>
     616           1 :                 public TEvent PublishAndWait<TCommand, TEvent>(TCommand command, Func<IEnumerable<IEvent<TAuthenticationToken>>, TEvent> condition, TimeSpan timeout,
     617             :                         IEventReceiver<TAuthenticationToken> eventReceiver = null) where TCommand : ICommand<TAuthenticationToken>
     618             :                 {
     619             :                         long num = (long)timeout.TotalMilliseconds;
     620             :                         if (num < -1L || num > int.MaxValue)
     621             :                                 throw new ArgumentOutOfRangeException("timeout", timeout, "SpinWait_SpinUntil_TimeoutWrong");
     622             :                         return PublishAndWait(command, condition, (int)timeout.TotalMilliseconds, eventReceiver);
     623             :                 }
     624             : 
     625             :                 #endregion
     626             : 
     627             :                 #region Implementation of IHandlerRegistrar
     628             : 
     629             :                 /// <summary>
     630             :                 /// Register an event or command handler that will listen and respond to events or commands.
     631             :                 /// </summary>
     632           1 :                 public virtual void RegisterHandler<TMessage>(Action<TMessage> handler, Type targetedType, bool holdMessageLock = true)
     633             :                         where TMessage : IMessage
     634             :                 {
     635             :                         Action<TMessage> registerableHandler = BusHelper.BuildTelemeteredActionHandler<TMessage, TAuthenticationToken>(TelemetryHelper, handler, holdMessageLock, "In-Process/Bus");
     636             : 
     637             :                         Routes.RegisterHandler(registerableHandler, targetedType);
     638             : 
     639             :                         TelemetryHelper.TrackEvent(string.Format("Cqrs/RegisterHandler/{0}", typeof(TMessage).FullName), new Dictionary<string, string> { { "Type", "In-Process/Bus" } });
     640             :                         TelemetryHelper.Flush();
     641             :                 }
     642             : 
     643             :                 /// <summary>
     644             :                 /// Register an event or command handler that will listen and respond to events or commands.
     645             :                 /// </summary>
     646           1 :                 public virtual void RegisterHandler<TMessage>(Action<TMessage> handler, bool holdMessageLock = true)
     647             :                         where TMessage : IMessage
     648             :                 {
     649             :                         RegisterHandler(handler, null, holdMessageLock);
     650             :                 }
     651             : 
     652             :                 /// <summary>
     653             :                 /// Register an event handler that will listen and respond to all events.
     654             :                 /// </summary>
     655           1 :                 public void RegisterGlobalEventHandler<TMessage>(Action<TMessage> handler, bool holdMessageLock = true) where TMessage : IMessage
     656             :                 {
     657             :                         Action<TMessage> registerableHandler = BusHelper.BuildTelemeteredActionHandler<TMessage, TAuthenticationToken>(TelemetryHelper, handler, holdMessageLock, "In-Process/Bus");
     658             : 
     659             :                         Routes.RegisterGlobalEventHandler(registerableHandler, holdMessageLock);
     660             : 
     661             :                         TelemetryHelper.TrackEvent(string.Format("Cqrs/RegisterGlobalEventHandler/{0}", typeof(TMessage).FullName), new Dictionary<string, string> { { "Type", "In-Process/Bus" } });
     662             :                         TelemetryHelper.Flush();
     663             :                 }
     664             : 
     665             :                 #endregion
     666             : 
     667             :                 #region Implementation of ICommandReceiver
     668             : 
     669             :                 /// <summary>
     670             :                 /// Receives a <see cref="ICommand{TAuthenticationToken}"/> from the command bus.
     671             :                 /// </summary>
     672           1 :                 public virtual bool? ReceiveCommand(ICommand<TAuthenticationToken> command)
     673             :                 {
     674             :                         Send(command);
     675             :                         return true;
     676             :                 }
     677             : 
     678             :                 /// <summary>
     679             :                 /// Receives an <see cref="IEvent{TAuthenticationToken}"/> from the event bus.
     680             :                 /// </summary>
     681           1 :                 public virtual bool? ReceiveEvent(IEvent<TAuthenticationToken> @event)
     682             :                 {
     683             :                         Publish(@event);
     684             :                         return true;
     685             :                 }
     686             : 
     687             :                 void ICommandReceiver.Start()
     688             :                 {
     689             :                         // This is in-process so doesn't need to do anything
     690             :                 }
     691             : 
     692             :                 #endregion
     693             : 
     694             :                 #region Implementation of IEventReceiver
     695             : 
     696             :                 void IEventReceiver.Start()
     697             :                 {
     698             :                         // This is in-process so doesn't need to do anything
     699             :                 }
     700             : 
     701             :                 #endregion
     702             :         }
     703             : }

Generated by: LCOV version 1.10