Changed ReceiveMessageAsync to return new abstract ResponseMessage to support the possibility that the server might return multiple message types in the future.

Author: rstam

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

@@ -147,31 +147,31 @@ public void OpenAsync_should_not_complete_the_second_call_until_the_first_is_com
         }
 
         [Test]
-        public void ReceiveMessageAsync_should_throw_an_ArgumentNullException_when_the_serializer_is_null()
+        public void ReceiveMessageAsync_should_throw_an_ArgumentNullException_when_the_encoderSelector_is_null()
         {
-            IBsonSerializer<int> serializer = null;
-            Action act = () => _subject.ReceiveMessageAsync(10, serializer, _messageEncoderSettings, CancellationToken.None).Wait();
+            IMessageEncoderSelector encoderSelector = null;
+            Action act = () => _subject.ReceiveMessageAsync(10, encoderSelector, _messageEncoderSettings, CancellationToken.None).Wait();
 
             act.ShouldThrow<ArgumentNullException>();
         }
 
         [Test]
         public void ReceiveMessageAsync_should_throw_an_ObjectDisposedException_if_the_connection_is_disposed()
         {
-            var serializer = Substitute.For<IBsonSerializer<BsonDocument>>();
+            var encoderSelector = Substitute.For<IMessageEncoderSelector>();
             _subject.Dispose();
 
-            Action act = () => _subject.ReceiveMessageAsync(10, serializer, _messageEncoderSettings, CancellationToken.None).Wait();
+            Action act = () => _subject.ReceiveMessageAsync(10, encoderSelector, _messageEncoderSettings, CancellationToken.None).Wait();
 
             act.ShouldThrow<ObjectDisposedException>();
         }
 
         [Test]
         public void ReceiveMessageAsync_should_throw_an_InvalidOperationException_if_the_connection_is_not_open()
         {
-            var serializer = Substitute.For<IBsonSerializer<BsonDocument>>();
+            var encoderSelector = Substitute.For<IMessageEncoderSelector>();
 
-            Action act = () => _subject.ReceiveMessageAsync(10, serializer, _messageEncoderSettings, CancellationToken.None).Wait();
+            Action act = () => _subject.ReceiveMessageAsync(10, encoderSelector, _messageEncoderSettings, CancellationToken.None).Wait();
 
             act.ShouldThrow<InvalidOperationException>();
         }
@@ -187,17 +187,18 @@ public void ReceiveMessageAsync_should_complete_when_reply_is_already_on_the_str
                 _subject.OpenAsync(CancellationToken.None).Wait();
 
                 var messageToReceive = MessageHelper.BuildSuccessReply<BsonDocument>(new BsonDocument(), BsonDocumentSerializer.Instance, responseTo: 10);
-                MessageHelper.WriteRepliesToStream(stream, new[] { messageToReceive });
+                MessageHelper.WriteResponsesToStream(stream, new[] { messageToReceive });
 
-                var received = _subject.ReceiveMessageAsync(10, BsonDocumentSerializer.Instance, _messageEncoderSettings, CancellationToken.None).Result;
+                var encoderSelector = new ReplyMessageEncoderSelector<BsonDocument>(BsonDocumentSerializer.Instance);
+                var received = _subject.ReceiveMessageAsync(10, encoderSelector, _messageEncoderSettings, CancellationToken.None).Result;
 
                 var expected = MessageHelper.TranslateMessagesToBsonDocuments(new[] { messageToReceive });
                 var actual = MessageHelper.TranslateMessagesToBsonDocuments(new[] { received });
 
                 actual.Should().BeEquivalentTo(expected);
 
                 _listener.ReceivedWithAnyArgs().ConnectionBeforeReceivingMessage(default(ConnectionBeforeReceivingMessageEvent));
-                _listener.ReceivedWithAnyArgs().ConnectionAfterReceivingMessage<BsonDocument>(default(ConnectionAfterReceivingMessageEvent<BsonDocument>));
+                _listener.ReceivedWithAnyArgs().ConnectionAfterReceivingMessage(default(ConnectionAfterReceivingMessageEvent));
             }
         }
 
@@ -211,12 +212,13 @@ public void ReceiveMessageAsync_should_complete_when_reply_is_not_already_on_the
 
                 _subject.OpenAsync(CancellationToken.None).Wait();
 
