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.Generic;
11 : using System.Diagnostics;
12 : using System.Linq;
13 : using Chinchilla.Logging;
14 : using Cqrs.Authentication;
15 : using Cqrs.Bus;
16 : using Cqrs.Commands;
17 : using Cqrs.Configuration;
18 : using Cqrs.Events;
19 : using Cqrs.Infrastructure;
20 : using Cqrs.Messages;
21 : #if NET452
22 : using Microsoft.ServiceBus.Messaging;
23 : using EventData = Microsoft.ServiceBus.Messaging.EventData;
24 : #endif
25 : #if NETCOREAPP3_0
26 : using Microsoft.Azure.EventHubs;
27 : using Microsoft.Azure.EventHubs.Processor;
28 : using EventData = Microsoft.Azure.EventHubs.EventData;
29 : #endif
30 :
31 : namespace Cqrs.Azure.ServiceBus
32 : {
33 : /// <summary>
34 : /// A <see cref="ICommandPublisher{TAuthenticationToken}"/> that resolves handlers , executes the handler and then publishes the <see cref="ICommand{TAuthenticationToken}"/> on the private command bus.
35 : /// </summary>
36 : /// <typeparam name="TAuthenticationToken">The <see cref="Type"/> of the authentication token.</typeparam>
37 : public class AzureCommandBusPublisher<TAuthenticationToken>
38 : : AzureCommandBus<TAuthenticationToken>
39 : , IPublishAndWaitCommandPublisher<TAuthenticationToken>
40 2 : {
41 : /// <summary>
42 : /// Instantiates a new instance of <see cref="AzureCommandBusPublisher{TAuthenticationToken}"/>.
43 : /// </summary>
44 : public AzureCommandBusPublisher(IConfigurationManager configurationManager, IMessageSerialiser<TAuthenticationToken> messageSerialiser, IAuthenticationTokenHelper<TAuthenticationToken> authenticationTokenHelper, ICorrelationIdHelper correlationIdHelper, ILogger logger, IHashAlgorithmFactory hashAlgorithmFactory, IAzureBusHelper<TAuthenticationToken> azureBusHelper)
45 : : base(configurationManager, messageSerialiser, authenticationTokenHelper, correlationIdHelper, logger, hashAlgorithmFactory, azureBusHelper, true)
46 : {
47 : TelemetryHelper = configurationManager.CreateTelemetryHelper("Cqrs.Azure.EventHub.EventBus.Publisher.UseApplicationInsightTelemetryHelper", correlationIdHelper);
48 : }
49 :
50 : #region Implementation of ICommandPublisher<TAuthenticationToken>
51 :
52 : /// <summary>
53 : /// Publishes the provided <paramref name="command"/> on the command bus.
54 : /// </summary>
55 : public virtual void Publish<TCommand>(TCommand command)
56 : where TCommand : ICommand<TAuthenticationToken>
57 : {
58 : if (command == null)
59 : {
60 : Logger.LogDebug("No command to publish.");
61 : return;
62 : }
63 : DateTimeOffset startedAt = DateTimeOffset.UtcNow;
64 : Stopwatch mainStopWatch = Stopwatch.StartNew();
65 : string responseCode = "200";
66 : bool wasSuccessfull = false;
67 :
68 : IDictionary<string, string> telemetryProperties = new Dictionary<string, string> { { "Type", "Azure/EventHub" } };
69 : string telemetryName = string.Format("{0}/{1}", command.GetType().FullName, command.Id);
70 : var telemeteredCommand = command as ITelemeteredMessage;
71 : if (telemeteredCommand != null)
72 : telemetryName = telemeteredCommand.TelemetryName;
73 : telemetryName = string.Format("Command/{0}", telemetryName);
74 :
75 : try
76 : {
77 : if (!AzureBusHelper.PrepareAndValidateCommand(command, "Azure-EventHub"))
78 : return;
79 :
80 : var brokeredMessage = CreateBrokeredMessage(MessageSerialiser.SerialiseCommand, command.GetType(), command);
81 :
82 : try
83 : {
84 : #if NET452
85 : EventHubPublisher.Send(brokeredMessage);
86 : #endif
87 : #if NETCOREAPP3_0
88 : EventHubPublisher.SendAsync(brokeredMessage).Wait();
89 : #endif
90 : }
91 : catch (Exception exception)
92 : {
93 : responseCode = "500";
94 : Logger.LogError("An issue occurred while trying to publish a command.", exception: exception, metaData: new Dictionary<string, object> { { "Command", command } });
95 : throw;
96 : }
97 : Logger.LogInfo(string.Format("A command was sent of type {0}.", command.GetType().FullName));
98 : wasSuccessfull = true;
99 : }
100 : finally
101 : {
102 : mainStopWatch.Stop();
103 : TelemetryHelper.TrackDependency("Azure/EventHub/CommandBus", "Command", telemetryName, null, startedAt, mainStopWatch.Elapsed, responseCode, wasSuccessfull, telemetryProperties);
104 : }
105 : }
106 :
107 : /// <summary>
108 : /// Publishes the provided <paramref name="commands"/> on the command bus.
109 : /// </summary>
110 : public virtual void Publish<TCommand>(IEnumerable<TCommand> commands)
111 : where TCommand : ICommand<TAuthenticationToken>
112 : {
113 : if (commands == null)
114 : {
115 : Logger.LogDebug("No commands to publish.");
116 : return;
117 : }
118 : IList<TCommand> sourceCommands = commands.ToList();
119 : if (!sourceCommands.Any())
120 : {
121 : Logger.LogDebug("An empty collection of commands to publish.");
122 : return;
123 : }
124 :
125 : DateTimeOffset startedAt = DateTimeOffset.UtcNow;
126 : Stopwatch mainStopWatch = Stopwatch.StartNew();
127 : string responseCode = "200";
128 : bool wasSuccessfull = false;
129 :
130 : IDictionary<string, string> telemetryProperties = new Dictionary<string, string> { { "Type", "Azure/EventHub" } };
131 : string telemetryName = "Commands";
132 : string telemetryNames = string.Empty;
133 : foreach (TCommand command in sourceCommands)
134 : {
135 : string subTelemetryName = string.Format("{0}/{1}", command.GetType().FullName, command.Id);
136 : var telemeteredCommand = command as ITelemeteredMessage;
137 : if (telemeteredCommand != null)
138 : subTelemetryName = telemeteredCommand.TelemetryName;
139 : telemetryNames = string.Format("{0}{1},", telemetryNames, subTelemetryName);
140 : }
141 : if (telemetryNames.Length > 0)
142 : telemetryNames = telemetryNames.Substring(0, telemetryNames.Length - 1);
143 : telemetryProperties.Add("Commands", telemetryNames);
144 :
145 : try
146 : {
147 : IList<string> sourceCommandMessages = new List<string>();
148 : IList<EventData> brokeredMessages = new List<EventData>(sourceCommands.Count);
149 : foreach (TCommand command in sourceCommands)
150 : {
151 : if (!AzureBusHelper.PrepareAndValidateCommand(command, "Azure-EventHub"))
152 : continue;
153 :
154 : var brokeredMessage = CreateBrokeredMessage(MessageSerialiser.SerialiseCommand, command.GetType(), command);
155 :
156 : brokeredMessages.Add(brokeredMessage);
157 : sourceCommandMessages.Add(string.Format("A command was sent of type {0}.", command.GetType().FullName));
158 : }
159 :
160 : try
161 : {
162 : if (brokeredMessages.Any())
163 : {
164 : #if NET452
165 : EventHubPublisher.SendBatch(brokeredMessages);
166 : #endif
167 : #if NETCOREAPP3_0
168 : EventHubPublisher.SendAsync(brokeredMessages).Wait();
169 : #endif
170 : }
171 : else
172 : Logger.LogDebug("An empty collection of commands to publish post validation.");
173 : }
174 : catch (Exception exception)
175 : {
176 : responseCode = "500";
177 : Logger.LogError("An issue occurred while trying to publish a command.", exception: exception, metaData: new Dictionary<string, object> { { "Commands", sourceCommands } });
178 : throw;
179 : }
180 :
181 : foreach (string message in sourceCommandMessages)
182 : Logger.LogInfo(message);
183 :
184 : wasSuccessfull = true;
185 : }
186 : finally
187 : {
188 : mainStopWatch.Stop();
189 : TelemetryHelper.TrackDependency("Azure/EventHub/CommandBus", "Command", telemetryName, null, startedAt, mainStopWatch.Elapsed, responseCode, wasSuccessfull, telemetryProperties);
190 : }
191 : }
192 :
193 : /// <summary>
194 : /// Sends the provided <paramref name="command"></paramref> and waits for an event of <typeparamref name="TEvent"/>
195 : /// </summary>
196 : /// <param name="command">The <typeparamref name="TCommand"/> to send.</param>
197 : /// <param name="eventReceiver">If provided, is the <see cref="IEventReceiver{TAuthenticationToken}" /> that the event is expected to be returned on.</param>
198 : public virtual TEvent PublishAndWait<TCommand, TEvent>(TCommand command, IEventReceiver<TAuthenticationToken> eventReceiver = null)
199 : where TCommand : ICommand<TAuthenticationToken>
200 : {
201 : return PublishAndWait<TCommand, TEvent>(command, -1, eventReceiver);
202 : }
203 :
204 : /// <summary>
205 : /// Sends the provided <paramref name="command"></paramref> and waits for an event of <typeparamref name="TEvent"/> or exits if the specified timeout is expired.
206 : /// </summary>
207 : /// <param name="command">The <typeparamref name="TCommand"/> to send.</param>
208 : /// <param name="millisecondsTimeout">The number of milliseconds to wait, or <see cref="F:System.Threading.Timeout.Infinite"/> (-1) to wait indefinitely.</param>
209 : /// <param name="eventReceiver">If provided, is the <see cref="IEventReceiver{TAuthenticationToken}" /> that the event is expected to be returned on.</param>
210 : public virtual TEvent PublishAndWait<TCommand, TEvent>(TCommand command, int millisecondsTimeout, IEventReceiver<TAuthenticationToken> eventReceiver = null)
211 : where TCommand : ICommand<TAuthenticationToken>
212 : {
213 : return PublishAndWait(command, events => (TEvent)events.SingleOrDefault(@event => @event is TEvent), millisecondsTimeout, eventReceiver);
214 : }
215 :
216 : /// <summary>
217 : /// Sends the provided <paramref name="command"></paramref> and waits for an event of <typeparamref name="TEvent"/> or exits if the specified timeout is expired.
218 : /// </summary>
219 : /// <param name="command">The <typeparamref name="TCommand"/> to send.</param>
220 : /// <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>
221 : /// <param name="eventReceiver">If provided, is the <see cref="IEventReceiver{TAuthenticationToken}" /> that the event is expected to be returned on.</param>
222 : public virtual TEvent PublishAndWait<TCommand, TEvent>(TCommand command, TimeSpan timeout, IEventReceiver<TAuthenticationToken> eventReceiver = null)
223 : where TCommand : ICommand<TAuthenticationToken>
224 : {
225 : long num = (long)timeout.TotalMilliseconds;
226 : if (num < -1L || num > int.MaxValue)
227 : throw new ArgumentOutOfRangeException("timeout", timeout, "SpinWait_SpinUntil_TimeoutWrong");
228 : return PublishAndWait<TCommand, TEvent>(command, (int)timeout.TotalMilliseconds, eventReceiver);
229 : }
230 :
231 : /// <summary>
232 : /// Sends the provided <paramref name="command"></paramref> and waits until the specified condition is satisfied an event of <typeparamref name="TEvent"/>
233 : /// </summary>
234 : /// <param name="command">The <typeparamref name="TCommand"/> to send.</param>
235 : /// <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>
236 : /// <param name="eventReceiver">If provided, is the <see cref="IEventReceiver{TAuthenticationToken}" /> that the event is expected to be returned on.</param>
237 : public virtual TEvent PublishAndWait<TCommand, TEvent>(TCommand command, Func<IEnumerable<IEvent<TAuthenticationToken>>, TEvent> condition, IEventReceiver<TAuthenticationToken> eventReceiver = null)
238 : where TCommand : ICommand<TAuthenticationToken>
239 : {
240 : return PublishAndWait(command, condition, -1, eventReceiver);
241 : }
242 :
243 : /// <summary>
244 : /// Sends the provided <paramref name="command"></paramref> and waits for an event of <typeparamref name="TEvent"/> or exits if the specified timeout is expired.
245 : /// </summary>
246 : /// <param name="command">The <typeparamref name="TCommand"/> to send.</param>
247 : /// <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>
248 : /// <param name="millisecondsTimeout">The number of milliseconds to wait, or <see cref="F:System.Threading.Timeout.Infinite"/> (-1) to wait indefinitely.</param>
249 : /// <param name="eventReceiver">If provided, is the <see cref="IEventReceiver{TAuthenticationToken}" /> that the event is expected to be returned on.</param>
250 : public virtual TEvent PublishAndWait<TCommand, TEvent>(TCommand command, Func<IEnumerable<IEvent<TAuthenticationToken>>, TEvent> condition, int millisecondsTimeout,
251 : IEventReceiver<TAuthenticationToken> eventReceiver = null) where TCommand : ICommand<TAuthenticationToken>
252 : {
253 : if (command == null)
254 : {
255 : Logger.LogDebug("No command to publish.");
256 : return (TEvent)(object)null;
257 : }
258 : DateTimeOffset startedAt = DateTimeOffset.UtcNow;
259 : Stopwatch mainStopWatch = Stopwatch.StartNew();
260 : string responseCode = "200";
261 : bool wasSuccessfull = false;
262 :
263 : IDictionary<string, string> telemetryProperties = new Dictionary<string, string> { { "Type", "Azure/EventHub" } };
264 : string telemetryName = string.Format("{0}/{1}", command.GetType().FullName, command.Id);
265 : var telemeteredCommand = command as ITelemeteredMessage;
266 : if (telemeteredCommand != null)
267 : telemetryName = telemeteredCommand.TelemetryName;
268 : telemetryName = string.Format("Command/{0}", telemetryName);
269 :
270 : TEvent result;
271 :
272 : try
273 : {
274 : if (eventReceiver != null)
275 : throw new NotSupportedException("Specifying a different event receiver is not yet supported.");
276 : if (!AzureBusHelper.PrepareAndValidateCommand(command, "Azure-EventHub"))
277 : return (TEvent)(object)null;
278 :
279 : result = (TEvent)(object)null;
280 : EventWaits.Add(command.CorrelationId, new List<IEvent<TAuthenticationToken>>());
281 :
282 : try
283 : {
284 : var brokeredMessage = CreateBrokeredMessage(MessageSerialiser.SerialiseCommand, command.GetType(), command);
285 : #if NET452
286 : EventHubPublisher.Send(brokeredMessage);
287 : #endif
288 : #if NETCOREAPP3_0
289 : EventHubPublisher.SendAsync(brokeredMessage).Wait();
290 : #endif
291 : }
292 : catch (Exception exception)
293 : {
294 : responseCode = "500";
295 : Logger.LogError("An issue occurred while trying to publish a command.", exception: exception, metaData: new Dictionary<string, object> { { "Command", command } });
296 : throw;
297 : }
298 : Logger.LogInfo(string.Format("A command was sent of type {0}.", command.GetType().FullName));
299 : wasSuccessfull = true;
300 : }
301 : finally
302 : {
303 : mainStopWatch.Stop();
304 : TelemetryHelper.TrackDependency("Azure/EventHub/CommandBus", "Command", telemetryName, null, startedAt, mainStopWatch.Elapsed, responseCode, wasSuccessfull, telemetryProperties);
305 : }
306 :
307 : SpinWait.SpinUntil(() =>
308 : {
309 : IList<IEvent<TAuthenticationToken>> events = EventWaits[command.CorrelationId];
310 :
311 : result = condition(events);
312 :
313 : return result != null;
314 : }, millisecondsTimeout, 1000);
315 :
316 : TelemetryHelper.TrackDependency("Azure/EventHub/CommandBus", "Command/AndWait", string.Format("Command/AndWait{0}", telemetryName.Substring(7)), null, startedAt, mainStopWatch.Elapsed, responseCode, wasSuccessfull, telemetryProperties);
317 : return result;
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="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>
325 : /// <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>
326 : /// <param name="eventReceiver">If provided, is the <see cref="IEventReceiver{TAuthenticationToken}" /> that the event is expected to be returned on.</param>
327 : public virtual TEvent PublishAndWait<TCommand, TEvent>(TCommand command, Func<IEnumerable<IEvent<TAuthenticationToken>>, TEvent> condition, TimeSpan timeout, IEventReceiver<TAuthenticationToken> eventReceiver = null)
328 : where TCommand : ICommand<TAuthenticationToken>
329 : {
330 : long num = (long)timeout.TotalMilliseconds;
331 : if (num < -1L || num > int.MaxValue)
332 : throw new ArgumentOutOfRangeException("timeout", timeout, "SpinWait_SpinUntil_TimeoutWrong");
333 : return PublishAndWait(command, condition, (int)timeout.TotalMilliseconds, eventReceiver);
334 : }
335 :
336 : #endregion
337 : }
338 : }
|
