Skip to content

Commit cc9c0d3

Browse files
michaelklishinlukebakken
michaelklishin
authored and
lukebakken
committed
Merge pull request #1141 from rabbitmq/lukebakken/fix-appveyor
Fix and update AppVeyor build (cherry picked from commit 4628927)
1 parent 860cc9f commit cc9c0d3

File tree

5 files changed

+182
-109
lines changed

5 files changed

+182
-109
lines changed

‎appveyor.yml

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -3,14 +3,13 @@ version: "8.0.0.{build}"
33
platform: Any CPU
44
configuration: Release
55
skip_tags: true
6-
skip_branch_with_pr: true
76
image: Visual Studio 2019
87

98
cache:
109
# Note: this must match the $rabbitmq_installer_path and $erlang_installer_path values in
1110
# tools\appveyor\install.ps1
12-
- "%HOMEDRIVE%%HOMEPATH%\\rabbitmq-server-3.8.9.exe"
13-
- "%HOMEDRIVE%%HOMEPATH%\\otp_win64_23.1.2.exe"
11+
- "%HOMEDRIVE%%HOMEPATH%\\rabbitmq-server-3.9.13.exe"
12+
- "%HOMEDRIVE%%HOMEPATH%\\otp_win64_24.2.1.exe"
1413

1514
install:
1615
- ps: .\tools\appveyor\install.ps1

‎projects/Unit/TestConnectionRecovery.cs

Lines changed: 143 additions & 72 deletions
Original file line numberDiff line numberDiff line change
@@ -64,10 +64,24 @@ public void Dispose()
6464
[Collection("NoParallelization")]
6565
public class TestConnectionRecovery : IntegrationFixture
6666
{
67+
private readonly byte[] _messageBody;
68+
private readonly ushort _totalMessageCount = 1024;
69+
private readonly ushort _closeAtCount = 16;
70+
private string _queueName;
71+
72+
public TestConnectionRecovery()
73+
{
74+
var rnd = new Random();
75+
_messageBody = new byte[4096];
76+
rnd.NextBytes(_messageBody);
77+
}
78+
6779
protected override void SetUp()
6880
{
81+
_queueName = $"TestConnectionRecovery-queue-{Guid.NewGuid()}";
6982
_conn = CreateAutorecoveringConnection();
7083
_model = _conn.CreateModel();
84+
_model.QueueDelete(_queueName);
7185
}
7286

7387
public override void Dispose()
@@ -76,13 +90,70 @@ public override void Dispose()
7690

7791
}
7892

79-
[Fact]
93+
[Fact(Skip="TODO flaky")]
8094
public void TestBasicAckAfterChannelRecovery()
8195
{
82-
var latch = new ManualResetEventSlim(false);
83-
var cons = new AckingBasicConsumer(_model, latch, CloseAndWaitForRecovery);
96+
var allMessagesSeenLatch = new ManualResetEventSlim(false);
97+
var cons = new AckingBasicConsumer(_model, _totalMessageCount, allMessagesSeenLatch);
98+
99+
string queueName = _model.QueueDeclare(_queueName, false, false, false, null).QueueName;
100+
Assert.Equal(queueName, _queueName);
101+
102+
_model.BasicQos(0, 1, false);
103+
string consumerTag = _model.BasicConsume(queueName, false, cons);
104+
105+
ManualResetEventSlim sl = PrepareForShutdown(_conn);
106+
ManualResetEventSlim rl = PrepareForRecovery(_conn);
107+
108+
PublishMessagesWhileClosingConn(queueName);
84109

85-
TestDelayedBasicAckNackAfterChannelRecovery(cons, latch);
110+
Wait(sl);
111+
Wait(rl);
112+
Wait(allMessagesSeenLatch);
113+
}
114+
115+
[Fact(Skip="TODO flaky")]
116+
public void TestBasicNackAfterChannelRecovery()
117+
{
118+
var allMessagesSeenLatch = new ManualResetEventSlim(false);
119+
var cons = new NackingBasicConsumer(_model, _totalMessageCount, allMessagesSeenLatch);
120+
121+
string queueName = _model.QueueDeclare(_queueName, false, false, false, null).QueueName;
122+
Assert.Equal(queueName, _queueName);
123+
124+
_model.BasicQos(0, 1, false);
125+
string consumerTag = _model.BasicConsume(queueName, false, cons);
126+
127+
ManualResetEventSlim sl = PrepareForShutdown(_conn);
128+
ManualResetEventSlim rl = PrepareForRecovery(_conn);
129+
130+
PublishMessagesWhileClosingConn(queueName);
131+
132+
Wait(sl);
133+
Wait(rl);
134+
Wait(allMessagesSeenLatch);
135+
}
136+
137+
[Fact(Skip="TODO flaky")]
138+
public void TestBasicRejectAfterChannelRecovery()
139+
{
140+
var allMessagesSeenLatch = new ManualResetEventSlim(false);
141+
var cons = new RejectingBasicConsumer(_model, _totalMessageCount, allMessagesSeenLatch);
142+
143+
string queueName = _model.QueueDeclare(_queueName, false, false, false, null).QueueName;
144+
Assert.Equal(queueName, _queueName);
145+
146+
_model.BasicQos(0, 1, false);
147+
string consumerTag = _model.BasicConsume(queueName, false, cons);
148+
149+
ManualResetEventSlim sl = PrepareForShutdown(_conn);
150+
ManualResetEventSlim rl = PrepareForRecovery(_conn);
151+
152+
PublishMessagesWhileClosingConn(queueName);
153+
154+
Wait(sl);
155+
Wait(rl);
156+
Wait(allMessagesSeenLatch);
86157
}
87158