-                var receivedTask = _subject.ReceiveMessageAsync(10, BsonDocumentSerializer.Instance, _messageEncoderSettings, CancellationToken.None);
+                var encoderSelector = new ReplyMessageEncoderSelector<BsonDocument>(BsonDocumentSerializer.Instance);
+                var receivedTask = _subject.ReceiveMessageAsync(10, encoderSelector, _messageEncoderSettings, CancellationToken.None);
 
                 receivedTask.IsCompleted.Should().BeFalse();
 
                 var messageToReceive = MessageHelper.BuildSuccessReply<BsonDocument>(new BsonDocument(), BsonDocumentSerializer.Instance, responseTo: 10);
-                MessageHelper.WriteRepliesToStream(stream, new[] { messageToReceive });
+                MessageHelper.WriteResponsesToStream(stream, new[] { messageToReceive });
 
                 var received = receivedTask.Result;
 
@@ -237,12 +239,13 @@ public void ReceiveMessageAsync_should_handle_out_of_order_replies()
 
                 _subject.OpenAsync(CancellationToken.None).Wait();
 
-                var receivedTask11 = _subject.ReceiveMessageAsync(11, BsonDocumentSerializer.Instance, _messageEncoderSettings, CancellationToken.None);
-                var receivedTask10 = _subject.ReceiveMessageAsync(10, BsonDocumentSerializer.Instance, _messageEncoderSettings, CancellationToken.None);
+                var encoderSelector = new ReplyMessageEncoderSelector<BsonDocument>(BsonDocumentSerializer.Instance);
+                var receivedTask11 = _subject.ReceiveMessageAsync(11, encoderSelector, _messageEncoderSettings, CancellationToken.None);
+                var receivedTask10 = _subject.ReceiveMessageAsync(10, encoderSelector, _messageEncoderSettings, CancellationToken.None);
 
                 var messageToReceive10 = MessageHelper.BuildSuccessReply<BsonDocument>(new BsonDocument(), BsonDocumentSerializer.Instance, responseTo: 10);
                 var messageToReceive11 = MessageHelper.BuildSuccessReply<BsonDocument>(new BsonDocument(), BsonDocumentSerializer.Instance, responseTo: 11);
-                MessageHelper.WriteRepliesToStream(stream, new[] { messageToReceive10, messageToReceive11 });
+                MessageHelper.WriteResponsesToStream(stream, new[] { messageToReceive10, messageToReceive11 });
 
                 var received11 = receivedTask11.Result;
                 var received10 = receivedTask10.Result;
@@ -272,8 +275,9 @@ public async Task ReceiveMessageAsync_should_throw_network_exception_to_all_awai
 
                 await _subject.OpenAsync(CancellationToken.None);
 
-                var task1 = _subject.ReceiveMessageAsync<BsonDocument>(1, BsonDocumentSerializer.Instance, _messageEncoderSettings, CancellationToken.None);
-                var task2 = _subject.ReceiveMessageAsync<BsonDocument>(2, BsonDocumentSerializer.Instance, _messageEncoderSettings, CancellationToken.None);
+                var encoderSelector = new ReplyMessageEncoderSelector<BsonDocument>(BsonDocumentSerializer.Instance);
+                var task1 = _subject.ReceiveMessageAsync(1, encoderSelector, _messageEncoderSettings, CancellationToken.None);
+                var task2 = _subject.ReceiveMessageAsync(2, encoderSelector, _messageEncoderSettings, CancellationToken.None);
 
                 readTcs.SetException(new SocketException());
 
@@ -310,7 +314,8 @@ public async Task ReceiveMessageAsync_should_throw_MongoConnectionClosedExceptio
 
                 await _subject.OpenAsync(CancellationToken.None);
 
-                var task1 = _subject.ReceiveMessageAsync<BsonDocument>(1, BsonDocumentSerializer.Instance, _messageEncoderSettings, CancellationToken.None);
+                var encoderSelector = new ReplyMessageEncoderSelector<BsonDocument>(BsonDocumentSerializer.Instance);
+                var task1 = _subject.ReceiveMessageAsync(1, encoderSelector, _messageEncoderSettings, CancellationToken.None);
 
                 readTcs.SetException(new SocketException());
 
