CSHARP-3984: Remove BinaryConnection.DropBox

Author: sanych-sun

src/MongoDB.Driver/Core/Connections/BinaryConnection.cs

@@ -15,7 +15,6 @@
 
 using System;
 using System.Buffers.Binary;
-using System.Collections.Concurrent;
 using System.Diagnostics;
 using System.IO;
 using System.Net;
@@ -46,7 +45,6 @@ internal sealed class BinaryConnection : IConnection
         private ConnectionInitializerContext _connectionInitializerContext;
         private EndPoint _endPoint;
         private ConnectionDescription _description;
-        private readonly Dropbox _dropbox = new Dropbox();
         private bool _failedEventHasBeenRaised;
         private DateTime _lastUsedAtUtc;
         private DateTime _openedAtUtc;
@@ -356,50 +354,6 @@ private IByteBuffer ReceiveBuffer(OperationContext operationContext)
             }
         }
 
-        private IByteBuffer ReceiveBuffer(OperationContext operationContext, int responseTo)
-        {
-            using (var receiveLockRequest = new SemaphoreSlimRequest(_receiveLock, operationContext.RemainingTimeout, operationContext.CancellationToken))
-            {
-                var messageTask = _dropbox.GetMessageAsync(responseTo);
-                try
-                {
-                    Task.WaitAny(messageTask, receiveLockRequest.Task);
-                    if (messageTask.IsCompleted)
-                    {
-                        return _dropbox.RemoveMessage(responseTo); // also propagates exception if any
-                    }
-
-                    receiveLockRequest.Task.GetAwaiter().GetResult(); // propagate exceptions
-                    while (true)
-                    {
-                        try
-                        {
-                            var buffer = ReceiveBuffer(operationContext);
-                            _dropbox.AddMessage(buffer);
-                        }
-                        catch (Exception ex)
-                        {
-                            _dropbox.AddException(ex);
-                        }
-
-                        if (messageTask.IsCompleted)
-                        {
-                            return _dropbox.RemoveMessage(responseTo); // also propagates exception if any
-                        }
-
-                        operationContext.ThrowIfTimedOutOrCanceled();
-                    }
-                }
-                catch
-                {
-                    var ignored = messageTask.ContinueWith(
-                        t => { _dropbox.RemoveMessage(responseTo).Dispose(); },
-                        TaskContinuationOptions.OnlyOnRanToCompletion);
-                    throw;
-                }
-            }
-        }
-
         private async Task<IByteBuffer> ReceiveBufferAsync(OperationContext operationContext)
         {
             try
@@ -425,50 +379,6 @@ private async Task<IByteBuffer> ReceiveBufferAsync(OperationContext operationCon
             }
         }
 
