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