Skip to content

Commit f6fb9d6

Browse files
ruakuikruakuik
ruakuik
authored and
ruakuik
committed
Release 0013
1 parent 034370a commit f6fb9d6

File tree

12 files changed

+168
-115
lines changed

12 files changed

+168
-115
lines changed

‎ViennaNET.Messaging.RabbitMQQueue/RabbitMqQueueMessageConverter.cs‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -85,10 +85,14 @@ public static BaseMessage ConvertToBaseMessage(this byte[] body, MessageProperti
8585
message.MessageId = messageProperties.MessageId;
8686
message.CorrelationId = messageProperties.CorrelationId;
8787
message.ReplyQueue = messageProperties.ReplyTo;
88-
message.LifeTime = TimeSpan.Parse(messageProperties.Expiration);
8988
message.SendDateTime = DateTime.FromFileTimeUtc(messageProperties.Timestamp);
9089
message.ReceiveDate = DateTime.Now;
9190

91+
if (TimeSpan.TryParse(messageProperties.Expiration, out var lifetime))
92+
{
93+
message.LifeTime = lifetime;
94+
}
95+
9296
foreach (var header in messageProperties.Headers)
9397
{
9498
var value = string.Empty;

‎ViennaNET.Messaging.Tests/Unit/DSL/MessageAdapterBuilder.cs‎

Lines changed: 23 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
using System;
2+
using System.Threading.Tasks;
23
using Moq;
34
using ViennaNET.Messaging.Configuration;
45
using ViennaNET.Messaging.Messages;
@@ -8,7 +9,6 @@ namespace ViennaNET.Messaging.Tests.Unit.DSL
89
{
910
internal class MessageAdapterBuilder
1011
{
11-
1212
private QueueConfigurationBase _queueConfiguration;
1313
private MessageProcessingType? _processingType;
1414

@@ -24,27 +24,39 @@ public MessageAdapterBuilder SupportsProcessingType(MessageProcessingType proces
2424
return this;
2525
}
2626

27-
public T Please<T>(Action<Mock<T>> configure = null) where T: class, IMessageAdapter
27+
public T Please<T>(Action<Mock<T>> configure = null) where T : class, IMessageAdapter => MockPlease(configure).Object;
28+
29+
public IMessageAdapter Please(Action<Mock<IMessageAdapter>> configure = null) => Please<IMessageAdapter>(configure);
30+
31+
public Mock<T> MockPlease<T>(Action<Mock<T>> configure = null) where T : class, IMessageAdapter
2832
{
2933
var messageAdapter = new Mock<T>();
3034
messageAdapter
3135
.Setup(x => x.Configuration)
32-
.Returns(_queueConfiguration ?? new MqSeriesQueueConfiguration { IntervalPollingQueue = 1000, ProcessingType = MessageProcessingType.ThreadStrategy });
36+
.Returns(_queueConfiguration ??
37+
new MqSeriesQueueConfiguration
38+
{
39+
IntervalPollingQueue = 1000, ProcessingType = MessageProcessingType.ThreadStrategy
40+
});
3341
messageAdapter
34-
.Setup(x => x.SupportProcessingType(It.Is<MessageProcessingType>(pt => !_processingType.HasValue || pt == _processingType.Value)))
42+
.Setup(x => x.SupportProcessingType(It.Is<MessageProcessingType>(pt => _processingType == null || pt == _processingType)))
3543
.Returns(true);
3644
messageAdapter
3745
.Setup(x => x.Send(It.IsAny<BaseMessage>()))
3846
.Returns<BaseMessage>(m => m);
3947

40-
configure?.Invoke(messageAdapter);
48+
if (typeof(T) == typeof(IMessageAdapterWithSubscribing))
49+
{
50+
messageAdapter.As<IMessageAdapterWithSubscribing>()
51+
.Setup(x => x.Subscribe(It.IsAny<Func<BaseMessage, Task>>()))
52+
.Callback<Func<BaseMessage, Task>>(cb => messageAdapter
53+
.Setup(x => x.Send(It.IsAny<BaseMessage>()))
54+
.Callback<BaseMessage>(msg => cb(msg).GetAwaiter().GetResult()));
55+
}
4156

42-
return messageAdapter.Object;
43-
}
57+
configure?.Invoke(messageAdapter);
4458

45-
public IMessageAdapter Please(Action<Mock<IMessageAdapter>> configure = null)
46-
{
47-
return Please<IMessageAdapter>(configure);
59+
return messageAdapter;
4860
}
4961
}
50-
}
62+
}

‎ViennaNET.Messaging.Tests/Unit/DSL/MessageAdapterConstructorMock.cs‎

Lines changed: 3 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -14,15 +14,10 @@ public MessageAdapterConstructorMock WithAdapter(IMessageAdapter adapter, string
1414
return this;
1515
}
1616

17-
public MessageAdapterConstructorMock WithAdapter(Func<MessageAdapterBuilder, MessageAdapterBuilder> builder,
18-
string queueId)
19-
{
20-
_adapters.Add(queueId, builder(new MessageAdapterBuilder()).Please());
21-
return this;
22-
}
17+
public MessageAdapterConstructorMock WithAdapter(Func<MessageAdapterBuilder, MessageAdapterBuilder> builder, string queueId)
18+
=> WithAdapter(b => builder(b).Please(), queueId);
2319

24-
public MessageAdapterConstructorMock WithAdapter(Func<MessageAdapterBuilder, IMessageAdapter> builder,
25-
string queueId)
20+
public MessageAdapterConstructorMock WithAdapter(Func<MessageAdapterBuilder, IMessageAdapter> builder, string queueId)
2621
{
2722
_adapters.Add(queueId, builder(new MessageAdapterBuilder()));
2823
return this;

‎ViennaNET.Messaging.Tests/Unit/Processing/Impl/Subscribe/QueueSubscribedReactorBaseTests.cs‎

Lines changed: 79 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -1,13 +1,14 @@
11
using System;
2+
using System.Linq;
23
using System.Threading.Tasks;
34
using Microsoft.Extensions.Logging;
45
using Moq;
56
using NUnit.Framework;
67
using ViennaNET.Diagnostic;
7-
using ViennaNET.Messaging.Configuration;
88
using ViennaNET.Messaging.Context;
99
using ViennaNET.Messaging.Messages;
1010
using ViennaNET.Messaging.Processing.Impl.Subscribe;
11+
using ViennaNET.Messaging.Tests.Unit.DSL;
1112

1213
namespace ViennaNET.Messaging.Tests.Unit.Processing.Impl.Subscribe
1314
{
@@ -19,7 +20,7 @@ public class QueueSubscribedReactorBaseTests
1920
public void StartProcessing_NoErrorsOnConnection_ReturnsTrue()
2021
{
2122
// arrange
22-
var reactor = GetReactor(null, out var fakeAdapter, out var fakeHealthService);
23+
var reactor = GetReactor();
2324

2425
// act
2526
var result = reactor.StartProcessing();
@@ -32,9 +33,10 @@ public void StartProcessing_NoErrorsOnConnection_ReturnsTrue()
3233
public void StartProcessing_AdapterAlreadyConnected_AdapterConnectNotCalled()
3334
{
3435
// arrange
35-
var reactor = GetReactor(true, out var fakeAdapter, out var fakeHealthService);
36+
var fakeAdapter = Given.MessageAdapter.MockPlease<IMessageAdapterWithSubscribing>();
3637
fakeAdapter.Setup(x => x.IsConnected)
3738
.Returns(true);
39+
var reactor = GetReactor(messageAdapter: fakeAdapter.Object);
3840
// act
3941
var result = reactor.StartProcessing();
4042

@@ -47,9 +49,11 @@ public void StartProcessing_AdapterAlreadyConnected_AdapterConnectNotCalled()
4749
public void StartProcessing_AdapterThrowsExceptionOnConnect_FailReturnsFalse()
4850
{
4951
// arrange
50-
var reactor = GetReactor(true, out var fakeAdapter, out var fakeHealthService);
52+
var fakeAdapter = Given.MessageAdapter.MockPlease<IMessageAdapterWithSubscribing>();
5153
fakeAdapter.Setup(x => x.Connect())
5254
.Throws<Exception>();
55+
var reactor = GetReactor(messageAdapter: fakeAdapter.Object);
56+
5357
// act
5458
var result = reactor.StartProcessing();
5559

@@ -62,59 +66,107 @@ public void StartProcessing_AdapterThrowsTimeoutExceptionAndDoNotThrowsOnRetry_S
6266
{
6367
// arrange
6468
var isConnectCalled = false;
65-
var reactor = GetReactor(true, out var fakeAdapter, out var fakeHealthService);
69+
var fakeAdapter = Given.MessageAdapter.MockPlease<IMessageAdapterWithSubscribing>();
6670
fakeAdapter.Setup(x => x.Connect())
67-
.Callback(() =>
71+
.Callback(() =>
6872
{
6973
if (!isConnectCalled)
7074
{
7175
isConnectCalled = true;
7276
throw new TimeoutException();
7377
}
7478
});
75-
79+
var reactor = GetReactor(messageAdapter: fakeAdapter.Object);
80+
7681
// act
7782
var result = reactor.StartProcessing();
7883

7984
// assert
8085
Assert.That(result, Is.True);
8186
}
8287

83-
private QueueSubscribedReactorBase GetReactor(
84-
bool? serviceHealthDependent, out Mock<IMessageAdapterWithSubscribing> fakeAdapter,
85-
out Mock<IHealthCheckingService> fakeHealthService)
88+
[Test]
89+
public async Task StartProcessing_ConcurrentMessages_ShouldProcessSuccessfully()
90+
{
91+
// arrange
92+
var messageAdapter = Given.MessageAdapter.Please<IMessageAdapterWithSubscribing>();
93+
var reactor = GetReactor(messageAdapter: messageAdapter);
94+
reactor.StartProcessing();
95+
96+
// act & assert
97+
Task SendMessage() => Task.Run(() => messageAdapter.Send(new TextMessage()));
98+
var tasks = Enumerable.Range(1, 10).Select(_ => SendMessage());
99+
await Task.WhenAll(tasks);
100+
}
101+
102+
[Test]
103+
public void StartProcessing_ClearCallContextFailed_ShouldProcessMessage()
104+
{
105+
var messageAdapter = Given.MessageAdapter.Please<IMessageAdapterWithSubscribing>();
106+
var messagingCallContextAccessorMock = new Mock<IMessagingCallContextAccessor>();
107+
messagingCallContextAccessorMock.Setup(x => x.CleanContext()).Throws<Exception>();
108+
var reactor = GetReactor(messageAdapter: messageAdapter,
109+
messagingCallContextAccessor: messagingCallContextAccessorMock.Object);
110+
111+
reactor.StartProcessing();
112+
messageAdapter.Send(new TextMessage());
113+
114+
Assert.That(reactor.WasProcessed, Is.True);
115+
}
116+
117+
[Test]
118+
public void StartProcessing_WhenDiagnosticFailed_ShouldUnsubscribe()
86119
{
87-
fakeAdapter = new Mock<IMessageAdapterWithSubscribing>();
88-
fakeAdapter.Setup(x => x.Configuration)
89-
.Returns(new Mock<QueueConfigurationBase>().Object);
90-
fakeHealthService = new Mock<IHealthCheckingService>();
91-
var fakeCallContextAccessor = new Mock<IMessagingCallContextAccessor>();
92-
93-
return new QueueSubscribedReactorWrapper(fakeAdapter.Object, 100, serviceHealthDependent, fakeHealthService.Object,
94-
fakeCallContextAccessor.Object);
120+
var messageAdapter = Given.MessageAdapter.MockPlease<IMessageAdapterWithSubscribing>();
121+
var healthCheckingServiceMock = new Mock<IHealthCheckingService>();
122+
var reactor = GetReactor(messageAdapter: messageAdapter.Object,
123+
healthService: healthCheckingServiceMock.Object);
124+
125+
reactor.StartProcessing();
126+
healthCheckingServiceMock.Raise(x => x.DiagnosticFailedEvent += null);
127+
128+
messageAdapter.Verify(x => x.Unsubscribe());
95129
}
96130

131+
private static QueueSubscribedReactorWrapper GetReactor(
132+
bool? serviceHealthDependent = true,
133+
IMessageAdapterWithSubscribing messageAdapter = null,
134+
IHealthCheckingService healthService = null,
135+
IMessagingCallContextAccessor messagingCallContextAccessor = null)
136+
=> new QueueSubscribedReactorWrapper(messageAdapter ?? Given.MessageAdapter.Please<IMessageAdapterWithSubscribing>(),
137+
100,
138+
serviceHealthDependent,
139+
healthService ?? Mock.Of<IHealthCheckingService>(),
140+
messagingCallContextAccessor ?? Mock.Of<IMessagingCallContextAccessor>());
141+
97142
private class QueueSubscribedReactorWrapper : QueueSubscribedReactorBase
98143
{
99144
public QueueSubscribedReactorWrapper(
100-
IMessageAdapterWithSubscribing messageAdapter, int reconnectTimeout, bool? serviceHealthDependent,
101-
IHealthCheckingService healthCheckingService, IMessagingCallContextAccessor messagingCallContextAccessor) : base(messageAdapter,
102-
reconnectTimeout,
103-
serviceHealthDependent,
104-
healthCheckingService,
105-
messagingCallContextAccessor,
106-
Mock.Of<ILogger>())
145+
IMessageAdapterWithSubscribing messageAdapter,
146+
int reconnectTimeout,
147+
bool? serviceHealthDependent,
148+
IHealthCheckingService healthCheckingService,
149+
IMessagingCallContextAccessor messagingCallContextAccessor) : base(messageAdapter,
150+
reconnectTimeout,
151+
serviceHealthDependent,
152+
healthCheckingService,
153+
messagingCallContextAccessor,
154+
Mock.Of<ILogger>())
107155
{
108156
}
157+
158+
public bool WasProcessed { get; private set; }
109159

110160
protected override bool GetProcessedMessage(BaseMessage message)
111161
{
112-
throw new System.NotImplementedException();
162+
WasProcessed = true;
163+
return true;
113164
}
114165

115166
protected override Task<bool> GetProcessedMessageAsync(BaseMessage message)
116167
{
117-
throw new System.NotImplementedException();
168+
WasProcessed = true;
169+
return Task.FromResult(true);
118170
}
119171
}
120172
}

0 commit comments

Comments
 (0)