Add test code for issue #1464

Author: lukebakken

projects/Test/Common/IntegrationFixture.cs

@@ -551,6 +551,22 @@ protected static byte[] GetRandomBody(ushort size = 1024)
             return body;
         }
 
+        protected static Task WaitForRecoveryAsync(IConnection conn)
+        {
+            TaskCompletionSource<bool> tcs = PrepareForRecovery((AutorecoveringConnection)conn);
+            return WaitAsync(tcs, "recovery succeded");
+        }
+
+        protected static TaskCompletionSource<bool> PrepareForRecovery(IConnection conn)
+        {
+            var tcs = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+
+            AutorecoveringConnection aconn = conn as AutorecoveringConnection;
+            aconn.RecoverySucceeded += (source, ea) => tcs.SetResult(true);
+
+            return tcs;
+        }
+
         public static string Now => DateTime.UtcNow.ToString("s", CultureInfo.InvariantCulture);
     }
 }

projects/Test/Common/TestConnectionRecoveryBase.cs

@@ -231,16 +231,6 @@ protected static TaskCompletionSource<bool> PrepareForShutdown(IConnection conn)
             return tcs;
         }
 
-        protected static TaskCompletionSource<bool> PrepareForRecovery(IConnection conn)
-        {
-            var tcs = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
-
-            AutorecoveringConnection aconn = conn as AutorecoveringConnection;
-            aconn.RecoverySucceeded += (source, ea) => tcs.SetResult(true);
-
-            return tcs;
-        }
-
         protected static Task<bool> WaitForConfirmsWithCancellationAsync(IChannel m)
         {
             using (var cts = new CancellationTokenSource(TimeSpan.FromSeconds(4)))
@@ -249,18 +239,6 @@ protected static Task<bool> WaitForConfirmsWithCancellationAsync(IChannel m)
             }
         }
 
-        protected Task WaitForRecoveryAsync()
-        {
-            TaskCompletionSource<bool> tcs = PrepareForRecovery((AutorecoveringConnection)_conn);
-            return WaitAsync(tcs, "recovery succeded");
-        }
-
-        internal Task WaitForRecoveryAsync(AutorecoveringConnection conn)
-        {
-            TaskCompletionSource<bool> tcs = PrepareForRecovery(conn);
-            return WaitAsync(tcs, "recovery succeeded");
-        }
-
         protected Task WaitForShutdownAsync()
         {
             TaskCompletionSource<bool> tcs = PrepareForShutdown(_conn);

projects/Test/Integration/TestToxiproxy.cs

@@ -169,6 +169,66 @@ await Assert.ThrowsAsync<AlreadyClosedException>(() =>
             _output.WriteLine($"[INFO] heartbeat timeout took {sw.Elapsed}");
         }
 
+        [SkippableFact]
+        [Trait("Category", "Toxiproxy")]
+        public async Task TestTcpReset_GH1464()
+        {
+            Skip.IfNot(AreToxiproxyTestsEnabled, "RABBITMQ_TOXIPROXY_TESTS is not set, skipping test");
+
+            ConnectionFactory cf = CreateConnectionFactory();
+            cf.Endpoint = new AmqpTcpEndpoint(IPAddress.Loopback.ToString(), ProxyPort);
+            cf.Port = ProxyPort;
+            cf.RequestedHeartbeat = TimeSpan.FromSeconds(5);
+            cf.AutomaticRecoveryEnabled = true;
+
+            var channelCreatedTcs = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+            var connectionShutdownTcs = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+
+            Task recoveryTask = Task.Run(async () =>
+            {
+                using (IConnection conn = await cf.CreateConnectionAsync())
+                {
+                    conn.ConnectionShutdown += (o, ea) =>
+                    {
+                        connectionShutdownTcs.SetResult(true);
+                    };
+
+                    using (IChannel ch = await conn.CreateChannelAsync())
+                    {
+                        channelCreatedTcs.SetResult(true);
+                        await WaitForRecoveryAsync(conn);
+                        await ch.CloseAsync();
+                    }
+
+                    await conn.CloseAsync();
+                }
+            });
+
+            Assert.True(await channelCreatedTcs.Task);
+
+            const string toxicName = "rmq-localhost-reset_peer";
+            var resetPeerToxic = new ResetPeerToxic();
+            resetPeerToxic.Name = toxicName;
+            resetPeerToxic.Attributes.Timeout = 500;
+            resetPeerToxic.Toxicity = 1.0;
+
+            var sw = new Stopwatch();
+            sw.Start();
+
+            await _rmqProxy.AddAsync(resetPeerToxic);
+            Task<Proxy> updateProxyTask = _rmqProxy.UpdateAsync();
+
+            await Task.WhenAll(updateProxyTask, connectionShutdownTcs.Task);
+
+            await _rmqProxy.RemoveToxicAsync(toxicName);
+
+            await recoveryTask;
+
+            sw.Stop();
+
+            _output.WriteLine($"[INFO] reset peer took {sw.Elapsed}");
+        }
+
         private bool AreToxiproxyTestsEnabled
         {
             get