88159
[Fact]
@@ -91,7 +162,7 @@ public void TestBasicAckAfterBasicGetAndChannelRecovery()
91162
string q = GenerateQueueName();
92163
_model.QueueDeclare(q, false, false, false, null);
93164
// create an offset
94-
_model.BasicPublish("", q, ReadOnlyMemory<byte>.Empty);
165+
_model.BasicPublish("", q, _messageBody);
95166
Thread.Sleep(50);
96167
BasicGetResult g = _model.BasicGet(q, false);
97168
CloseAndWaitForRecovery();
@@ -115,7 +186,7 @@ public void TestBasicAckEventHandlerRecovery()
115186
CloseAndWaitForRecovery();
116187
Assert.True(_model.IsOpen);
117188

118-
WithTemporaryNonExclusiveQueue(_model, (m, q) => m.BasicPublish("", q, _encoding.GetBytes("")));
189+
WithTemporaryNonExclusiveQueue(_model, (m, q) => m.BasicPublish("", q, _messageBody));
119190
Wait(latch);
120191
}
121192

@@ -224,24 +295,6 @@ public void TestBasicModelRecoveryOnServerRestart()
224295
Assert.True(_model.IsOpen);
225296
}
226297

227-
[Fact]
228-
public void TestBasicNackAfterChannelRecovery()
229-
{
230-
var latch = new ManualResetEventSlim(false);
231-
var cons = new NackingBasicConsumer(_model, latch, CloseAndWaitForRecovery);
232-
233-
TestDelayedBasicAckNackAfterChannelRecovery(cons, latch);
234-
}
235-
236-
[Fact]
237-
public void TestBasicRejectAfterChannelRecovery()
238-
{
239-
var latch = new ManualResetEventSlim(false);
240-
var cons = new RejectingBasicConsumer(_model, latch, CloseAndWaitForRecovery);
241-
242-
TestDelayedBasicAckNackAfterChannelRecovery(cons, latch);
243-
}
244-
245298
[Fact]
246299
public void TestBlockedListenersRecovery()
247300
{
@@ -780,7 +833,7 @@ public void TestRecoverTopologyOnDisposedChannel()
780833
var latch = new ManualResetEventSlim(false);
781834
cons.Received += (s, args) => latch.Set();
782835

783-
_model.BasicPublish("", q);
836+
_model.BasicPublish("", q, _messageBody);
784837
Wait(latch);
785838

786839
_model.QueueUnbind(q, x, rk);
@@ -817,7 +870,7 @@ public void TestPublishRpcRightAfterReconnect()
817870
{
818871
try
819872
{
820-
_model.BasicPublish(string.Empty, testQueueName, ref properties, ReadOnlyMemory<byte>.Empty);
873+
_model.BasicPublish(string.Empty, testQueueName, ref properties, _messageBody);
821874
}
822875
catch (Exception e)
823876
{
@@ -968,8 +1021,7 @@ internal void AssertExchangeRecovery(IModel m, string x)
9681021
{
9691022
string rk = "routing-key";
9701023
m.QueueBind(q, x, rk);
971-
byte[] mb = RandomMessageBody();
972-
m.BasicPublish(x, rk, mb);
1024+
m.BasicPublish(x, rk, _messageBody);
9731025

9741026
Assert.True(WaitForConfirms(m));
9751027
m.ExchangeDeclarePassive(x);
@@ -987,7 +1039,7 @@ internal void AssertQueueRecovery(IModel m, string q, bool exclusive)
9871039
m.QueueDeclarePassive(q);
9881040
QueueDeclareOk ok1 = m.QueueDeclare(q, false, exclusive, false, null);
9891041
Assert.Equal(0u, ok1.MessageCount);
990-
m.BasicPublish("", q, _encoding.GetBytes(""));
1042+
m.BasicPublish("", q, _messageBody);
9911043
Assert.True(WaitForConfirms(m));
9921044
QueueDeclareOk ok2 = m.QueueDeclare(q, false, exclusive, false, null);
9931045
Assert.Equal(1u, ok2.MessageCount);
@@ -1017,18 +1069,29 @@ internal void CloseAndWaitForRecovery(AutorecoveringConnection conn)
10171069
Wait(rl);
10181070
}
10191071

1020-
internal static ManualResetEventSlim PrepareForRecovery(AutorecoveringConnection conn)
1072+
internal void CloseAndWaitForShutdown(AutorecoveringConnection conn)
1073+
{
1074+
ManualResetEventSlim sl = PrepareForShutdown(conn);
1075+
CloseConnection(conn);
1076+
Wait(sl);
1077+
}
1078+
1079+
internal ManualResetEventSlim PrepareForRecovery(IConnection conn)
10211080
{
10221081
var latch = new ManualResetEventSlim(false);
1023-
conn.RecoverySucceeded += (source, ea) => latch.Set();
1082+
1083+
AutorecoveringConnection aconn = conn as AutorecoveringConnection;
1084+
aconn.RecoverySucceeded += (source, ea) => latch.Set();
10241085

10251086
return latch;
10261087
}
10271088

10281089
internal static ManualResetEventSlim PrepareForShutdown(IConnection conn)
10291090
{
10301091
var latch = new ManualResetEventSlim(false);
1031-
conn.ConnectionShutdown += (c, args) => latch.Set();
1092+
1093+
AutorecoveringConnection aconn = conn as AutorecoveringConnection;
1094+
aconn.ConnectionShutdown += (c, args) => latch.Set();
10321095

10331096
return latch;
10341097
}
@@ -1052,36 +1115,48 @@ internal void RestartServerAndWaitForRecovery(AutorecoveringConnection conn)
10521115
Wait(rl);
10531116
}
10541117

1055-
internal void TestDelayedBasicAckNackAfterChannelRecovery(TestBasicConsumer1 cons, ManualResetEventSlim latch)
1118+
internal void WaitForRecovery()
10561119
{
1057-
string q = _model.QueueDeclare(GenerateQueueName(), false, false, false, null).QueueName;
1058-
int n = 30;
1059-
_model.BasicQos(0, 1, false);
1060-
_model.BasicConsume(q, false, cons);
1061-
1062-
AutorecoveringConnection publishingConn = CreateAutorecoveringConnection();
1063-
IModel publishingModel = publishingConn.CreateModel();
1120+
Wait(PrepareForRecovery((AutorecoveringConnection)_conn));
1121+
}
10641122

1065-
for (int i = 0; i < n; i++)
1066-
{
1067-
publishingModel.BasicPublish("", q, _encoding.GetBytes(""));
1068-
}
1123+
internal void WaitForRecovery(AutorecoveringConnection conn)
1124+
{
1125+
Wait(PrepareForRecovery(conn));
1126+
}
10691127

1070-
Wait(latch, TimeSpan.FromSeconds(20));
1071-
_model.QueueDelete(q);
1072-
publishingModel.Close();
1073-
publishingConn.Close();
1128+
internal void WaitForShutdown()
1129+
{
1130+
Wait(PrepareForShutdown(_conn));
10741131
}
10751132

10761133
internal void WaitForShutdown(IConnection conn)
10771134
{
10781135
Wait(PrepareForShutdown(conn));
10791136
}
10801137

1081-
public class AckingBasicConsumer : TestBasicConsumer1
1138+
internal void PublishMessagesWhileClosingConn(string queueName)
10821139
{
1083-
public AckingBasicConsumer(IModel model, ManualResetEventSlim latch, Action fn)
1084-
: base(model, latch, fn)
1140+
using (AutorecoveringConnection publishingConn = CreateAutorecoveringConnection())
1141+
{
1142+
using (IModel publishingModel = publishingConn.CreateModel())
1143+
{
1144+
for (ushort i = 0; i < _totalMessageCount; i++)
1145+
{
1146+
if (i == _closeAtCount)
1147+
{
1148+
CloseConnection(_conn);
1149+
}
1150+
publishingModel.BasicPublish(string.Empty, queueName, _messageBody);
1151+
}
1152+
}
1153+
}
1154+
}
1155+
1156+
public class AckingBasicConsumer : TestBasicConsumer
1157+
{
1158+
public AckingBasicConsumer(IModel model, ushort totalMessageCount, ManualResetEventSlim allMessagesSeenLatch)
1159+
: base(model, totalMessageCount, allMessagesSeenLatch)
10851160
{
10861161
}
10871162

@@ -1091,10 +1166,10 @@ public override void PostHandleDelivery(ulong deliveryTag)
10911166
}
10921167
}
10931168

1094-
public class NackingBasicConsumer : TestBasicConsumer1
1169+
public class NackingBasicConsumer : TestBasicConsumer
10951170
{
1096-
public NackingBasicConsumer(IModel model, ManualResetEventSlim latch, Action fn)
1097-
: base(model, latch, fn)
1171+
public NackingBasicConsumer(IModel model, ushort totalMessageCount, ManualResetEventSlim allMessagesSeenLatch)
1172+
: base(model, totalMessageCount, allMessagesSeenLatch)
10981173
{
10991174
}
11001175

@@ -1104,10 +1179,10 @@ public override void PostHandleDelivery(ulong deliveryTag)
11041179
}
11051180
}
11061181

