CSHARP-2020: OP_MSG support.

Author: rstam

src/MongoDB.Driver.Core/Core/Bindings/IChannel.cs

@@ -53,6 +53,7 @@ public interface IChannel : IDisposable
         /// <param name="messageEncoderSettings">The message encoder settings.</param>
         /// <param name="cancellationToken">The cancellation token.</param>
         /// <returns>The result of the Command protocol.</returns>
+        [Obsolete("Use the newest overload instead.")]
         TResult Command<TResult>(
             DatabaseNamespace databaseNamespace,
             BsonDocument command,
@@ -81,6 +82,7 @@ TResult Command<TResult>(
         /// <returns>
         /// The result of the Command protocol.
         /// </returns>
+        [Obsolete("Use the newest overload instead.")]
         TResult Command<TResult>(
             ICoreSession session,
             ReadPreference readPreference,
@@ -94,6 +96,35 @@ TResult Command<TResult>(
             MessageEncoderSettings messageEncoderSettings,
             CancellationToken cancellationToken);
 
+        /// <summary>
+        /// Executes a Command protocol.
+        /// </summary>
+        /// <typeparam name="TResult">The type of the result.</typeparam>
+        /// <param name="session">The session.</param>
+        /// <param name="readPreference">The read preference.</param>
+        /// <param name="databaseNamespace">The database namespace.</param>
+        /// <param name="command">The command.</param>
+        /// <param name="commandValidator">The command validator.</param>
+        /// <param name="additionalOptions">The additional options.</param>
+        /// <param name="responseHandling">The response handling.</param>
+        /// <param name="resultSerializer">The result serializer.</param>
+        /// <param name="messageEncoderSettings">The message encoder settings.</param>
+        /// <param name="cancellationToken">The cancellation token.</param>
+        /// <returns>
+        /// The result of the Command protocol.
+        /// </returns>
+        TResult Command<TResult>(
+            ICoreSession session,
+            ReadPreference readPreference,
+            DatabaseNamespace databaseNamespace,
+            BsonDocument command,
+            IElementNameValidator commandValidator,
+            BsonDocument additionalOptions,
+            Func<CommandResponseHandling> responseHandling,
+            IBsonSerializer<TResult> resultSerializer,
+            MessageEncoderSettings messageEncoderSettings,
+            CancellationToken cancellationToken);
+
         /// <summary>
         /// Executes a Command protocol.
         /// </summary>
@@ -107,6 +138,7 @@ TResult Command<TResult>(
         /// <param name="messageEncoderSettings">The message encoder settings.</param>
         /// <param name="cancellationToken">The cancellation token.</param>
         /// <returns>A Task whose result is the result of the Command protocol.</returns>
+        [Obsolete("Use the newest overload instead.")]
         Task<TResult> CommandAsync<TResult>(
             DatabaseNamespace databaseNamespace,
             BsonDocument command,
@@ -135,6 +167,7 @@ Task<TResult> CommandAsync<TResult>(
         /// <returns>
         /// A Task whose result is the result of the Command protocol.
         /// </returns>
+        [Obsolete("Use the newest overload instead.")]
         Task<TResult> CommandAsync<TResult>(
             ICoreSession session,
             ReadPreference readPreference,
@@ -148,6 +181,35 @@ Task<TResult> CommandAsync<TResult>(
             MessageEncoderSettings messageEncoderSettings,
             CancellationToken cancellationToken);
 
+        /// <summary>
+        /// Executes a Command protocol.
+        /// </summary>
+        /// <typeparam name="TResult">The type of the result.</typeparam>
+        /// <param name="session">The session.</param>
+        /// <param name="readPreference">The read preference.</param>
+        /// <param name="databaseNamespace">The database namespace.</param>
+        /// <param name="command">The command.</param>
+        /// <param name="commandValidator">The command validator.</param>
+        /// <param name="additionalOptions">The additional options.</param>
+        /// <param name="responseHandling">The response handling.</param>
+        /// <param name="resultSerializer">The result serializer.</param>
+        /// <param name="messageEncoderSettings">The message encoder settings.</param>
+        /// <param name="cancellationToken">The cancellation token.</param>
+        /// <returns>
+        /// A Task whose result is the result of the Command protocol.
+        /// </returns>
+        Task<TResult> CommandAsync<TResult>(
+            ICoreSession session,
+            ReadPreference readPreference,
+            DatabaseNamespace databaseNamespace,
+            BsonDocument command,
+            IElementNameValidator commandValidator,
+            BsonDocument additionalOptions,
+            Func<CommandResponseHandling> responseHandling,
+            IBsonSerializer<TResult> resultSerializer,
+            MessageEncoderSettings messageEncoderSettings,
+            CancellationToken cancellationToken);
+
         /// <summary>
         /// Executes a Delete protocol.
         /// </summary>

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

@@ -168,7 +168,14 @@ public void AfterReceiving(ResponseMessage message, IByteBuffer buffer, Connecti
                 return;
             }
 
-            ProcessReplyMessage(state, message, buffer, connectionId, encoderSettings);
+            if (message is CommandResponseMessage)
+            {
+                ProcessCommandResponseMessage(state, (CommandResponseMessage)message, buffer, connectionId, encoderSettings);
+            }
+            else
+            {
+                ProcessReplyMessage(state, message, buffer, connectionId, encoderSettings);
+            }
         }
 
         public void ErrorReceiving(int responseTo, ConnectionId connectionId, Exception exception)
@@ -225,6 +232,9 @@ private void ProcessRequestMessages(Queue<RequestMessage> messageQueue, Connecti
             var message = messageQueue.Dequeue();
             switch (message.MessageType)
             {
+                case MongoDBMessageType.Command:
+                    ProcessCommandRequestMessage((CommandRequestMessage)message, messageQueue, connectionId, new CommandMessageBinaryEncoder(stream, encoderSettings), stopwatch);
+                    break;
                 case MongoDBMessageType.Delete:
                     ProcessDeleteMessage((DeleteMessage)message, messageQueue, connectionId, new DeleteMessageBinaryEncoder(stream, encoderSettings), stopwatch);
                     break;
@@ -248,6 +258,103 @@ private void ProcessRequestMessages(Queue<RequestMessage> messageQueue, Connecti
             }
         }
 
+        private void ProcessCommandRequestMessage(CommandRequestMessage originalMessage, Queue<RequestMessage> messageQueue, ConnectionId connectionId, CommandMessageBinaryEncoder encoder, Stopwatch stopwatch)
+        {
+            var requestId = originalMessage.RequestId;
+            var operationId = EventContext.OperationId;
+
+            var decodedMessage = encoder.ReadMessage();
+            using (new CommandMessageDisposer(decodedMessage))
+            {
+                var type0Section = decodedMessage.Sections.OfType<Type0CommandMessageSection>().Single();
+                var command = (BsonDocument)type0Section.Document;
+                var commandName = command.GetElement(0).Name;
+                var databaseName = command["$db"].AsString;
+                var databaseNamespace = new DatabaseNamespace(databaseName);
+                if (__securitySensitiveCommands.Contains(commandName))
+                {
+                    command = new BsonDocument();
+                }
+
+                if (_startedEvent != null)
+                {
+
+                    var @event = new CommandStartedEvent(
+                        commandName,
+                        command,
+                        databaseNamespace,
+                        operationId,
+                        requestId,
+                        connectionId);
+
+                    _startedEvent(@event);
+                }
+
+                if (_shouldTrackState)
+                {
+                    _state.TryAdd(requestId, new CommandState
+                    {
+                        CommandName = commandName,
+                        OperationId = operationId,
+                        Stopwatch = stopwatch,
+                        QueryNamespace = new CollectionNamespace(databaseNamespace, "$cmd"),
+                        ExpectedResponseType = ExpectedResponseType.Command
+                    });
+                }
+            }
+        }
+
+        private void ProcessCommandResponseMessage(CommandState state, CommandResponseMessage message, IByteBuffer buffer, ConnectionId connectionId, MessageEncoderSettings encoderSettings)
+        {
+            var wrappedMessage = message.WrappedMessage;
+            var type0Section = wrappedMessage.Sections.OfType<Type0CommandMessageSection<RawBsonDocument>>().Single();
+            var reply = (BsonDocument)type0Section.Document;
+
+            BsonValue ok;
+            if (!reply.TryGetValue("ok", out ok))
+            {
+                // this is a degenerate case with the server and 
+                // we don't really know what to do here...
+                return;
+            }
+
+            if (__securitySensitiveCommands.Contains(state.CommandName))
+            {
+                reply = new BsonDocument();
+            }
+
+            if (ok.ToBoolean())
+            {
+                if (_succeededEvent != null)
+                {
+                    _succeededEvent(new CommandSucceededEvent(
+                        state.CommandName,
+                        reply,
+                        state.OperationId,
+                        message.ResponseTo,
+                        connectionId,
+                        state.Stopwatch.Elapsed));
+                }
+            }
+            else
+            {
+                if (_failedEvent != null)
+                {
+                    _failedEvent(new CommandFailedEvent(
+                        state.CommandName,
+                        new MongoCommandException(
+                            connectionId,
+                            string.Format("{0} command failed", state.CommandName),
+                            null,
+                            reply),
+                        state.OperationId,
+                        message.ResponseTo,
+                        connectionId,
+                        state.Stopwatch.Elapsed));
+                }
+            }
+        }
+
         private void ProcessDeleteMessage(DeleteMessage originalMessage, Queue<RequestMessage> messageQueue, ConnectionId connectionId, DeleteMessageBinaryEncoder encoder, Stopwatch stopwatch)
         {
             var commandName = "delete";

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

@@ -23,7 +23,7 @@ namespace MongoDB.Driver.Core.Misc
     /// Represents a batch of items that can be split if not all items can be processed at once.
     /// </summary>
     /// <typeparam name="T">The type of the items.</typeparam>
-    public sealed class BatchableSource<T>
+    public sealed class BatchableSource<T> : IBatchableSource<T>
     {
         #region static
         // private static methods

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

@@ -38,6 +38,7 @@ public class Feature
         private static readonly Feature __bypassDocumentValidation = new Feature("BypassDocumentValidation", new SemanticVersion(3, 2, 0));
         private static readonly Feature __changeStreamStage = new Feature("ChangeStreamStage", new SemanticVersion(3, 5, 11));
         private static readonly CollationFeature __collation = new CollationFeature("Collation", new SemanticVersion(3, 3, 11));
+        private static readonly Feature __commandMessage = new Feature("CommandMessage", new SemanticVersion(3, 6, 0));
         private static readonly CommandsThatWriteAcceptWriteConcernFeature __commandsThatWriteAcceptWriteConcern = new CommandsThatWriteAcceptWriteConcernFeature("CommandsThatWriteAcceptWriteConcern", new SemanticVersion(3, 3, 11));
         private static readonly Feature __createIndexesCommand = new Feature("CreateIndexesCommand", new SemanticVersion(3, 0, 0));
         private static readonly Feature __currentOpCommand = new Feature("CurrentOpCommand", new SemanticVersion(3, 2, 0));
@@ -136,6 +137,11 @@ public class Feature
         /// </summary>
         public static CollationFeature Collation => __collation;
 
+        /// <summary>
+        /// Gets the command message feature.
+        /// </summary>
+        public static Feature CommandMessage => __commandMessage;
+
         /// <summary>
         /// Gets the commands that write accept write concern feature.
         /// </summary>

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

@@ -0,0 +1,113 @@
+/* Copyright 2018-present MongoDB Inc.
+*
+* Licensed under the Apache License, Version 2.0 (the "License");
+* you may not use this file except in compliance with the License.
+* You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+
+namespace MongoDB.Driver.Core.Misc
+{
+    /// <summary>
+    /// Represents a batch of items that can be split if not all items can be processed at once.
+    /// </summary>
+    /// <typeparam name="T">The type of the items.</typeparam>
+    public interface IBatchableSource<out T>
+    {
+        // properties
+        /// <summary>
+        /// Gets a value indicating whether all items were processed.
+        /// </summary>
+        /// <value>
+        ///   <c>true</c> if all items were processed; otherwise, <c>false</c>.
+        /// </value>
+        bool AllItemsWereProcessed { get; }
+
+        /// <summary>
+        /// Gets a value indicating whether the batch can be split.
+        /// </summary>
+        /// <value>
+        ///   <c>true</c> if the batch can be split; otherwise, <c>false</c>.
+        /// </value>
+        bool CanBeSplit { get; }
+
+        /// <summary>
+        /// Gets the count.
+        /// </summary>
+        /// <value>
+        /// The count.
+        /// </value>
+        int Count { get; }
+
+        /// <summary>
+        /// Gets the items.
+        /// </summary>
+        /// <value>
+        /// The items.
+        /// </value>
+        IReadOnlyList<T> Items { get; }
+
+        /// <summary>
+        /// Gets the offset.
+        /// </summary>
+        /// <value>
+        /// The offset.
+        /// </value>
+        int Offset { get; }
+
+        /// <summary>
+        /// Gets the count of processed items. Equal to zero until SetProcessedCount has been called.
+        /// </summary>
+        /// <value>
+        /// The count of processed items.
+        /// </value>
+        int ProcessedCount { get; }
+
+        // methods
+        /// <summary>
+        /// Advances past the processed items.
+        /// </summary>
+        void AdvancePastProcessedItems();
+
+        /// <summary>
+        /// Gets the items in the batch.
+        /// </summary>
+        /// <returns>
+        /// The items in the batch.
+        /// </returns>
+        IReadOnlyList<T> GetBatchItems();
+
+        /// <summary>
+        /// Gets the items that were processed.
+        /// </summary>
+        /// <returns>
+        /// The items that were processed.
+        /// </returns>
+        IReadOnlyList<T> GetProcessedItems();
+
+        /// <summary>
+        /// Gets the items that were not processed.
+        /// </summary>
+        /// <returns>
+        /// The items that were not processed.
+        /// </returns>
+        IReadOnlyList<T> GetUnprocessedItems();
+
+        /// <summary>
+        /// Sets the processed count.
+        /// </summary>
+        /// <param name="value">The value.</param>
+        void SetProcessedCount(int value);
+    }
+}

src/MongoDB.Driver.Core/Core/Operations/AsyncCursor.cs

@@ -175,7 +175,6 @@ private CursorBatch<TDocument> ExecuteGetMoreCommand(IChannelHandle channel, Can
                 NoOpElementNameValidator.Instance,
                 null, // additionalOptions
                 () => CommandResponseHandling.Return,
-                false, // slaveOk
                 __getMoreCommandResultSerializer,
                 _messageEncoderSettings,
                 cancellationToken);
@@ -194,7 +193,6 @@ private async Task<CursorBatch<TDocument>> ExecuteGetMoreCommandAsync(IChannelHa
                 NoOpElementNameValidator.Instance,
                 null, // additionalOptions
                 () => CommandResponseHandling.Return,
-                false, // slaveOk
                 __getMoreCommandResultSerializer,
                 _messageEncoderSettings,
                 cancellationToken).ConfigureAwait(false);