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