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