@@ -319,7 +324,7 @@ public async Task ReceiveMessageAsync_should_throw_MongoConnectionClosedExceptio
                     .WithInnerException<SocketException>()
                     .And.ConnectionId.Should().Be(_subject.ConnectionId);
 
-                var task2 = _subject.ReceiveMessageAsync<BsonDocument>(2, BsonDocumentSerializer.Instance, _messageEncoderSettings, CancellationToken.None);
+                var task2 = _subject.ReceiveMessageAsync(2, encoderSelector, _messageEncoderSettings, CancellationToken.None);
 
                 Func<Task> act2 = () => task2;
                 act2.ShouldThrow<MongoConnectionClosedException>()

src/MongoDB.Driver.Core.Tests/Core/Helpers/MessageHelper.cs

@@ -118,14 +118,14 @@ public static List<BsonDocument> TranslateMessagesToBsonDocuments(byte[] bytes)
             return TranslateMessagesToBsonDocuments(TranslateBytesToRequests(bytes));
         }
 
-        public static void WriteRepliesToStream(Stream stream, IEnumerable<ReplyMessage> replies)
+        public static void WriteResponsesToStream(Stream stream, IEnumerable<ResponseMessage> responses)
         {
             var startPosition = stream.Position;
-            foreach (var reply in replies)
+            foreach (var response in responses)
             {
                 var encoderFactory = new BinaryMessageEncoderFactory(stream, null);
-                var encoder = reply.GetEncoder(encoderFactory);
-                encoder.WriteMessage(reply);
+                var encoder = response.GetEncoder(encoderFactory);
+                encoder.WriteMessage(response);
             }
             stream.Position = startPosition;
         }

src/MongoDB.Driver.Core.Tests/Core/Helpers/MockConnection.cs

@@ -91,9 +91,9 @@ public Task OpenAsync(CancellationToken cancellationToken)
             return Task.FromResult<object>(null);
         }
 
-        public Task<ReplyMessage<TDocument>> ReceiveMessageAsync<TDocument>(int responseTo, Bson.Serialization.IBsonSerializer<TDocument> serializer, MessageEncoderSettings messageEncoderSettings, CancellationToken cancellationToken)
+        public Task<ResponseMessage> ReceiveMessageAsync(int responseTo, IMessageEncoderSelector encoderSelector, MessageEncoderSettings messageEncoderSettings, CancellationToken cancellationToken)
         {
-            return Task.FromResult<ReplyMessage<TDocument>>((ReplyMessage<TDocument>)_replyMessages.Dequeue());
+            return Task.FromResult((ResponseMessage)_replyMessages.Dequeue());
         }
 
         public Task SendMessagesAsync(IEnumerable<RequestMessage> messages, MessageEncoderSettings messageEncoderSettings, CancellationToken cancellationToken)

src/MongoDB.Driver.Core/Core/ConnectionPools/ExclusiveConnectionPool.cs

@@ -475,9 +475,9 @@ public Task OpenAsync(CancellationToken cancellationToken)
                 return _connection.OpenAsync(cancellationToken);
             }
 
-            public Task<ReplyMessage<TDocument>> ReceiveMessageAsync<TDocument>(int responseTo, IBsonSerializer<TDocument> serializer, MessageEncoderSettings messageEncoderSettings, CancellationToken cancellationToken)
+            public Task<ResponseMessage> ReceiveMessageAsync(int responseTo, IMessageEncoderSelector encoderSelector, MessageEncoderSettings messageEncoderSettings, CancellationToken cancellationToken)
             {
-                return _connection.ReceiveMessageAsync<TDocument>(responseTo, serializer, messageEncoderSettings, cancellationToken);
+                return _connection.ReceiveMessageAsync(responseTo, encoderSelector, messageEncoderSettings, cancellationToken);
             }
 
             public Task SendMessagesAsync(IEnumerable<RequestMessage> messages, MessageEncoderSettings messageEncoderSettings, CancellationToken cancellationToken)
@@ -548,10 +548,10 @@ public Task OpenAsync(CancellationToken cancellationToken)
                 return _reference.Instance.OpenAsync(cancellationToken);
             }
 
