You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by mi...@apache.org on 2021/09/13 11:08:20 UTC
[activemq-nms-amqp] 03/09: AMQNET-637 Try to: fix failing tests,
add little more information in case of failing frame assertion,
simplify AmqpProducer.SendAsync
This is an automated email from the ASF dual-hosted git repository.
michaelpearce pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/activemq-nms-amqp.git
commit 8eca44ae90ad5ba7e9ce7a51acd8bc14c1ce085c
Author: lukeabsent <tr...@interia.pl>
AuthorDate: Wed Jul 7 11:12:07 2021 +0200
AMQNET-637 Try to: fix failing tests, add little more information in case of failing frame assertion, simplify AmqpProducer.SendAsync
---
src/NMS.AMQP/NmsConnection.cs | 2 +-
src/NMS.AMQP/Provider/Amqp/AmqpConnection.cs | 6 +-
.../Provider/Amqp/AmqpConnectionSession.cs | 7 +-
src/NMS.AMQP/Provider/Amqp/AmqpProducer.cs | 63 ++++-------
.../NmsMessageConsumerTest.cs | 5 +-
... => AmqpAcknowledgmentsIntegrationTestAsync.cs} | 2 +-
...onTest.cs => ConnectionIntegrationTestAsync.cs} | 0
...tionTest.cs => ConsumerIntegrationTestAsync.cs} | 0
...tionTest.cs => FailoverIntegrationTestAsync.cs} | 14 +--
...TimeTest.cs => MessageDeliveryTimeTestAsync.cs} | 0
...cs => MessageExpirationIntegrationTestAsync.cs} | 0
...nTest.cs => NMSConsumerIntegrationTestAsync.cs} | 0
...onTest.cs => NMSContextIntegrationTestAsync.cs} | 0
...nTest.cs => NMSProducerIntegrationTestAsync.cs} | 0
...est.cs => ProducerIntegrationAsyncTestAsync.cs} | 0
...tionTest.cs => ProducerIntegrationTestAsync.cs} | 0
...ationTest.cs => SessionIntegrationTestAsync.cs} | 0
...est.cs => SubscriptionsIntegrationTestAsync.cs} | 0
...st.cs => TemporaryQueueIntegrationTestAsync.cs} | 0
...st.cs => TemporaryTopicIntegrationTestAsync.cs} | 0
...Test.cs => TransactionsIntegrationTestAsync.cs} | 13 ++-
.../Integration/FailoverIntegrationTest.cs | 12 +-
.../Integration/TransactionsIntegrationTest.cs | 14 ++-
.../TestAmqp/Matchers/FrameContext.cs | 31 ++++++
.../TestAmqp/Matchers/FrameMatcher.cs | 10 +-
test/Apache-NMS-AMQP-Test/TestAmqp/TestAmqpPeer.cs | 13 ++-
.../TestAmqp/TestAmqpPeerRunner.cs | 22 +++-
.../NmsSynchronizationMonitorTest.cs | 122 ++++++++++++++-------
28 files changed, 221 insertions(+), 115 deletions(-)
diff --git a/src/NMS.AMQP/NmsConnection.cs b/src/NMS.AMQP/NmsConnection.cs
index e0a98d8..5afae35 100644
--- a/src/NMS.AMQP/NmsConnection.cs
+++ b/src/NMS.AMQP/NmsConnection.cs
@@ -197,7 +197,7 @@ namespace Apache.NMS.AMQP
public async Task StartAsync()
{
- await CreateNmsConnectionAsync().Await();;
+ await CreateNmsConnectionAsync().AwaitRunContinuationAsync();;
if (started.CompareAndSet(false, true))
{
diff --git a/src/NMS.AMQP/Provider/Amqp/AmqpConnection.cs b/src/NMS.AMQP/Provider/Amqp/AmqpConnection.cs
index 0e30ad6..8644c8e 100644
--- a/src/NMS.AMQP/Provider/Amqp/AmqpConnection.cs
+++ b/src/NMS.AMQP/Provider/Amqp/AmqpConnection.cs
@@ -50,6 +50,8 @@ namespace Apache.NMS.AMQP.Provider.Amqp
private readonly AmqpMessageFactory messageFactory;
private AmqpConnectionSession connectionSession;
private TaskCompletionSource<bool> tsc;
+
+ private bool closeCalled = false;
public AmqpConnection(AmqpProvider provider, ITransportContext transport, NmsConnectionInfo info)
{
@@ -97,7 +99,7 @@ namespace Apache.NMS.AMQP.Provider.Amqp
}
bool connectionExplicitlyClosed = error == null;
- if (!connectionExplicitlyClosed)
+ if (!closeCalled || !connectionExplicitlyClosed)
{
var exception = ExceptionSupport.GetException(error);
if (!this.tsc.TrySetException(exception))
@@ -183,6 +185,7 @@ namespace Apache.NMS.AMQP.Provider.Amqp
{
try
{
+ closeCalled = true;
UnderlyingConnection?.Close();
}
catch (Exception ex)
@@ -197,6 +200,7 @@ namespace Apache.NMS.AMQP.Provider.Amqp
{
try
{
+ closeCalled = true;
if (UnderlyingConnection != null) await UnderlyingConnection.CloseAsync().AwaitRunContinuationAsync();
}
catch (Exception ex)
diff --git a/src/NMS.AMQP/Provider/Amqp/AmqpConnectionSession.cs b/src/NMS.AMQP/Provider/Amqp/AmqpConnectionSession.cs
index ba9399e..452c16b 100644
--- a/src/NMS.AMQP/Provider/Amqp/AmqpConnectionSession.cs
+++ b/src/NMS.AMQP/Provider/Amqp/AmqpConnectionSession.cs
@@ -54,8 +54,11 @@ namespace Apache.NMS.AMQP.Provider.Amqp
? $"Cannot remove Subscription {subscriptionName} that does not exists"
: $"Subscription {subscriptionName} unsubscribe operation failure";
- NMSException exception = ExceptionSupport.GetException(sender, failureMessage);
- tcs.TrySetException(exception);
+ if (!tcs.Task.IsCompleted || error != null)
+ {
+ NMSException exception = ExceptionSupport.GetException(sender, failureMessage);
+ tcs.TrySetException(exception);
+ }
});
await tcs.Task.Await();
diff --git a/src/NMS.AMQP/Provider/Amqp/AmqpProducer.cs b/src/NMS.AMQP/Provider/Amqp/AmqpProducer.cs
index 5da3e31..22ce43a 100644
--- a/src/NMS.AMQP/Provider/Amqp/AmqpProducer.cs
+++ b/src/NMS.AMQP/Provider/Amqp/AmqpProducer.cs
@@ -30,8 +30,6 @@ namespace Apache.NMS.AMQP.Provider.Amqp
{
public class AmqpProducer
{
- private static readonly OutcomeCallback _onOutcome = OnOutcome;
-
private readonly AmqpSession session;
private readonly NmsProducerInfo info;
private SenderLink senderLink;
@@ -165,52 +163,37 @@ namespace Apache.NMS.AMQP.Provider.Amqp
private async Task SendAsync(global::Amqp.Message message, DeliveryState deliveryState)
{
- var tcs = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
- CancellationTokenSource cts = null;
- if (session.Connection.Provider.SendTimeout != NmsConnectionInfo.INFINITE)
- {
- cts = new CancellationTokenSource(TimeSpan.FromMilliseconds(session.Connection.Provider.SendTimeout));
- cts.Token.Register(_ =>
- {
- var timeoutException = ExceptionSupport.GetTimeoutException(this.senderLink, $"The operation did not complete within the allocated time {session.Connection.Provider.SendTimeout}ms.");
- tcs.TrySetException(timeoutException);
- }, null);
- }
+ Outcome outcome = null;
try
{
- senderLink.Send(message, deliveryState, _onOutcome, tcs);
- await tcs.Task.Await();
- }
- finally
- {
- cts?.Dispose();
+ outcome = await senderLink.SendAsync(message, deliveryState, session.Connection.Provider.SendTimeout).Await();
+ // SendTimeout being NmsConnectionInfo.INFINITE is handled in SendAsync AmqpSendTask
}
- }
-
- private static void OnOutcome(ILink sender, global::Amqp.Message message, Outcome outcome, object state)
- {
- var tcs = (TaskCompletionSource<bool>) state;
- if (outcome.Descriptor.Code == MessageSupport.ACCEPTED_INSTANCE.Descriptor.Code)
+ catch (TimeoutException)
{
- tcs.TrySetResult(true);
- }
- else if (outcome.Descriptor.Code == MessageSupport.REJECTED_INSTANCE.Descriptor.Code)
- {
- Rejected rejected = (Rejected) outcome;
- tcs.TrySetException(ExceptionSupport.GetException(rejected.Error, $"Message {message.Properties.GetMessageId()} rejected"));
- }
- else if (outcome.Descriptor.Code == MessageSupport.RELEASED_INSTANCE.Descriptor.Code)
- {
- Error error = new Error(ErrorCode.MessageReleased);
- tcs.TrySetException(ExceptionSupport.GetException(error, $"Message {message.Properties.GetMessageId()} released"));
+ throw;
}
- else
+
+ if (outcome.Descriptor.Code != MessageSupport.ACCEPTED_INSTANCE.Descriptor.Code)
{
- Error error = new Error(ErrorCode.InternalError);
- tcs.TrySetException(ExceptionSupport.GetException(error, outcome.ToString()));
+ if (outcome.Descriptor.Code == MessageSupport.REJECTED_INSTANCE.Descriptor.Code)
+ {
+ Rejected rejected = (Rejected) outcome;
+ throw ExceptionSupport.GetException(rejected.Error, $"Message {message.Properties.GetMessageId()} rejected");
+ }
+ else if (outcome.Descriptor.Code == MessageSupport.RELEASED_INSTANCE.Descriptor.Code)
+ {
+ Error error = new Error(ErrorCode.MessageReleased);
+ throw ExceptionSupport.GetException(error, $"Message {message.Properties.GetMessageId()} released");
+ }
+ else
+ {
+ Error error = new Error(ErrorCode.InternalError);
+ throw ExceptionSupport.GetException(error, outcome.ToString());
+ }
}
}
-
+
public async Task CloseAsync()
{
try
diff --git a/test/Apache-NMS-AMQP-Interop-Test/NmsMessageConsumerTest.cs b/test/Apache-NMS-AMQP-Interop-Test/NmsMessageConsumerTest.cs
index 42e74eb..dc8c260 100644
--- a/test/Apache-NMS-AMQP-Interop-Test/NmsMessageConsumerTest.cs
+++ b/test/Apache-NMS-AMQP-Interop-Test/NmsMessageConsumerTest.cs
@@ -56,6 +56,9 @@ namespace NMS.AMQP.Test
Assert.IsInstanceOf<ITextMessage>(msg);
Assert.AreEqual(text, ((ITextMessage) msg).Text);
Assert.IsNull(messageConsumer.Receive(TimeSpan.FromSeconds(1)));
+
+ messageConsumer.Close();
+
}
[Test, Timeout(60_000)]
@@ -310,8 +313,8 @@ namespace NMS.AMQP.Test
messageConsumer1.Close();
messageConsumer2.Close();
+
sessionConsumer1.Unsubscribe(subscriptionName);
- sessionConsumer2.Unsubscribe(subscriptionName);
}
}
diff --git a/test/Apache-NMS-AMQP-Test/Integration/Async/AmqpAcknowledgmentsIntegrationTest.cs b/test/Apache-NMS-AMQP-Test/Integration/Async/AmqpAcknowledgmentsIntegrationTestAsync.cs
similarity index 99%
rename from test/Apache-NMS-AMQP-Test/Integration/Async/AmqpAcknowledgmentsIntegrationTest.cs
rename to test/Apache-NMS-AMQP-Test/Integration/Async/AmqpAcknowledgmentsIntegrationTestAsync.cs
index b419ad4..207b1e9 100644
--- a/test/Apache-NMS-AMQP-Test/Integration/Async/AmqpAcknowledgmentsIntegrationTest.cs
+++ b/test/Apache-NMS-AMQP-Test/Integration/Async/AmqpAcknowledgmentsIntegrationTestAsync.cs
@@ -247,7 +247,7 @@ namespace NMS.AMQP.Test.Integration.Async
latch.Signal();
};
- Assert.True(latch.Wait(TimeSpan.FromMilliseconds(1000)), $"Should receive: {msgCount}, but received: {messages.Count}");
+ Assert.True(latch.Wait(TimeSpan.FromMilliseconds(3000)), $"Should receive: {msgCount}, but received: {messages.Count}");
Action<DeliveryState> dispositionMatcher = state => { Assert.AreEqual(state.Descriptor.Code, MessageSupport.ACCEPTED_INSTANCE.Descriptor.Code); };
diff --git a/test/Apache-NMS-AMQP-Test/Integration/Async/ConnectionIntegrationTest.cs b/test/Apache-NMS-AMQP-Test/Integration/Async/ConnectionIntegrationTestAsync.cs
similarity index 100%
rename from test/Apache-NMS-AMQP-Test/Integration/Async/ConnectionIntegrationTest.cs
rename to test/Apache-NMS-AMQP-Test/Integration/Async/ConnectionIntegrationTestAsync.cs
diff --git a/test/Apache-NMS-AMQP-Test/Integration/Async/ConsumerIntegrationTest.cs b/test/Apache-NMS-AMQP-Test/Integration/Async/ConsumerIntegrationTestAsync.cs
similarity index 100%
rename from test/Apache-NMS-AMQP-Test/Integration/Async/ConsumerIntegrationTest.cs
rename to test/Apache-NMS-AMQP-Test/Integration/Async/ConsumerIntegrationTestAsync.cs
diff --git a/test/Apache-NMS-AMQP-Test/Integration/Async/FailoverIntegrationTest.cs b/test/Apache-NMS-AMQP-Test/Integration/Async/FailoverIntegrationTestAsync.cs
similarity index 99%
rename from test/Apache-NMS-AMQP-Test/Integration/Async/FailoverIntegrationTest.cs
rename to test/Apache-NMS-AMQP-Test/Integration/Async/FailoverIntegrationTestAsync.cs
index 74f66ad..9cae2d9 100644
--- a/test/Apache-NMS-AMQP-Test/Integration/Async/FailoverIntegrationTest.cs
+++ b/test/Apache-NMS-AMQP-Test/Integration/Async/FailoverIntegrationTestAsync.cs
@@ -139,7 +139,7 @@ namespace NMS.AMQP.Test.Integration.Async
await connection.StartAsync();
- Assert.True(originalConnected.WaitOne(TimeSpan.FromSeconds(5)), "Should connect to original peer");
+ Assert.True(originalConnected.WaitOne(TimeSpan.FromSeconds(10)), "Should connect to original peer");
// Post Failover Expectations of FinalPeer
finalPeer.ExpectSaslAnonymous();
@@ -149,7 +149,7 @@ namespace NMS.AMQP.Test.Integration.Async
await connection.CreateSessionAsync(AcknowledgementMode.AutoAcknowledge);
- Assert.True(finalConnected.WaitOne(TimeSpan.FromSeconds(5)), "Should connect to final peer");
+ Assert.True(finalConnected.WaitOne(TimeSpan.FromSeconds(10)), "Should connect to final peer");
// Shut it down
finalPeer.ExpectClose();
@@ -564,7 +564,7 @@ namespace NMS.AMQP.Test.Integration.Async
ISession session = await connection.CreateSessionAsync();
- Assert.True(finalConnected.WaitOne(TimeSpan.FromSeconds(5)), "Should connect to final peer");
+ Assert.True(finalConnected.WaitOne(TimeSpan.FromSeconds(10)), "Should connect to final peer");
await session.CloseAsync();
await connection.CloseAsync();
@@ -627,7 +627,7 @@ namespace NMS.AMQP.Test.Integration.Async
Assert.IsNull(await consumer.ReceiveAsync(TimeSpan.FromMilliseconds(500)));
- Assert.True(finalConnected.WaitOne(TimeSpan.FromSeconds(5)), "Should connect to final peer");
+ Assert.True(finalConnected.WaitOne(TimeSpan.FromSeconds(10)), "Should connect to final peer");
await consumer.CloseAsync();
@@ -689,7 +689,7 @@ namespace NMS.AMQP.Test.Integration.Async
IQueue queue = await session.GetQueueAsync("myQueue");
IMessageProducer producer = await session.CreateProducerAsync(queue);
- Assert.True(finalConnected.WaitOne(TimeSpan.FromSeconds(5)), "Should connect to final peer");
+ Assert.True(finalConnected.WaitOne(TimeSpan.FromSeconds(10)), "Should connect to final peer");
await producer.CloseAsync();
@@ -898,7 +898,7 @@ namespace NMS.AMQP.Test.Integration.Async
originalPeer.ExpectBegin();
string dynamicAddress1 = "myTempTopicAddress";
originalPeer.ExpectTempTopicCreationAttach(dynamicAddress1);
- originalPeer.DropAfterLastMatcher();
+ originalPeer.DropAfterLastMatcher(100); // Give original side some time to process
NmsConnection connection = await EstablishAnonymousConnection(originalPeer, finalPeer);
@@ -931,7 +931,7 @@ namespace NMS.AMQP.Test.Integration.Async
ISession session = await connection.CreateSessionAsync(AcknowledgementMode.AutoAcknowledge);
ITemporaryTopic temporaryTopic = await session.CreateTemporaryTopicAsync();
- Assert.True(finalConnected.WaitOne(TimeSpan.FromSeconds(5)), "Should connect to final peer");
+ Assert.True(finalConnected.WaitOne(TimeSpan.FromSeconds(10)), "Should connect to final peer");
// Delete the temporary Topic and close the session.
finalPeer.ExpectDetach(expectClosed: true, sendResponse: true, replyClosed: true);
diff --git a/test/Apache-NMS-AMQP-Test/Integration/Async/MessageDeliveryTimeTest.cs b/test/Apache-NMS-AMQP-Test/Integration/Async/MessageDeliveryTimeTestAsync.cs
similarity index 100%
rename from test/Apache-NMS-AMQP-Test/Integration/Async/MessageDeliveryTimeTest.cs
rename to test/Apache-NMS-AMQP-Test/Integration/Async/MessageDeliveryTimeTestAsync.cs
diff --git a/test/Apache-NMS-AMQP-Test/Integration/Async/MessageExpirationIntegrationTest.cs b/test/Apache-NMS-AMQP-Test/Integration/Async/MessageExpirationIntegrationTestAsync.cs
similarity index 100%
rename from test/Apache-NMS-AMQP-Test/Integration/Async/MessageExpirationIntegrationTest.cs
rename to test/Apache-NMS-AMQP-Test/Integration/Async/MessageExpirationIntegrationTestAsync.cs
diff --git a/test/Apache-NMS-AMQP-Test/Integration/Async/NMSConsumerIntegrationTest.cs b/test/Apache-NMS-AMQP-Test/Integration/Async/NMSConsumerIntegrationTestAsync.cs
similarity index 100%
rename from test/Apache-NMS-AMQP-Test/Integration/Async/NMSConsumerIntegrationTest.cs
rename to test/Apache-NMS-AMQP-Test/Integration/Async/NMSConsumerIntegrationTestAsync.cs
diff --git a/test/Apache-NMS-AMQP-Test/Integration/Async/NMSContextIntegrationTest.cs b/test/Apache-NMS-AMQP-Test/Integration/Async/NMSContextIntegrationTestAsync.cs
similarity index 100%
rename from test/Apache-NMS-AMQP-Test/Integration/Async/NMSContextIntegrationTest.cs
rename to test/Apache-NMS-AMQP-Test/Integration/Async/NMSContextIntegrationTestAsync.cs
diff --git a/test/Apache-NMS-AMQP-Test/Integration/Async/NMSProducerIntegrationTest.cs b/test/Apache-NMS-AMQP-Test/Integration/Async/NMSProducerIntegrationTestAsync.cs
similarity index 100%
rename from test/Apache-NMS-AMQP-Test/Integration/Async/NMSProducerIntegrationTest.cs
rename to test/Apache-NMS-AMQP-Test/Integration/Async/NMSProducerIntegrationTestAsync.cs
diff --git a/test/Apache-NMS-AMQP-Test/Integration/Async/ProducerIntegrationAsyncTest.cs b/test/Apache-NMS-AMQP-Test/Integration/Async/ProducerIntegrationAsyncTestAsync.cs
similarity index 100%
rename from test/Apache-NMS-AMQP-Test/Integration/Async/ProducerIntegrationAsyncTest.cs
rename to test/Apache-NMS-AMQP-Test/Integration/Async/ProducerIntegrationAsyncTestAsync.cs
diff --git a/test/Apache-NMS-AMQP-Test/Integration/Async/ProducerIntegrationTest.cs b/test/Apache-NMS-AMQP-Test/Integration/Async/ProducerIntegrationTestAsync.cs
similarity index 100%
rename from test/Apache-NMS-AMQP-Test/Integration/Async/ProducerIntegrationTest.cs
rename to test/Apache-NMS-AMQP-Test/Integration/Async/ProducerIntegrationTestAsync.cs
diff --git a/test/Apache-NMS-AMQP-Test/Integration/Async/SessionIntegrationTest.cs b/test/Apache-NMS-AMQP-Test/Integration/Async/SessionIntegrationTestAsync.cs
similarity index 100%
rename from test/Apache-NMS-AMQP-Test/Integration/Async/SessionIntegrationTest.cs
rename to test/Apache-NMS-AMQP-Test/Integration/Async/SessionIntegrationTestAsync.cs
diff --git a/test/Apache-NMS-AMQP-Test/Integration/Async/SubscriptionsIntegrationTest.cs b/test/Apache-NMS-AMQP-Test/Integration/Async/SubscriptionsIntegrationTestAsync.cs
similarity index 100%
rename from test/Apache-NMS-AMQP-Test/Integration/Async/SubscriptionsIntegrationTest.cs
rename to test/Apache-NMS-AMQP-Test/Integration/Async/SubscriptionsIntegrationTestAsync.cs
diff --git a/test/Apache-NMS-AMQP-Test/Integration/Async/TemporaryQueueIntegrationTest.cs b/test/Apache-NMS-AMQP-Test/Integration/Async/TemporaryQueueIntegrationTestAsync.cs
similarity index 100%
rename from test/Apache-NMS-AMQP-Test/Integration/Async/TemporaryQueueIntegrationTest.cs
rename to test/Apache-NMS-AMQP-Test/Integration/Async/TemporaryQueueIntegrationTestAsync.cs
diff --git a/test/Apache-NMS-AMQP-Test/Integration/Async/TemporaryTopicIntegrationTest.cs b/test/Apache-NMS-AMQP-Test/Integration/Async/TemporaryTopicIntegrationTestAsync.cs
similarity index 100%
rename from test/Apache-NMS-AMQP-Test/Integration/Async/TemporaryTopicIntegrationTest.cs
rename to test/Apache-NMS-AMQP-Test/Integration/Async/TemporaryTopicIntegrationTestAsync.cs
diff --git a/test/Apache-NMS-AMQP-Test/Integration/Async/TransactionsIntegrationTest.cs b/test/Apache-NMS-AMQP-Test/Integration/Async/TransactionsIntegrationTestAsync.cs
similarity index 98%
rename from test/Apache-NMS-AMQP-Test/Integration/Async/TransactionsIntegrationTest.cs
rename to test/Apache-NMS-AMQP-Test/Integration/Async/TransactionsIntegrationTestAsync.cs
index d2fa236..d4739cc 100644
--- a/test/Apache-NMS-AMQP-Test/Integration/Async/TransactionsIntegrationTest.cs
+++ b/test/Apache-NMS-AMQP-Test/Integration/Async/TransactionsIntegrationTestAsync.cs
@@ -17,6 +17,7 @@
using System;
using System.Collections.Generic;
+using System.Threading;
using System.Threading.Tasks;
using Amqp;
using Amqp.Framing;
@@ -997,6 +998,12 @@ namespace NMS.AMQP.Test.Integration.Async
testPeer.WaitForAllMatchersToComplete(2000);
+ // Make sure that above line 'WaitForAllMatchersToComplete' also finishes on session/producer side
+ // In order for producer.Send to not really send, transaction on this end needs to be IsTransactionFailed
+ // in AmqpProducer if (session.IsTransacted && session.IsTransactionFailed), and sometimes producer.Send tries
+ // to send before this flag is set, which would cause actual transfer
+ await Task.Delay(10);
+
await producer.SendAsync(await session.CreateMessageAsync());
// Expect that a new link will be created in order to start the next TX.
@@ -1016,7 +1023,7 @@ namespace NMS.AMQP.Test.Integration.Async
}
}
- [Test, Timeout(20_000)]
+ [Test, Timeout(20_000), Property("FrameContext.SendDelayMs", 10)]
public async Task TestReceiveAfterCoordinatorLinkClosedDuringTX()
{
using (TestAmqpPeer testPeer = new TestAmqpPeer())
@@ -1040,7 +1047,7 @@ namespace NMS.AMQP.Test.Integration.Async
testPeer.ExpectLinkFlowRespondWithTransfer(message: CreateMessageWithNullContent());
// Close the link, the messages should now just get dropped on the floor.
- testPeer.RemotelyCloseLastCoordinatorLink();
+ testPeer.RemotelyCloseLastCoordinatorLink(1000);
IMessageConsumer consumer = await session.CreateConsumerAsync(queue);
@@ -1380,7 +1387,7 @@ namespace NMS.AMQP.Test.Integration.Async
IMessageConsumer consumer = await session.CreateConsumerAsync(queue);
for (int i = 0; i < messageCount; i++)
{
- IMessage message = await consumer.ReceiveAsync(TimeSpan.FromMilliseconds(500));
+ IMessage message = await consumer.ReceiveAsync(TimeSpan.FromMilliseconds(1000));
Assert.IsNotNull(message);
Assert.AreEqual(i, message.Properties.GetInt(TestAmqpPeer.MESSAGE_NUMBER));
}
diff --git a/test/Apache-NMS-AMQP-Test/Integration/FailoverIntegrationTest.cs b/test/Apache-NMS-AMQP-Test/Integration/FailoverIntegrationTest.cs
index d853df0..42919cc 100644
--- a/test/Apache-NMS-AMQP-Test/Integration/FailoverIntegrationTest.cs
+++ b/test/Apache-NMS-AMQP-Test/Integration/FailoverIntegrationTest.cs
@@ -149,7 +149,7 @@ namespace NMS.AMQP.Test.Integration
connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
- Assert.True(finalConnected.WaitOne(TimeSpan.FromSeconds(5)), "Should connect to final peer");
+ Assert.True(finalConnected.WaitOne(TimeSpan.FromSeconds(10)), "Should connect to final peer");
// Shut it down
finalPeer.ExpectClose();
@@ -564,7 +564,7 @@ namespace NMS.AMQP.Test.Integration
ISession session = connection.CreateSession();
- Assert.True(finalConnected.WaitOne(TimeSpan.FromSeconds(5)), "Should connect to final peer");
+ Assert.True(finalConnected.WaitOne(TimeSpan.FromSeconds(10)), "Should connect to final peer");
session.Close();
connection.Close();
@@ -627,7 +627,7 @@ namespace NMS.AMQP.Test.Integration
Assert.IsNull(consumer.Receive(TimeSpan.FromMilliseconds(500)));
- Assert.True(finalConnected.WaitOne(TimeSpan.FromSeconds(5)), "Should connect to final peer");
+ Assert.True(finalConnected.WaitOne(TimeSpan.FromSeconds(10)), "Should connect to final peer");
consumer.Close();
@@ -689,7 +689,7 @@ namespace NMS.AMQP.Test.Integration
IQueue queue = session.GetQueue("myQueue");
IMessageProducer producer = session.CreateProducer(queue);
- Assert.True(finalConnected.WaitOne(TimeSpan.FromSeconds(5)), "Should connect to final peer");
+ Assert.True(finalConnected.WaitOne(TimeSpan.FromSeconds(10)), "Should connect to final peer");
producer.Close();
@@ -898,7 +898,7 @@ namespace NMS.AMQP.Test.Integration
originalPeer.ExpectBegin();
string dynamicAddress1 = "myTempTopicAddress";
originalPeer.ExpectTempTopicCreationAttach(dynamicAddress1);
- originalPeer.DropAfterLastMatcher();
+ originalPeer.DropAfterLastMatcher(100); // Give original side some time to process
NmsConnection connection = EstablishAnonymousConnection(originalPeer, finalPeer);
@@ -931,7 +931,7 @@ namespace NMS.AMQP.Test.Integration
ISession session = connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
ITemporaryTopic temporaryTopic = session.CreateTemporaryTopic();
- Assert.True(finalConnected.WaitOne(TimeSpan.FromSeconds(5)), "Should connect to final peer");
+ Assert.True(finalConnected.WaitOne(TimeSpan.FromSeconds(10)), "Should connect to final peer");
// Delete the temporary Topic and close the session.
finalPeer.ExpectDetach(expectClosed: true, sendResponse: true, replyClosed: true);
diff --git a/test/Apache-NMS-AMQP-Test/Integration/TransactionsIntegrationTest.cs b/test/Apache-NMS-AMQP-Test/Integration/TransactionsIntegrationTest.cs
index 203155d..dd510ed 100644
--- a/test/Apache-NMS-AMQP-Test/Integration/TransactionsIntegrationTest.cs
+++ b/test/Apache-NMS-AMQP-Test/Integration/TransactionsIntegrationTest.cs
@@ -17,7 +17,7 @@
using System;
using System.Collections.Generic;
-using System.Transactions;
+using System.Threading;
using Amqp;
using Amqp.Framing;
using Amqp.Transactions;
@@ -997,6 +997,12 @@ namespace NMS.AMQP.Test.Integration
testPeer.WaitForAllMatchersToComplete(2000);
+ // Make sure that above line 'WaitForAllMatchersToComplete' also finishes on session/producer side
+ // In order for producer.Send to not really send, transaction on this end needs to be IsTransactionFailed
+ // in AmqpProducer if (session.IsTransacted && session.IsTransactionFailed), and sometimes producer.Send tries
+ // to send before this flag is set, which would cause actual transfer
+ Thread.Sleep(10);
+
producer.Send(session.CreateMessage());
// Expect that a new link will be created in order to start the next TX.
@@ -1016,7 +1022,7 @@ namespace NMS.AMQP.Test.Integration
}
}
- [Test, Timeout(20_000)]
+ [Test, Timeout(20_000), Property("FrameContext.SendDelayMs", 10)]
public void TestReceiveAfterCoordinatorLinkClosedDuringTX()
{
using (TestAmqpPeer testPeer = new TestAmqpPeer())
@@ -1040,7 +1046,7 @@ namespace NMS.AMQP.Test.Integration
testPeer.ExpectLinkFlowRespondWithTransfer(message: CreateMessageWithNullContent());
// Close the link, the messages should now just get dropped on the floor.
- testPeer.RemotelyCloseLastCoordinatorLink();
+ testPeer.RemotelyCloseLastCoordinatorLink(100);
IMessageConsumer consumer = session.CreateConsumer(queue);
@@ -1380,7 +1386,7 @@ namespace NMS.AMQP.Test.Integration
IMessageConsumer consumer = session.CreateConsumer(queue);
for (int i = 0; i < messageCount; i++)
{
- IMessage message = consumer.Receive(TimeSpan.FromMilliseconds(500));
+ IMessage message = consumer.Receive(TimeSpan.FromMilliseconds(1000));
Assert.IsNotNull(message);
Assert.AreEqual(i, message.Properties.GetInt(TestAmqpPeer.MESSAGE_NUMBER));
}
diff --git a/test/Apache-NMS-AMQP-Test/TestAmqp/Matchers/FrameContext.cs b/test/Apache-NMS-AMQP-Test/TestAmqp/Matchers/FrameContext.cs
index c8ad027..77f4733 100644
--- a/test/Apache-NMS-AMQP-Test/TestAmqp/Matchers/FrameContext.cs
+++ b/test/Apache-NMS-AMQP-Test/TestAmqp/Matchers/FrameContext.cs
@@ -17,10 +17,12 @@
using System;
using System.IO;
+using System.Threading;
using Amqp;
using Amqp.Framing;
using Amqp.Types;
using NMS.AMQP.Test.TestAmqp.BasicTypes;
+using NUnit.Framework;
namespace NMS.AMQP.Test.TestAmqp.Matchers
{
@@ -28,6 +30,30 @@ namespace NMS.AMQP.Test.TestAmqp.Matchers
{
private readonly Stream stream;
+ private static double? SendDelayMs
+ {
+ get
+ {
+ try
+ {
+ object value = TestContext.CurrentContext.Test.Properties.Get("FrameContext.SendDelayMs");
+ if (value != null)
+ {
+ return Convert.ToDouble(value);
+ }
+ }
+ catch (Exception)
+ {
+ // empty
+ }
+
+ return null;
+ }
+ }
+
+
+
+
public FrameContext(Stream stream, ushort channel)
{
this.stream = stream;
@@ -47,6 +73,11 @@ namespace NMS.AMQP.Test.TestAmqp.Matchers
private void Send(Action<ByteBuffer> encode)
{
+ if (SendDelayMs != null)
+ {
+ Thread.Sleep(TimeSpan.FromMilliseconds(SendDelayMs.Value));
+ }
+
ByteBuffer buffer = new ByteBuffer(128, true);
encode(buffer);
stream.Write(buffer.Buffer, buffer.Offset, buffer.Length);
diff --git a/test/Apache-NMS-AMQP-Test/TestAmqp/Matchers/FrameMatcher.cs b/test/Apache-NMS-AMQP-Test/TestAmqp/Matchers/FrameMatcher.cs
index 3a9b52d..4410ffc 100644
--- a/test/Apache-NMS-AMQP-Test/TestAmqp/Matchers/FrameMatcher.cs
+++ b/test/Apache-NMS-AMQP-Test/TestAmqp/Matchers/FrameMatcher.cs
@@ -17,6 +17,7 @@
using System;
using System.Collections.Generic;
+using System.Diagnostics;
using System.IO;
using Amqp.Types;
using NUnit.Framework;
@@ -28,10 +29,17 @@ namespace NMS.AMQP.Test.TestAmqp.Matchers
private bool shouldContinue = true;
private readonly List<Action<FrameContext<T>>> onCompleteActions = new List<Action<FrameContext<T>>>();
private readonly List<Action<T, Amqp.Message>> assertions = new List<Action<T, Amqp.Message>>();
+ private string creationStackTrace = null;
+
+ public FrameMatcher()
+ {
+ creationStackTrace = new StackTrace(true).ToString();
+ }
+
public bool OnFrame(Stream stream, ushort channel, DescribedList describedList, Amqp.Message message)
{
Assert.IsNotNull(describedList);
- Assert.IsInstanceOf<T>(describedList, $"Wrong frame! Expected: {typeof(T).Name} but received: {describedList.GetType().Name}");
+ Assert.IsInstanceOf<T>(describedList, $"Wrong frame! Expected: {typeof(T).Name} but received: {describedList.GetType().Name}. Stack trace of expected frame being set: {creationStackTrace}");
T command = (T) describedList;
diff --git a/test/Apache-NMS-AMQP-Test/TestAmqp/TestAmqpPeer.cs b/test/Apache-NMS-AMQP-Test/TestAmqp/TestAmqpPeer.cs
index eccf3e9..e61cc88 100644
--- a/test/Apache-NMS-AMQP-Test/TestAmqp/TestAmqpPeer.cs
+++ b/test/Apache-NMS-AMQP-Test/TestAmqp/TestAmqpPeer.cs
@@ -1033,12 +1033,12 @@ namespace NMS.AMQP.Test.TestAmqp
RemotelyCloseLastCoordinatorLink(expectDetachResponse: true, closed: true, error: new Error(ErrorCode.TransactionRollback) { Description = "Discharge of TX failed." });
}
- public void RemotelyCloseLastCoordinatorLink()
+ public void RemotelyCloseLastCoordinatorLink(int delayMs = 0)
{
- RemotelyCloseLastCoordinatorLink(expectDetachResponse: true, closed: true, error: new Error(ErrorCode.TransactionRollback) { Description = "Discharge of TX failed." });
+ RemotelyCloseLastCoordinatorLink(expectDetachResponse: true, closed: true, error: new Error(ErrorCode.TransactionRollback) { Description = "Discharge of TX failed." }, delayMs);
}
- private void RemotelyCloseLastCoordinatorLink(bool expectDetachResponse, bool closed, Error error)
+ private void RemotelyCloseLastCoordinatorLink(bool expectDetachResponse, bool closed, Error error, int delayMs = 0)
{
lock (matchersLock)
{
@@ -1051,6 +1051,11 @@ namespace NMS.AMQP.Test.TestAmqp
detach.Error = error;
}
+ if (delayMs > 0)
+ {
+ Thread.Sleep(delayMs);
+ }
+
context.SendCommand(detach);
});
@@ -1076,6 +1081,8 @@ namespace NMS.AMQP.Test.TestAmqp
if (sendClose)
{
+ testAmqpPeerRunner.StopPump();
+
var close = new Close();
this.testAmqpPeerRunner.Send(CONNECTION_CHANNEL, close);
}
diff --git a/test/Apache-NMS-AMQP-Test/TestAmqp/TestAmqpPeerRunner.cs b/test/Apache-NMS-AMQP-Test/TestAmqp/TestAmqpPeerRunner.cs
index e0bb031..cb69fe5 100644
--- a/test/Apache-NMS-AMQP-Test/TestAmqp/TestAmqpPeerRunner.cs
+++ b/test/Apache-NMS-AMQP-Test/TestAmqp/TestAmqpPeerRunner.cs
@@ -21,16 +21,15 @@ using System.Net;
using System.Net.Sockets;
using System.Threading.Tasks;
using Amqp;
-using Amqp.Framing;
using Amqp.Types;
-using NLog.Fluent;
+using NLog;
using NMS.AMQP.Test.TestAmqp.BasicTypes;
namespace NMS.AMQP.Test.TestAmqp
{
public class TestAmqpPeerRunner
{
- private static readonly NLog.Logger Logger = NLog.LogManager.GetCurrentClassLogger();
+ private static readonly Logger Logger = LogManager.GetCurrentClassLogger();
private readonly TestAmqpPeer testAmqpPeer;
private readonly IPEndPoint ip;
@@ -131,7 +130,6 @@ namespace NMS.AMQP.Test.TestAmqp
catch(Exception e)
{
Logger.Info(e);
- stream.Dispose();
}
}
@@ -163,7 +161,14 @@ namespace NMS.AMQP.Test.TestAmqp
message = Amqp.Message.Decode(buffer);
}
- return testAmqpPeer.OnFrame(stream, channel, command, message);
+ if (pumpEnabled)
+ {
+ return testAmqpPeer.OnFrame(stream, channel, command, message);
+ }
+ else
+ {
+ return false;
+ }
}
public void Send(ushort channel, DescribedList command, FrameType type = FrameType.Amqp)
@@ -175,7 +180,7 @@ namespace NMS.AMQP.Test.TestAmqp
public void Close()
{
- pumpEnabled = false;
+ StopPump();
acceptSocket?.Dispose();
acceptSocket = null;
@@ -185,5 +190,10 @@ namespace NMS.AMQP.Test.TestAmqp
s?.Dispose();
args?.Dispose();
}
+
+ public void StopPump()
+ {
+ pumpEnabled = false;
+ }
}
}
\ No newline at end of file
diff --git a/test/Apache-NMS-AMQP-Test/Utils/Synchronization/NmsSynchronizationMonitorTest.cs b/test/Apache-NMS-AMQP-Test/Utils/Synchronization/NmsSynchronizationMonitorTest.cs
index 6ac8e66..2694c72 100644
--- a/test/Apache-NMS-AMQP-Test/Utils/Synchronization/NmsSynchronizationMonitorTest.cs
+++ b/test/Apache-NMS-AMQP-Test/Utils/Synchronization/NmsSynchronizationMonitorTest.cs
@@ -238,20 +238,30 @@ namespace NMS.AMQP.Test.Utils.Synchronization
Assert.Greater(Math.Abs(a1.Last().Item2 - a2.First().Item2), 1, "A1 and A2, should be intersected by B, and not happening one right after another");
}
-
-
- [TestCase(1,3500,6)]
- [TestCase(0,2000,200)]
+
+ [TestCase(0.25,1000,6)]
+ [TestCase(0,1000,200)]
[Timeout(20_000)]
- public void TestConcurrentProducersSyncAndAsync(int sleepTimeMs, int testTimeMs, int minimumOccurences)
+ public void TestConcurrentProducersSyncAndAsync(double sleepTimeMs, int minTestTimeMs, int minimumOccurences)
{
+ TimeSpan sleepTime = TimeSpan.FromMilliseconds(sleepTimeMs);
+
EventList evListCommon = new EventList();
NmsSynchronizationMonitor syncRootA = new NmsSynchronizationMonitor();
NmsSynchronizationMonitor syncRootB = new NmsSynchronizationMonitor();
bool runTest = true;
- // int sleepTimeMs = 1;
+ // Counter helping us signal situation when we can stop running tasks/threads
+ Dictionary<String, CountdownEventEx> counters = new Dictionary<string, CountdownEventEx>()
+ {
+ ["A1"] = new CountdownEventEx((int) Math.Ceiling(minimumOccurences / 3.0)),
+ ["B1"] = new CountdownEventEx((int) Math.Ceiling(minimumOccurences / 3.0)),
+ ["A2"] = new CountdownEventEx((int) Math.Ceiling(minimumOccurences / 3.0)),
+ ["B2"] = new CountdownEventEx((int) Math.Ceiling(minimumOccurences / 3.0)),
+ ["A3"] = new CountdownEventEx((int) Math.Ceiling(minimumOccurences / 3.0)),
+ ["B3"] = new CountdownEventEx((int) Math.Ceiling(minimumOccurences / 3.0)),
+ };
var task1 = Task.Run(async () =>
{
@@ -261,42 +271,45 @@ namespace NMS.AMQP.Test.Utils.Synchronization
var lockA = (counter % 2 == 0) ? syncRootA.Lock() : await syncRootA.LockAsync();
using (lockA)
{
- await Task.Delay(sleepTimeMs);
+ await Task.Delay(sleepTime);
await Task.Yield();
evListCommon.Add("A1");
- await Task.Delay(sleepTimeMs);
+ await Task.Delay(sleepTime);
await Task.Yield();
evListCommon.Add("A1");
- await Task.Delay(sleepTimeMs);
+ await Task.Delay(sleepTime);
await Task.Yield();
evListCommon.Add("A1");
- await Task.Delay(sleepTimeMs);
+ counters["A1"].TrySignal();
+ await Task.Delay(sleepTime);
await Task.Yield();
var lockB = (counter % 2 == 0) ? syncRootB.Lock() : await syncRootB.LockAsync();
using (lockB)
{
evListCommon.Add("B1");
- await Task.Delay(sleepTimeMs);
+ await Task.Delay(sleepTime);
await Task.Yield();
evListCommon.Add("B1");
- await Task.Delay(sleepTimeMs);
+ await Task.Delay(sleepTime);
await Task.Yield();
evListCommon.Add("B1");
- await Task.Delay(sleepTimeMs);
+ counters["B1"].TrySignal();
+ await Task.Delay(sleepTime);
await Task.Yield();
}
- await Task.Delay(sleepTimeMs);
+ await Task.Delay(sleepTime);
await Task.Yield();
evListCommon.Add("A1");
- await Task.Delay(sleepTimeMs);
+ await Task.Delay(sleepTime);
await Task.Yield();
evListCommon.Add("A1");
- await Task.Delay(sleepTimeMs);
+ await Task.Delay(sleepTime);
await Task.Yield();
evListCommon.Add("A1");
- await Task.Delay(sleepTimeMs);
+ counters["A1"].TrySignal();
+ await Task.Delay(sleepTime);
await Task.Yield();
}
@@ -317,16 +330,17 @@ namespace NMS.AMQP.Test.Utils.Synchronization
int depth = counter % 4;
if (level == depth)
{
- await Task.Delay(sleepTimeMs);
+ await Task.Delay(sleepTime);
await Task.Yield();
evListCommon.Add(symbol);
- await Task.Delay(sleepTimeMs);
+ await Task.Delay(sleepTime);
await Task.Yield();
evListCommon.Add(symbol);
- await Task.Delay(sleepTimeMs);
+ await Task.Delay(sleepTime);
await Task.Yield();
evListCommon.Add(symbol);
- await Task.Delay(sleepTimeMs);
+ counters[symbol].TrySignal();
+ await Task.Delay(sleepTime);
await Task.Yield();
}
else
@@ -339,13 +353,13 @@ namespace NMS.AMQP.Test.Utils.Synchronization
while (runTest)
{
await ResourceAccess("A2",0, syncRootA);
- await ResourceAccess("B2",0,syncRootB);
+ await ResourceAccess("B2",0, syncRootB);
counter++;
}
});
- var task3 = Task.Run(() =>
+ var task3 = new Thread(() =>
{
int counter = 0;
while (runTest)
@@ -353,45 +367,50 @@ namespace NMS.AMQP.Test.Utils.Synchronization
var lockA = (counter % 2 == 0) ? syncRootA.Lock() : syncRootA.LockAsync().GetAsyncResult();
using (lockA)
{
- Thread.Sleep(sleepTimeMs);
+ Thread.Sleep(sleepTime);
evListCommon.Add("A3");
- Thread.Sleep(sleepTimeMs);
+ Thread.Sleep(sleepTime);
evListCommon.Add("A3");
- Thread.Sleep(sleepTimeMs);
+ Thread.Sleep(sleepTime);
evListCommon.Add("A3");
- Thread.Sleep(sleepTimeMs);
+ counters["A3"].TrySignal();
+ Thread.Sleep(sleepTime);
}
var lockB = (counter % 2 == 0) ? syncRootB.Lock() : syncRootB.LockAsync().GetAsyncResult();
using (lockB)
{
evListCommon.Add("B3");
- Thread.Sleep(sleepTimeMs);
+ Thread.Sleep(sleepTime);
evListCommon.Add("B3");
- Thread.Sleep(sleepTimeMs);
+ Thread.Sleep(sleepTime);
evListCommon.Add("B3");
- Thread.Sleep(sleepTimeMs);
+ counters["B3"].TrySignal();
+ Thread.Sleep(sleepTime);
}
-
counter++;
}
});
+ task3.IsBackground = true;
+ task3.Start();
- // Let it run for one sec
- Thread.Sleep(testTimeMs);
-
+ // Let it run for a short while
+ Thread.Sleep(minTestTimeMs);
+
+ // Wait for all counter events to get to zero so min occurences should be met
+ counters.ToList().ForEach(counterEvent => counterEvent.Value.Wait());
+
runTest = false;
task1.Wait();
task2.Wait();
- task3.Wait();
+ task3.Join();
var sequenceCommon = evListCommon.ToString();
var sequenceA = evListCommon.ToString("A");
var sequenceB = evListCommon.ToString("B");
- // remove B locks interfering with A sequence of rvalidating task1
-
+ // remove B locks interfering with A sequence of validating task1
Enumerable.Range(0,10).ToList().ForEach( (i) => sequenceCommon = sequenceCommon
.Replace("A1B2", "A1")
.Replace("A1B3", "A1")
@@ -429,9 +448,34 @@ namespace NMS.AMQP.Test.Utils.Synchronization
sequenceB = sequenceB.Replace("B3B3B3", "");
- Assert.AreEqual(0,sequenceA.Length, "There were illegal sequences of execution for resource A: "+sequenceA);
- Assert.AreEqual(0,sequenceB.Length, "There were illegal sequences of execution for resource B: "+sequenceB);
+ Assert.AreEqual(0, sequenceA.Length, "There were illegal sequences of execution for resource A: " + sequenceA);
+ Assert.AreEqual(0, sequenceB.Length, "There were illegal sequences of execution for resource B: " + sequenceB);
}
+
+ /// <summary>
+ /// Extended class to have signaling method that does not throw if count gets to minus
+ /// </summary>
+ private class CountdownEventEx : CountdownEvent
+ {
+ public CountdownEventEx(int initialCount) : base(initialCount)
+ {
+ }
+
+ public bool TrySignal()
+ {
+ try
+ {
+ this.Signal();
+ return true;
+ }
+ catch (Exception)
+ {
+ // empty
+ }
+
+ return false;
+ }
+ }
}
}
\ No newline at end of file