-        private async Task<IByteBuffer> ReceiveBufferAsync(OperationContext operationContext, int responseTo)
-        {
-            using (var receiveLockRequest = new SemaphoreSlimRequest(_receiveLock, operationContext.RemainingTimeout, operationContext.CancellationToken))
-            {
-                var messageTask = _dropbox.GetMessageAsync(responseTo);
-                try
-                {
-                    await Task.WhenAny(messageTask, receiveLockRequest.Task).ConfigureAwait(false);
-                    if (messageTask.IsCompleted)
-                    {
-                        return _dropbox.RemoveMessage(responseTo); // also propagates exception if any
-                    }
-
-                    await receiveLockRequest.Task.ConfigureAwait(false); // propagate exceptions
-                    while (true)
-                    {
-                        try
-                        {
-                            var buffer = await ReceiveBufferAsync(operationContext).ConfigureAwait(false);
-                            _dropbox.AddMessage(buffer);
-                        }
-                        catch (Exception ex)
-                        {
-                            _dropbox.AddException(ex);
-                        }
-
-                        if (messageTask.IsCompleted)
-                        {
-                            return _dropbox.RemoveMessage(responseTo); // also propagates exception if any
-                        }
-
-                        operationContext.ThrowIfTimedOutOrCanceled();
-                    }
-                }
-                catch
-                {
-                    var ignored = messageTask.ContinueWith(
-                        t => { _dropbox.RemoveMessage(responseTo).Dispose(); },
-                        TaskContinuationOptions.OnlyOnRanToCompletion);
-                    throw;
-                }
-            }
-        }
-
         public ResponseMessage ReceiveMessage(
             OperationContext operationContext,
             int responseTo,
@@ -482,8 +392,9 @@ public ResponseMessage ReceiveMessage(
             try
             {
                 helper.ReceivingMessage();
-                using (var buffer = ReceiveBuffer(operationContext, responseTo))
+                using (var buffer = ReceiveBuffer(operationContext))
                 {
+                    EnsureResponseToIsCorrect(buffer, responseTo);
                     var message = helper.DecodeMessage(operationContext, buffer, encoderSelector);
                     helper.ReceivedMessage(buffer, message);
                     return message;
@@ -497,7 +408,9 @@ public ResponseMessage ReceiveMessage(
             }
         }
 
-        public async Task<ResponseMessage> ReceiveMessageAsync(OperationContext operationContext, int responseTo,
+        public async Task<ResponseMessage> ReceiveMessageAsync(
+            OperationContext operationContext,
+            int responseTo,
             IMessageEncoderSelector encoderSelector,
             MessageEncoderSettings messageEncoderSettings)
         {
@@ -508,8 +421,9 @@ public async Task<ResponseMessage> ReceiveMessageAsync(OperationContext operatio
             try
             {
                 helper.ReceivingMessage();
-                using (var buffer = await ReceiveBufferAsync(operationContext, responseTo).ConfigureAwait(false))
+                using (var buffer = await ReceiveBufferAsync(operationContext).ConfigureAwait(false))
                 {
+                    EnsureResponseToIsCorrect(buffer, responseTo);
                     var message = helper.DecodeMessage(operationContext, buffer, encoderSelector);
                     helper.ReceivedMessage(buffer, message);
                     return message;
@@ -523,6 +437,23 @@ public async Task<ResponseMessage> ReceiveMessageAsync(OperationContext operatio
             }
         }
 
+        private IByteBuffer EnsureResponseToIsCorrect(IByteBuffer message, int responseTo)
+        {
+            var receivedResponseTo = GetResponseTo(message);
+            if (receivedResponseTo != responseTo)
+            {
+                throw new InvalidOperationException($"Expected responseTo to be {responseTo} but was {receivedResponseTo}."); // should not be reached
+            }
+
+            return message;
+
+            int GetResponseTo(IByteBuffer message)
+            {
+                var backingBytes = message.AccessBackingBytes(8);
+                return BitConverter.ToInt32(backingBytes.Array, backingBytes.Offset);
+            }
+        }
+
         private void SendBuffer(OperationContext operationContext, IByteBuffer buffer)
         {
             _sendLock.Wait(operationContext.RemainingTimeout, operationContext.CancellationToken);
@@ -770,47 +701,6 @@ private void ThrowOperationCanceledExceptionIfRequired(Exception exception)
         }
 
         // nested classes
-        private class Dropbox
-        {
-            private readonly ConcurrentDictionary<int, TaskCompletionSource<IByteBuffer>> _messages = new ConcurrentDictionary<int, TaskCompletionSource<IByteBuffer>>();
-
-            // public methods
-            public void AddException(Exception exception)
-            {
-                foreach (var taskCompletionSource in _messages.Values)
-                {
-                    taskCompletionSource.TrySetException(exception); // has no effect on already completed tasks
-                }
-            }
-
-            public void AddMessage(IByteBuffer message)
-            {
-                var responseTo = GetResponseTo(message);
-                var tcs = _messages.GetOrAdd(responseTo, x => new TaskCompletionSource<IByteBuffer>());
-                tcs.TrySetResult(message);
-            }
-
-            public Task<IByteBuffer> GetMessageAsync(int responseTo)
-            {
-                var tcs = _messages.GetOrAdd(responseTo, _ => new TaskCompletionSource<IByteBuffer>());
-                return tcs.Task;
-            }
-
-            public IByteBuffer RemoveMessage(int responseTo)
-            {
-                TaskCompletionSource<IByteBuffer> tcs;
-                _messages.TryRemove(responseTo, out tcs);
-                return tcs.Task.GetAwaiter().GetResult(); // RemoveMessage is only called when Task is complete
-            }
-
-            // private methods
-            private int GetResponseTo(IByteBuffer message)
-            {
-                var backingBytes = message.AccessBackingBytes(8);
-                return BinaryPrimitives.ReadInt32LittleEndian(new ReadOnlySpan<byte>(backingBytes.Array, backingBytes.Offset, 4));
-            }
-        }
-
         private class OpenConnectionHelper
         {
             private readonly BinaryConnection _connection;

src/MongoDB.Driver/Core/Misc/SemaphoreSlimRequest.cs

@@ -22,6 +22,7 @@ namespace MongoDB.Driver.Core.Misc
     /// <summary>
     /// Represents a tentative request to acquire a SemaphoreSlim.
     /// </summary>
+    [Obsolete("SemaphoreSlimRequest is deprecated and will be removed in future release")]
     public sealed class SemaphoreSlimRequest : IDisposable
     {
         // private fields

tests/MongoDB.Driver.Tests/Core/Connections/BinaryConnectionTests.cs

@@ -489,52 +489,6 @@ public async Task ReceiveMessage_should_complete_when_reply_is_not_already_on_th
             }
         }
 
-        [Theory]
-        [ParameterAttributeData]
-        public async Task ReceiveMessage_should_handle_out_of_order_replies(
-            [Values(false, true)]
-            bool async1,
-            [Values(false, true)]
-            bool async2)
-        {
-            using (var stream = new BlockingMemoryStream())
-            {
-                _mockStreamFactory.Setup(f => f.CreateStreamAsync(_endPoint, It.IsAny<CancellationToken>())).ReturnsAsync(stream);
-                await _subject.OpenAsync(OperationContext.NoTimeout);
-                _capturedEvents.Clear();
-
-                var encoderSelector = new ReplyMessageEncoderSelector<BsonDocument>(BsonDocumentSerializer.Instance);
-
-                var receivedTask10 = async1 ?
-                    _subject.ReceiveMessageAsync(OperationContext.NoTimeout, 10, encoderSelector, _messageEncoderSettings) :
-                    Task.Run(() => _subject.ReceiveMessage(OperationContext.NoTimeout, 10, encoderSelector, _messageEncoderSettings));
-
-                var receivedTask11 = async2 ?
-                    _subject.ReceiveMessageAsync(OperationContext.NoTimeout, 11, encoderSelector, _messageEncoderSettings) :
-                    Task.Run(() => _subject.ReceiveMessage(OperationContext.NoTimeout, 11, encoderSelector, _messageEncoderSettings));
-
-                SpinWait.SpinUntil(() => _capturedEvents.Count >= 2, TimeSpan.FromSeconds(5)).Should().BeTrue();
-
-                var messageToReceive10 = MessageHelper.BuildReply<BsonDocument>(new BsonDocument("_id", 10), BsonDocumentSerializer.Instance, responseTo: 10);
-                var messageToReceive11 = MessageHelper.BuildReply<BsonDocument>(new BsonDocument("_id", 11), BsonDocumentSerializer.Instance, responseTo: 11);
-                MessageHelper.WriteResponsesToStream(stream, messageToReceive11, messageToReceive10); // out of order
-
-                var received10 = await receivedTask10;
-                var received11 = await receivedTask11;
-
-                var expected = MessageHelper.TranslateMessagesToBsonDocuments(new[] { messageToReceive10, messageToReceive11 });
-                var actual = MessageHelper.TranslateMessagesToBsonDocuments(new[] { received10, received11 });
-
-                actual.Should().BeEquivalentTo(expected);
-
-                _capturedEvents.Next().Should().BeOfType<ConnectionReceivingMessageEvent>();
-                _capturedEvents.Next().Should().BeOfType<ConnectionReceivingMessageEvent>();
-                _capturedEvents.Next().Should().BeOfType<ConnectionReceivedMessageEvent>();
-                _capturedEvents.Next().Should().BeOfType<ConnectionReceivedMessageEvent>();
-                _capturedEvents.Any().Should().BeFalse();
-            }
-        }
-
         [Fact]
         public async Task ReceiveMessage_should_not_produce_unobserved_task_exceptions_on_fail()
         {