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