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