-            public Task<ReplyMessage<TDocument>> ReceiveMessageAsync<TDocument>(int responseTo, IBsonSerializer<TDocument> serializer, MessageEncoderSettings messageEncoderSettings, CancellationToken cancellationToken)
+            public Task<ResponseMessage> ReceiveMessageAsync(int responseTo, IMessageEncoderSelector encoderSelector, MessageEncoderSettings messageEncoderSettings, CancellationToken cancellationToken)
             {
                 ThrowIfDisposed();
-                return _reference.Instance.ReceiveMessageAsync<TDocument>(responseTo, serializer, messageEncoderSettings, cancellationToken);
+                return _reference.Instance.ReceiveMessageAsync(responseTo, encoderSelector, messageEncoderSettings, cancellationToken);
             }
 
             public Task SendMessagesAsync(IEnumerable<RequestMessage> messages, MessageEncoderSettings messageEncoderSettings, CancellationToken cancellationToken)

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

@@ -289,13 +289,13 @@ private async Task<IByteBuffer> ReceiveBufferAsync(int responseTo, CancellationT
             }
         }
 
-        public async Task<ReplyMessage<TDocument>> ReceiveMessageAsync<TDocument>(
+        public async Task<ResponseMessage> ReceiveMessageAsync(
             int responseTo,
-            IBsonSerializer<TDocument> serializer,
+            IMessageEncoderSelector encoderSelector,
             MessageEncoderSettings messageEncoderSettings,
             CancellationToken cancellationToken)
         {
-            Ensure.IsNotNull(serializer, "serializer");
+            Ensure.IsNotNull(encoderSelector, "encoderSelector");
             ThrowIfDisposedOrNotOpen();
 
             try
@@ -306,7 +306,7 @@ public async Task<ReplyMessage<TDocument>> ReceiveMessageAsync<TDocument>(
                 }
 
                 var stopwatch = Stopwatch.StartNew();
-                ReplyMessage<TDocument> reply;
+                ResponseMessage reply;
                 int length;
                 using (var buffer = await ReceiveBufferAsync(responseTo, cancellationToken).ConfigureAwait(false))
                 {
@@ -315,15 +315,15 @@ public async Task<ReplyMessage<TDocument>> ReceiveMessageAsync<TDocument>(
                     using (var stream = new ByteBufferStream(buffer))
                     {
                         var encoderFactory = new BinaryMessageEncoderFactory(stream, messageEncoderSettings);
-                        var encoder = encoderFactory.GetReplyMessageEncoder<TDocument>(serializer);
-                        reply = (ReplyMessage<TDocument>)encoder.ReadMessage();
+                        var encoder = encoderSelector.GetEncoder(encoderFactory);
+                        reply = (ResponseMessage)encoder.ReadMessage();
                     }
                 }
                 stopwatch.Stop();
 
                 if (_listener != null)
                 {
-                    _listener.ConnectionAfterReceivingMessage<TDocument>(new ConnectionAfterReceivingMessageEvent<TDocument>(_connectionId, reply, length, stopwatch.Elapsed));
+                    _listener.ConnectionAfterReceivingMessage(new ConnectionAfterReceivingMessageEvent(_connectionId, reply, length, stopwatch.Elapsed));
                 }
 
                 return reply;

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

@@ -84,13 +84,14 @@ public interface IConnection : IDisposable
         /// <summary>
         /// Receives a message.
         /// </summary>
-        /// <typeparam name="TDocument">The type of the document.</typeparam>
         /// <param name="responseTo">The id of the sent message for which a response is to be received.</param>
-        /// <param name="serializer">The serializer.</param>
+        /// <param name="encoderSelector">The encoder selector.</param>
         /// <param name="messageEncoderSettings">The message encoder settings.</param>
         /// <param name="cancellationToken">The cancellation token.</param>
-        /// <returns>A Task whose result is the reply message.</returns>
-        Task<ReplyMessage<TDocument>> ReceiveMessageAsync<TDocument>(int responseTo, IBsonSerializer<TDocument> serializer, MessageEncoderSettings messageEncoderSettings, CancellationToken cancellationToken);
+        /// <returns>
+        /// A Task whose result is the response message.
+        /// </returns>
+        Task<ResponseMessage> ReceiveMessageAsync(int responseTo, IMessageEncoderSelector encoderSelector, MessageEncoderSettings messageEncoderSettings, CancellationToken cancellationToken);
 
         /// <summary>
         /// Sends the messages.

src/MongoDB.Driver.Core/Core/Events/ConnectionAfterReceivingMessageEvent.cs

@@ -24,25 +24,24 @@ namespace MongoDB.Driver.Core.Events
     /// <summary>
     /// Represents information about a ConnectionAfterReceivingMessage event.
     /// </summary>
-    /// <typeparam name="TDocument">The type of the documents.</typeparam>
-    public struct ConnectionAfterReceivingMessageEvent<TDocument>
+    public struct ConnectionAfterReceivingMessageEvent
     {
         private readonly ConnectionId _connectionId;
         private readonly TimeSpan _elapsed;
         private readonly int _length;
-        private readonly ReplyMessage<TDocument> _replyMessage;
+        private readonly ResponseMessage _receivedMessage;
 
         /// <summary>
-        /// Initializes a new instance of the <see cref="ConnectionAfterReceivingMessageEvent{TDocument}"/> struct.
+        /// Initializes a new instance of the <see cref="ConnectionAfterReceivingMessageEvent"/> struct.
         /// </summary>
         /// <param name="connectionId">The connection identifier.</param>
-        /// <param name="replyMessage">The reply message.</param>
+        /// <param name="receivedMessage">The received message.</param>
         /// <param name="length">The length.</param>
         /// <param name="elapsed">The elapsed time.</param>
-        public ConnectionAfterReceivingMessageEvent(ConnectionId connectionId, ReplyMessage<TDocument> replyMessage, int length, TimeSpan elapsed)
+        public ConnectionAfterReceivingMessageEvent(ConnectionId connectionId, ResponseMessage receivedMessage, int length, TimeSpan elapsed)
         {
             _connectionId = connectionId;
-            _replyMessage = replyMessage;
+            _receivedMessage = receivedMessage;
             _length = length;
             _elapsed = elapsed;
         }
@@ -81,14 +80,14 @@ public int Length
         }
 
         /// <summary>
-        /// Gets the reply message.
+        /// Gets the received message.
         /// </summary>
         /// <value>
-        /// The reply message.
+        /// The received message.
         /// </value>
-        public ReplyMessage<TDocument> ReplyMessage
+        public ResponseMessage ReceivedMessage
         {
-            get { return _replyMessage; }
+            get { return _receivedMessage; }
         }
     }
 }

src/MongoDB.Driver.Core/Core/Events/ConnectionListenerPair.cs

@@ -117,10 +117,10 @@ public void ConnectionBeforeReceivingMessage(ConnectionBeforeReceivingMessageEve
         }
 
         /// <inheritdoc/>
-        public void ConnectionAfterReceivingMessage<TDocument>(ConnectionAfterReceivingMessageEvent<TDocument> @event)
+        public void ConnectionAfterReceivingMessage(ConnectionAfterReceivingMessageEvent @event)
         {
-            _first.ConnectionAfterReceivingMessage<TDocument>(@event);
-            _second.ConnectionAfterReceivingMessage<TDocument>(@event);
+            _first.ConnectionAfterReceivingMessage(@event);
+            _second.ConnectionAfterReceivingMessage(@event);
         }
 
         /// <inheritdoc/>

src/MongoDB.Driver.Core/Core/Events/Diagnostics/LogListener.cs

@@ -295,9 +295,9 @@ public void ConnectionBeforeReceivingMessage(ConnectionBeforeReceivingMessageEve
         }
 
         /// <inheritdoc/>
-        public void ConnectionAfterReceivingMessage<TDocument>(ConnectionAfterReceivingMessageEvent<TDocument> @event)
+        public void ConnectionAfterReceivingMessage(ConnectionAfterReceivingMessageEvent @event)
         {
-            Log(LogLevel.Info, "{0}: received message in response to {1} of length {2} bytes in {3}ms.", Label(@event.ConnectionId), @event.ReplyMessage.ResponseTo.ToString(), @event.Length.ToString(), @event.Elapsed.TotalMilliseconds.ToString());
+            Log(LogLevel.Info, "{0}: received message in response to {1} of length {2} bytes in {3}ms.", Label(@event.ConnectionId), @event.ReceivedMessage.ResponseTo.ToString(), @event.Length.ToString(), @event.Elapsed.TotalMilliseconds.ToString());
         }
 
         /// <inheritdoc/>