Skip to content

Commit b5393b4

Browse files
joefeserjoefeser
committed
Start async send code. DO NOT USE.
1 parent 0f87e93 commit b5393b4

8 files changed

Lines changed: 144 additions & 19 deletions

File tree

‎src/ProjectExtensions.Azure.ServiceBus/AzureBus.cs‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -65,6 +65,18 @@ public void Publish<T>(T message, IDictionary<string, object> metadata) {
6565
sender.Send<T>(message, metadata);
6666
}
6767

68+
/// <summary>
69+
/// Publish a Message with the given signature.
70+
/// </summary>
71+
/// <typeparam name="T"></typeparam>
72+
/// <param name="message">The message to publish.</param>
73+
/// <param name="resultCallBack">The callback when the operation completes</param>
74+
/// <param name="metadata">Metadata to sent with the message.</param>
75+
public void PublishAsync<T>(T message, Action<IMessageSentResult<T>> resultCallBack, IDictionary<string, object> metadata) {
76+
logger.Info("Publish={0}", message.GetType().FullName);
77+
sender.SendAsync<T>(message, resultCallBack, metadata);
78+
}
79+
6880
/// <summary>
6981
/// Subscribes to recieve published messages of type T.
7082
/// This method is only necessary if you turned off auto-subscribe

‎src/ProjectExtensions.Azure.ServiceBus/AzureBusSender.cs‎

Lines changed: 56 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -34,20 +34,60 @@ public void Close() {
3434
}
3535
}
3636

37+
public void Send<T>(T obj) {
38+
Send<T>(obj, null);
39+
}
40+
3741
public void Send<T>(T obj, IDictionary<string, object> metadata) {
38-
Send<T>(obj, metadata, configuration.DefaultSerializer.Create());
42+
Send<T>(obj, configuration.DefaultSerializer.Create(), metadata);
3943
}
4044

41-
public void Send<T>(T obj, IDictionary<string, object> metadata, IServiceBusSerializer serializer) {
45+
public void Send<T>(T obj, IServiceBusSerializer serializer = null, IDictionary<string, object> metadata = null) {
4246

4347
// Declare a wait object that will be used for synchronization.
4448
var waitObject = new ManualResetEvent(false);
4549

4650
// Declare a timeout value during which the messages are expected to be sent.
4751
var sentTimeout = TimeSpan.FromMinutes(2);
4852

53+
Exception failureException = null;
54+
55+
SendAsync<T>(obj, (result) => {
56+
waitObject.Set();
57+
failureException = result.ThrownException;
58+
});
59+
60+
// Wait until the messaging operations are completed.
61+
bool completed = waitObject.WaitOne(sentTimeout);
62+
waitObject.Dispose();
63+
64+
if (completed) {
65+
//DO Nothing
66+
}
67+
else {
68+
if (failureException != null) {
69+
throw failureException;
70+
}
71+
throw new Exception("Failed to Send Message for Unknown Reason.");
72+
}
73+
}
74+
75+
public void SendAsync<T>(T obj, Action<IMessageSentResult<T>> resultCallBack) {
76+
SendAsync<T>(obj, resultCallBack, configuration.DefaultSerializer.Create());
77+
}
78+
79+
public void SendAsync<T>(T obj, Action<IMessageSentResult<T>> resultCallBack, IDictionary<string, object> metadata) {
80+
SendAsync<T>(obj, resultCallBack, configuration.DefaultSerializer.Create(), metadata);
81+
}
82+
83+
public void SendAsync<T>(T obj, Action<IMessageSentResult<T>> resultCallBack, IServiceBusSerializer serializer = null, IDictionary<string, object> metadata = null) {
84+
85+
serializer = serializer ?? configuration.DefaultSerializer.Create();
86+
4987
Exception failureException = null;
5088
BrokeredMessage message = null;
89+
var sw = new Stopwatch();
90+
sw.Start();
5191

5292
// Use a retry policy to execute the Send action in an asynchronous and reliable fashion.
5393
retryPolicy.ExecuteAction
@@ -86,7 +126,12 @@ public void Send<T>(T obj, IDictionary<string, object> metadata, IServiceBusSeri
86126
serializer.Dispose();
87127
serializer = null;
88128
}
89-
waitObject.Set();
129+
sw.Stop();
130+
resultCallBack(new MessageSentResult<T>() {
131+
IsSuccess = true,
132+
Message = obj,
133+
TimeSpent = sw.Elapsed
134+
});
90135
}
91136
},
92137
(ex) => {
@@ -105,23 +150,19 @@ public void Send<T>(T obj, IDictionary<string, object> metadata, IServiceBusSeri
105150
}
106151
);
107152

108-
// Wait until the messaging operations are completed.
109-
bool completed = waitObject.WaitOne(sentTimeout);
110-
waitObject.Dispose();
153+
sw.Stop();
154+
resultCallBack(new MessageSentResult<T>() {
155+
IsSuccess = true,
156+
Message = obj,
157+
ThrownException = failureException,
158+
TimeSpent = sw.Elapsed
159+
});
111160

