Merge branch '6.x' of https://github.com/mot256/rabbitmq-dotnet-client into 6.x

Author: mot256

CHANGELOG.md

@@ -1,3 +1,18 @@
+## Changes Between 6.2.4 and 6.3.0
+
+GitHub milestone: [`6.3.0`](https://github.com/rabbitmq/rabbitmq-dotnet-client/milestone/52?closed=1)
+
+## Changes Between 6.2.3 and 6.2.4
+
+GitHub milestone: [`6.2.4`](https://github.com/rabbitmq/rabbitmq-dotnet-client/milestone/55?closed=1)
+
+This release contains some important bug fixes:
+
+* [Fix connection leaks on auto recovery](https://github.com/rabbitmq/rabbitmq-dotnet-client/pull/1145)
+* [Fix buffer overflow when writing long strings](https://github.com/rabbitmq/rabbitmq-dotnet-client/pull/1162)
+* [Fix regression resulting in `ObjectDisposedException`](https://github.com/rabbitmq/rabbitmq-dotnet-client/pull/1154)
+* [Fix regression that could affect consuming after auto recovery](https://github.com/rabbitmq/rabbitmq-dotnet-client/pull/1148)
+
 ## Changes Between 6.2.1 and 6.2.3
 
 GitHub milestone: [`6.2.3`](https://github.com/rabbitmq/rabbitmq-dotnet-client/milestone/54?closed=1)

_site

@@ -31,6 +31,7 @@
 
 using System;
 using System.Collections.Generic;
+using System.Buffers;
 using System.Linq;
 using System.Net.Security;
 using System.Security.Authentication;
@@ -188,6 +189,16 @@ public sealed class ConnectionFactory : ConnectionFactoryBase, IAsyncConnectionF
 
         // just here to hold the value that was set through the setter
         private Uri _uri;
+        private ArrayPool<byte> _memoryPool = ArrayPool<byte>.Shared;
+
+        /// <summary>
+        /// The memory pool used for allocating buffers. Default is <see cref="MemoryPool{T}.Shared"/>.
+        /// </summary>
+        public ArrayPool<byte> MemoryPool
+        {
+            get { return _memoryPool; }
+            set { _memoryPool = value ?? ArrayPool<byte>.Shared; }
+        }
 
         /// <summary>
         /// Amount of time protocol handshake operations are allowed to take before
@@ -497,7 +508,8 @@ public IConnection CreateConnection(IEndpointResolver endpointResolver, string c
                 else
                 {
                     var protocol = new RabbitMQ.Client.Framing.Protocol();
-                    conn = protocol.CreateConnection(this, false, endpointResolver.SelectOne(CreateFrameHandler), clientProvidedName);
+                    conn = protocol.CreateConnection(this, false, endpointResolver.SelectOne(CreateFrameHandler),
+                           _memoryPool, clientProvidedName);
                 }
             }
             catch (Exception e)
@@ -510,7 +522,7 @@ public IConnection CreateConnection(IEndpointResolver endpointResolver, string c
 
         internal IFrameHandler CreateFrameHandler(AmqpTcpEndpoint endpoint)
         {
-            IFrameHandler fh = Protocols.DefaultProtocol.CreateFrameHandler(endpoint, SocketFactory,
+            IFrameHandler fh = Protocols.DefaultProtocol.CreateFrameHandler(endpoint, _memoryPool, SocketFactory,
                 RequestedConnectionTimeout, SocketReadTimeout, SocketWriteTimeout);
             return ConfigureFrameHandler(fh);
         }

appveyor.yml

@@ -30,12 +30,8 @@
 //---------------------------------------------------------------------------
 
 using System;
-using System.Collections.Generic;
-using System.IO;
-using System.Threading;
 
 using RabbitMQ.Client.Events;
-using RabbitMQ.Client.Exceptions;
 
 namespace RabbitMQ.Client
 {

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

@@ -47,10 +47,11 @@ public void HandleBasicDeliver(IBasicConsumer consumer,
             IBasicProperties basicProperties,
             ReadOnlySpan<byte> body)
         {
-            byte[] bodyBytes = ArrayPool<byte>.Shared.Rent(body.Length);
+            var pool = _model.Session.Connection.MemoryPool;
+            byte[] bodyBytes = pool.Rent(body.Length);
             Memory<byte> bodyCopy = new Memory<byte>(bodyBytes, 0, body.Length);
             body.CopyTo(bodyCopy.Span);
-            ScheduleUnlessShuttingDown(new BasicDeliver(consumer, consumerTag, deliveryTag, redelivered, exchange, routingKey, basicProperties, bodyCopy));
+            ScheduleUnlessShuttingDown(new BasicDeliver(consumer, consumerTag, deliveryTag, redelivered, exchange, routingKey, basicProperties, bodyCopy, pool));
         }
 
         public void HandleBasicCancelOk(IBasicConsumer consumer, string consumerTag)

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

@@ -668,15 +668,22 @@ public void Init(IEndpointResolver endpoints)
             Init(fh);
         }
 
+        internal IFrameHandler FrameHandler
+        {
+            get
+            {
+                return _delegate.FrameHandler;
+            }
+        }
+
         private void Init(IFrameHandler fh)
         {
             if (_disposed)
             {
                 throw new ObjectDisposedException(GetType().FullName);
             }
 
-            _delegate = new Connection(_factory, false,
-                fh, ClientProvidedName);
+            _delegate = new Connection(_factory, false, fh, _factory.MemoryPool, ClientProvidedName);
 
             _recoveryTask = Task.Run(MainRecoveryLoop);
 
@@ -1009,7 +1016,7 @@ private bool TryRecoverConnectionDelegate()
             try
             {
                 IFrameHandler fh = _endpoints.SelectOne(_factory.CreateFrameHandler);
-                _delegate = new Connection(_factory, false, fh, ClientProvidedName);
+                _delegate = new Connection(_factory, false, fh, _factory.MemoryPool, ClientProvidedName);
                 return true;
             }
             catch (Exception e)

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

@@ -360,14 +360,7 @@ public bool IsClosed
         {
             get
             {
-                if (_delegate == null)
-                {
-                    return false;
-                }
-                else
-                {
-                    return _delegate.IsClosed;
-                }
+                return !IsOpen;
             }
         }
 

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

@@ -14,6 +14,7 @@ internal sealed class BasicDeliver : Work
         private readonly string _routingKey;
         private readonly IBasicProperties _basicProperties;
         private readonly ReadOnlyMemory<byte> _body;
+        private readonly ArrayPool<byte> _bodyOwner;
 
         public override string Context => "HandleBasicDeliver";
 
@@ -24,7 +25,8 @@ public BasicDeliver(IBasicConsumer consumer,
             string exchange,
             string routingKey,
             IBasicProperties basicProperties,
-            ReadOnlyMemory<byte> body) : base(consumer)
+            ReadOnlyMemory<byte> body,
+            ArrayPool<byte> pool) : base(consumer)
         {
             _consumerTag = consumerTag;
             _deliveryTag = deliveryTag;
@@ -33,6 +35,7 @@ public BasicDeliver(IBasicConsumer consumer,
             _routingKey = routingKey;
             _basicProperties = basicProperties;
             _body = body;
+            _bodyOwner = pool;
         }
 
         protected override Task Execute(IAsyncBasicConsumer consumer)
@@ -50,7 +53,7 @@ public override void PostExecute()
         {
             if (MemoryMarshal.TryGetArray(_body, out ArraySegment<byte> segment))
             {
-                ArrayPool<byte>.Shared.Return(segment.Array);
+                _bodyOwner.Return(segment.Array);
             }
         }
     }

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

@@ -88,7 +88,7 @@ public IncomingCommand HandleFrame(in InboundFrame frame)
                 return IncomingCommand.Empty;
             }
 
-            var result = new IncomingCommand(_method, _header, _body, _bodyBytes);
+            var result = new IncomingCommand(_method, _header, _body, _bodyBytes, _protocol.MemoryPool);
             Reset();
             return result;
         }
@@ -123,7 +123,7 @@ private void ParseHeaderFrame(in InboundFrame frame)
             _remainingBodyBytes = (int) totalBodyBytes;
 
             // Is returned by IncomingCommand.Dispose in Session.HandleFrame
-            _bodyBytes = ArrayPool<byte>.Shared.Rent(_remainingBodyBytes);
+            _bodyBytes = _protocol.MemoryPool.Rent(_remainingBodyBytes);
             _body = new Memory<byte>(_bodyBytes, 0, _remainingBodyBytes);
             UpdateContentBodyState();
         }