Merge branch 'master' into eliminate.command

Author: michaelklishin

RUNNING_TESTS.md

@@ -42,10 +42,16 @@ Two options to accomplish this are covered below.
 
 ### Option One: Using a RabbitMQ Release
 
-It is possible to install and run a node using any [binary build](https://www.rabbitmq.com/download.html) suitable
-for the platform. Its [CLI tools]() then must be added to `PATH` so that `rabbitmqctl` (`rabbitmqctl.bat` on Windows)
-can be invoked directly without using an absolute file path.
+It is possible to install and run a node using any [binary build](https://www.rabbitmq.com/download.html)
+suitable for the platform. Its [CLI tools]() then must be added to `PATH` so that `rabbitmqctl` can be
+invoked directly without using an absolute file path. Note that this method does *not* work on Windows.
 
+On Windows, you must run unit tests as follows (replace `X.Y.Z` with your RabbitMQ version):
+
+```
+set RABBITMQ_RABBITMQCTL_PATH="C:\Program Files\RabbitMQ Server\rabbitmq_server-X.Y.Z\sbin\rabbitmqctl.bat"
+.\run-test.bat
+```
 
 ### Option Two: Using RabbitMQ Umbrella Repository
 

projects/Apigen/apigen/Apigen.cs

@@ -38,6 +38,8 @@
 //  Copyright (c) 2007-2020 VMware, Inc.  All rights reserved.
 //---------------------------------------------------------------------------
 
+using System;
+
 namespace RabbitMQ.Client
 {
     // time representations in mainstream languages: the horror, the horror
@@ -58,7 +60,7 @@ namespace RabbitMQ.Client
     /// timestamps are signed or unsigned.
     /// </para>
     /// </remarks>
-    public struct AmqpTimestamp
+    public struct AmqpTimestamp : IEquatable<AmqpTimestamp>
     {
         /// <summary>
         /// Construct an <see cref="AmqpTimestamp"/>.
@@ -74,6 +76,16 @@ public AmqpTimestamp(long unixTime) : this()
         /// </summary>
         public long UnixTime { get; private set; }
 
+        public bool Equals(AmqpTimestamp other) => UnixTime == other.UnixTime;
+
+        public override bool Equals(object obj) => obj is AmqpTimestamp other && Equals(other);
+
+        public override int GetHashCode() => UnixTime.GetHashCode();
+
+        public static bool operator ==(AmqpTimestamp left, AmqpTimestamp right) => left.Equals(right);
+
+        public static bool operator !=(AmqpTimestamp left, AmqpTimestamp right) => !left.Equals(right);
+
         /// <summary>
         /// Provides a debugger-friendly display.
         /// </summary>

projects/RabbitMQ.Client/client/api/AmqpTimestamp.cs

@@ -44,8 +44,7 @@ namespace RabbitMQ.Client
 {
     public interface IBasicPublishBatch
     {
-        [Obsolete("Use Add(string exchange, string routingKey, bool mandatory, IBasicProperties properties, ReadOnlyMemory<byte> body) instead. Will be replaced in version 7.0", false)]
-        void Add(string exchange, string routingKey, bool mandatory, IBasicProperties properties, byte[] body);
+        void Add(string exchange, string routingKey, bool mandatory, IBasicProperties properties, ReadOnlyMemory<byte> body);
         void Publish();
     }
 }

projects/RabbitMQ.Client/client/api/IBasicPublishBatch.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Net;
 using System.Net.Sockets;
 using System.Threading.Tasks;
 
@@ -17,6 +18,7 @@ public interface ITcpClient : IDisposable
         Socket Client { get; }
 
         Task ConnectAsync(string host, int port);
+        Task ConnectAsync(IPAddress host, int port);
 
         NetworkStream GetStream();
 

projects/RabbitMQ.Client/client/api/ITcpClient.cs

@@ -16,7 +16,7 @@ public AsyncConsumerWorkService(int concurrency) : base(concurrency)
             _startNewWorkPoolFunc = model => StartNewWorkPool(model);
         }
 
-        public void Schedule<TWork>(ModelBase model, TWork work) where TWork : Work
+        public void Schedule<TWork>(IModel model, TWork work) where TWork : Work
         {
             /*
              * rabbitmq/rabbitmq-dotnet-client#841
@@ -29,7 +29,7 @@ public void Schedule<TWork>(ModelBase model, TWork work) where TWork : Work
 
         private WorkPool StartNewWorkPool(IModel model)
         {
-            var newWorkPool = new WorkPool(model as ModelBase, _concurrency);
+            var newWorkPool = new WorkPool(model, _concurrency);
             newWorkPool.Start();
             return newWorkPool;
         }
@@ -47,13 +47,13 @@ public Task Stop(IModel model)
         class WorkPool
         {
             readonly Channel<Work> _channel;
-            readonly ModelBase _model;
+            readonly IModel _model;
             private Task _worker;
             private readonly int _concurrency;
             private SemaphoreSlim _limiter;
             private CancellationTokenSource _tokenSource;
 
-            public WorkPool(ModelBase model, int concurrency)
+            public WorkPool(IModel model, int concurrency)
             {
                 _concurrency = concurrency;
                 _model = model;
@@ -125,7 +125,7 @@ async Task LoopWithConcurrency(CancellationToken cancellationToken)
                 }
             }
 
-            static async Task HandleConcurrent(Work work, ModelBase model, SemaphoreSlim limiter)
+            static async Task HandleConcurrent(Work work, IModel model, SemaphoreSlim limiter)
             {
                 try
                 {

projects/RabbitMQ.Client/client/impl/AsyncConsumerWorkService.cs

@@ -15,20 +15,25 @@ public BasicCancel(IBasicConsumer consumer, string consumerTag) : base(consumer)
             _consumerTag = consumerTag;
         }
 
-        protected override async Task Execute(ModelBase model, IAsyncBasicConsumer consumer)
+        protected override async Task Execute(IModel model, IAsyncBasicConsumer consumer)
         {
             try
             {
                 await consumer.HandleBasicCancel(_consumerTag).ConfigureAwait(false);
             }
             catch (Exception e)
             {
+                if (!(model is ModelBase modelBase))
+                {
+                    return;
+                }
+
                 var details = new Dictionary<string, object>
                 {
                     {"consumer", consumer},
                     {"context",  "HandleBasicCancel"}
                 };
-                model.OnCallbackException(CallbackExceptionEventArgs.Build(e, details));
+                modelBase.OnCallbackException(CallbackExceptionEventArgs.Build(e, details));
             }
         }
     }

projects/RabbitMQ.Client/client/impl/BasicCancel.cs

@@ -15,20 +15,25 @@ public BasicCancelOk(IBasicConsumer consumer, string consumerTag) : base(consume
             _consumerTag = consumerTag;
         }
 
-        protected override async Task Execute(ModelBase model, IAsyncBasicConsumer consumer)
+        protected override async Task Execute(IModel model, IAsyncBasicConsumer consumer)
         {
             try
             {
                 await consumer.HandleBasicCancelOk(_consumerTag).ConfigureAwait(false);
             }
             catch (Exception e)
             {
+                if (!(model is ModelBase modelBase))
+                {
+                    return;
+                }
+
                 var details = new Dictionary<string, object>()
                 {
                     {"consumer", consumer},
                     {"context",  "HandleBasicCancelOk"}
                 };
-                model.OnCallbackException(CallbackExceptionEventArgs.Build(e, details));
+                modelBase.OnCallbackException(CallbackExceptionEventArgs.Build(e, details));
             }
         }
     }

projects/RabbitMQ.Client/client/impl/BasicCancelOk.cs

@@ -15,20 +15,25 @@ public BasicConsumeOk(IBasicConsumer consumer, string consumerTag) : base(consum
             _consumerTag = consumerTag;
         }
 
-        protected override async Task Execute(ModelBase model, IAsyncBasicConsumer consumer)
+        protected override async Task Execute(IModel model, IAsyncBasicConsumer consumer)
         {
             try
             {
                 await consumer.HandleBasicConsumeOk(_consumerTag).ConfigureAwait(false);
             }
             catch (Exception e)
             {
+                if (!(model is ModelBase modelBase))
+                {
+                    return;
+                }
+
                 var details = new Dictionary<string, object>()
                 {
                     {"consumer", consumer},
                     {"context",  "HandleBasicConsumeOk"}
                 };
-                model.OnCallbackException(CallbackExceptionEventArgs.Build(e, details));
+                modelBase.OnCallbackException(CallbackExceptionEventArgs.Build(e, details));
             }
         }
     }

projects/RabbitMQ.Client/client/impl/BasicConsumeOk.cs

@@ -18,12 +18,12 @@ sealed class BasicDeliver : Work
         readonly IBasicProperties _basicProperties;
         readonly ReadOnlyMemory<byte> _body;
 
-        public BasicDeliver(IBasicConsumer consumer, 
-            string consumerTag, 
-            ulong deliveryTag, 
-            bool redelivered, 
-            string exchange, 
-            string routingKey, 
+        public BasicDeliver(IBasicConsumer consumer,
+            string consumerTag,
+            ulong deliveryTag,
+            bool redelivered,
+            string exchange,
+            string routingKey,
             IBasicProperties basicProperties,
             ReadOnlyMemory<byte> body) : base(consumer)
         {
@@ -36,7 +36,7 @@ public BasicDeliver(IBasicConsumer consumer,
             _body = body;
         }
 
-        protected override async Task Execute(ModelBase model, IAsyncBasicConsumer consumer)
+        protected override async Task Execute(IModel model, IAsyncBasicConsumer consumer)
         {
             try
             {
@@ -50,12 +50,17 @@ await consumer.HandleBasicDeliver(_consumerTag,
             }
             catch (Exception e)
             {
+                if (!(model is ModelBase modelBase))
+                {
+                    return;
+                }
+
                 var details = new Dictionary<string, object>()
                 {
                     {"consumer", consumer},
                     {"context",  "HandleBasicDeliver"}
                 };
-                model.OnCallbackException(CallbackExceptionEventArgs.Build(e, details));
+                modelBase.OnCallbackException(CallbackExceptionEventArgs.Build(e, details));
             }
             finally
             {

projects/RabbitMQ.Client/client/impl/BasicDeliver.cs

@@ -62,12 +62,6 @@ internal BasicPublishBatch (ModelBase model, int sizeHint)
             _commands = new List<OutgoingCommand>(sizeHint);
         }
 
-        public void Add(string exchange, string routingKey, bool mandatory, IBasicProperties basicProperties, byte[] body)
-        {
-            ReadOnlyMemory<byte> bodyAsMemory = body;
-            Add(exchange, routingKey, mandatory, basicProperties, bodyAsMemory);
-        }
-
         public void Add(string exchange, string routingKey, bool mandatory, IBasicProperties basicProperties, ReadOnlyMemory<byte> body)
         {
             var method = new BasicPublish

projects/RabbitMQ.Client/client/impl/BasicPublishBatch.cs

@@ -15,20 +15,25 @@ public ModelShutdown(IBasicConsumer consumer, ShutdownEventArgs reason) : base(c
             _reason = reason;
         }
 
-        protected override async Task Execute(ModelBase model, IAsyncBasicConsumer consumer)
+        protected override async Task Execute(IModel model, IAsyncBasicConsumer consumer)
         {
             try
             {
                 await consumer.HandleModelShutdown(model, _reason).ConfigureAwait(false);
             }
             catch (Exception e)
             {
+                if (!(model is ModelBase modelBase))
+                {
+                    return;
+                }
+
                 var details = new Dictionary<string, object>()
                 {
                     { "consumer", consumer },
                     { "context", "HandleModelShutdown" }
                 };
-                model.OnCallbackException(CallbackExceptionEventArgs.Build(e, details));
+                modelBase.OnCallbackException(CallbackExceptionEventArgs.Build(e, details));
             }
         }
     }