1107-
public class RejectingBasicConsumer : TestBasicConsumer1
1182+
public class RejectingBasicConsumer : TestBasicConsumer
11081183
{
1109-
public RejectingBasicConsumer(IModel model, ManualResetEventSlim latch, Action fn)
1110-
: base(model, latch, fn)
1184+
public RejectingBasicConsumer(IModel model, ushort totalMessageCount, ManualResetEventSlim allMessagesSeenLatch)
1185+
: base(model, totalMessageCount, allMessagesSeenLatch)
11111186
{
11121187
}
11131188

@@ -1117,17 +1192,17 @@ public override void PostHandleDelivery(ulong deliveryTag)
11171192
}
11181193
}
11191194

1120-
public class TestBasicConsumer1 : DefaultBasicConsumer
1195+
public class TestBasicConsumer : DefaultBasicConsumer
11211196
{
1122-
private readonly Action _action;
1123-
private readonly ManualResetEventSlim _latch;
1197+
private readonly ManualResetEventSlim _allMessagesSeenLatch;
1198+
private readonly ushort _totalMessageCount;
11241199
private ushort _counter = 0;
11251200

1126-
public TestBasicConsumer1(IModel model, ManualResetEventSlim latch, Action fn)
1201+
public TestBasicConsumer(IModel model, ushort totalMessageCount, ManualResetEventSlim allMessagesSeenLatch)
11271202
: base(model)
11281203
{
1129-
_latch = latch;
1130-
_action = fn;
1204+
_totalMessageCount = totalMessageCount;
1205+
_allMessagesSeenLatch = allMessagesSeenLatch;
11311206
}
11321207

11331208
public override void HandleBasicDeliver(string consumerTag,
@@ -1140,19 +1215,15 @@ public override void HandleBasicDeliver(string consumerTag,
11401215
{
11411216
try
11421217
{
1143-
if (deliveryTag == 7 && _counter < 10)
1144-
{
1145-
_action();
1146-
}
1147-
if (_counter == 9)
1148-
{
1149-
_latch.Set();
1150-
}
11511218
PostHandleDelivery(deliveryTag);
11521219
}
11531220
finally
11541221
{
1155-
_counter += 1;
1222+
++_counter;
1223+
if (_counter >= _totalMessageCount)
1224+
{
1225+
_allMessagesSeenLatch.Set();
1226+
}
11561227
}
11571228
}
11581229

0 commit comments

Comments
 (0)