112-
if (completed) {
113-
//DO Nothing
114-
}
115-
else {
116-
if (failureException != null) {
117-
throw failureException;
118-
}
119-
throw new Exception("Failed to Send Message for Unknown Reason.");
120-
}
121161
}
122162

123163
public override void Dispose(bool disposing) {
124164
Close();
125165
}
166+
126167
}
127168
}

‎src/ProjectExtensions.Azure.ServiceBus/IAzureBusSender.cs‎

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -3,12 +3,15 @@
33
using System.Collections.Generic;
44

55
namespace ProjectExtensions.Azure.ServiceBus {
6-
6+
77
interface IAzureBusSender {
88
void Close();
99
void Dispose(bool disposing);
10+
void Send<T>(T obj);
1011
void Send<T>(T obj, IDictionary<string, object> metadata);
11-
void Send<T>(T obj, IDictionary<string, object> metadata, IServiceBusSerializer serializer);
12+
void Send<T>(T obj, IServiceBusSerializer serializer, IDictionary<string, object> metadata);
13+
void SendAsync<T>(T obj, Action<IMessageSentResult<T>> resultCallBack);
14+
void SendAsync<T>(T obj, Action<IMessageSentResult<T>> resultCallBack, IDictionary<string, object> metadata);
15+
void SendAsync<T>(T obj, Action<IMessageSentResult<T>> resultCallBack, IServiceBusSerializer serializer = null, IDictionary<string, object> metadata = null);
1216
}
13-
1417
}

‎src/ProjectExtensions.Azure.ServiceBus/IBus.cs‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,15 @@ public interface IBus {
2525
/// <param name="metadata">Metadata to sent with the message.</param>
2626
void Publish<T>(T message, IDictionary<string, object> metadata);
2727

28+
/// <summary>
29+
/// Publish a Message with the given signature.
30+
/// </summary>
31+
/// <typeparam name="T"></typeparam>
32+
/// <param name="message">The message to publish.</param>
33+
/// <param name="resultCallBack">The callback when the message is complete</param>
34+
/// <param name="metadata">Metadata to sent with the message.</param>
35+
void PublishAsync<T>(T message, Action<IMessageSentResult<T>> resultCallBack, IDictionary<string, object> metadata);
36+
2837
/// <summary>
2938
/// Subscribes to recieve published messages of type T.
3039
/// This method is only necessary if you turned off auto-subscribe.

‎src/ProjectExtensions.Azure.ServiceBus/IMessageSentResult.cs‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,22 @@
1+
using System;
2+
namespace ProjectExtensions.Azure.ServiceBus {
3+
4+
public interface IMessageSentResult<T> {
5+
bool IsSuccess {
6+
get;
7+
set;
8+
}
9+
Exception ThrownException {
10+
get;
11+
set;
12+
}
13+
TimeSpan TimeSpent {
14+
get;
15+
set;
16+
}
17+
T Message {
18+
get;
19+
set;
20+
}
21+
}
22+
}

‎src/ProjectExtensions.Azure.ServiceBus/MessageSentResult.cs‎

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,34 @@
1+
using System;
2+
using System.Collections.Generic;
3+
using System.Linq;
4+
using System.Text;
5+
6+
namespace ProjectExtensions.Azure.ServiceBus {
7+
8+
/// <summary>
9+
/// Result of a call to SendAsyc on the message bus
10+
/// </summary>
11+
public class MessageSentResult<T> : IMessageSentResult<T> {
12+
13+
public bool IsSuccess {
14+
get;
15+
set;
16+
}
17+
18+
public Exception ThrownException {
19+
get;
20+
set;
21+
}
22+
23+
public TimeSpan TimeSpent {
24+
get;
25+
set;
26+
}
27+
28+
public T Message {
29+
get;
30+
set;
31+
}
32+
33+
}
34+
}

‎src/ProjectExtensions.Azure.ServiceBus/ProjectExtensions.Azure.ServiceBus.csproj‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -82,7 +82,9 @@
8282
<Compile Include="IBusConfiguration.cs" />
8383
<Compile Include="IMessageHandler.cs" />
8484
<Compile Include="IBus.cs" />
85+
<Compile Include="IMessageSentResult.cs" />
8586
<Compile Include="IReceivedMessage.cs" />
87+
<Compile Include="MessageSentResult.cs" />
8688
<Compile Include="ReflectionHelper.cs" />
8789
<Compile Include="ReceivedMessage.cs" />
8890
<Compile Include="Serialization\IServiceBusSerializer.cs" />

‎src/Samples/PubSubUsingConfiguration/Program.cs‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -38,7 +38,9 @@ static void Main(string[] args) {
3838
Value = i,
3939
MessageId = DateTime.Now.ToString()
4040
};
41-
BusConfiguration.Instance.Bus.Publish(message1, null);
41+
BusConfiguration.Instance.Bus.PublishAsync(message1, (result) => {
42+
Console.WriteLine(result.TimeSpent);
43+
}, null);
4244
}
4345

4446
var message2 = new AnotherTestMessage() {

0 commit comments

Comments
 (0)