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