CSHARP-2280: Add Watch method to IMongoClient and IMongoDatabase.

Author: rstam

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

@@ -43,6 +43,7 @@ public class AggregateOperation<TResult> : IReadOperation<IAsyncCursor<TResult>>
         private Collation _collation;
         private readonly CollectionNamespace _collectionNamespace;
         private string _comment;
+        private readonly DatabaseNamespace _databaseNamespace;
         private BsonValue _hint;
         private TimeSpan? _maxAwaitTime;
         private TimeSpan? _maxTime;
@@ -53,6 +54,19 @@ public class AggregateOperation<TResult> : IReadOperation<IAsyncCursor<TResult>>
         private bool? _useCursor;
 
         // constructors
+        /// <summary>
+        /// Initializes a new instance of the <see cref="AggregateOperation{TResult}"/> class.
+        /// </summary>
+        /// <param name="databaseNamespace">The database namespace.</param>
+        /// <param name="pipeline">The pipeline.</param>
+        /// <param name="resultSerializer">The result value serializer.</param>
+        /// <param name="messageEncoderSettings">The message encoder settings.</param>
+        public AggregateOperation(DatabaseNamespace databaseNamespace, IEnumerable<BsonDocument> pipeline, IBsonSerializer<TResult> resultSerializer, MessageEncoderSettings messageEncoderSettings)
+            : this(pipeline, resultSerializer, messageEncoderSettings)
+        {
+            _databaseNamespace = Ensure.IsNotNull(databaseNamespace, nameof(databaseNamespace));
+        }
+
         /// <summary>
         /// Initializes a new instance of the <see cref="AggregateOperation{TResult}"/> class.
         /// </summary>
@@ -61,8 +75,13 @@ public class AggregateOperation<TResult> : IReadOperation<IAsyncCursor<TResult>>
         /// <param name="resultSerializer">The result value serializer.</param>
         /// <param name="messageEncoderSettings">The message encoder settings.</param>
         public AggregateOperation(CollectionNamespace collectionNamespace, IEnumerable<BsonDocument> pipeline, IBsonSerializer<TResult> resultSerializer, MessageEncoderSettings messageEncoderSettings)
+            : this(pipeline, resultSerializer, messageEncoderSettings)
         {
             _collectionNamespace = Ensure.IsNotNull(collectionNamespace, nameof(collectionNamespace));
+        }
+
+        private AggregateOperation(IEnumerable<BsonDocument> pipeline, IBsonSerializer<TResult> resultSerializer, MessageEncoderSettings messageEncoderSettings)
+        {
             _pipeline = Ensure.IsNotNull(pipeline, nameof(pipeline)).ToList();
             _resultSerializer = Ensure.IsNotNull(resultSerializer, nameof(resultSerializer));
             _messageEncoderSettings = Ensure.IsNotNull(messageEncoderSettings, nameof(messageEncoderSettings));
@@ -125,6 +144,17 @@ public string Comment
             set { _comment = value; }
         }
 
+        /// <summary>
+        /// Gets the database namespace.
+        /// </summary>
+        /// <value>
+        /// The database namespace.
+        /// </value>
+        public DatabaseNamespace DatabaseNamespace
+        {
+            get { return _databaseNamespace; }
+        }
+
         /// <summary>
         /// Gets or sets the hint. This must either be a BsonString representing the index name or a BsonDocument representing the key pattern of the index.
         /// </summary>
@@ -276,7 +306,7 @@ internal BsonDocument CreateCommand(ConnectionDescription connectionDescription,
             var readConcern = ReadConcernHelper.GetReadConcernForCommand(session, connectionDescription, _readConcern);
             var command = new BsonDocument
             {
-                { "aggregate", _collectionNamespace.CollectionName },
+                { "aggregate", _collectionNamespace == null ? (BsonValue)1 : _collectionNamespace.CollectionName },
                 { "pipeline", new BsonArray(_pipeline) },
                 { "allowDiskUse", () => _allowDiskUse.Value, _allowDiskUse.HasValue },
                 { "maxTimeMS", () => MaxTimeHelper.ToMaxTimeMS(_maxTime.Value), _maxTime.HasValue },
@@ -325,7 +355,7 @@ private AsyncCursor<TResult> CreateCursorFromCursorResult(IChannelSourceHandle c
             var getMoreChannelSource = new ServerChannelSource(channelSource.Server, channelSource.Session.Fork());
             return new AsyncCursor<TResult>(
                 getMoreChannelSource,
-                CollectionNamespace,
+                result.CollectionNamespace,
                 command,
                 result.Results,
                 result.CursorId.GetValueOrDefault(0),
@@ -362,6 +392,7 @@ private void EnsureIsReadOnlyPipeline()
         private class AggregateResult
         {
             public long? CursorId;
+            public CollectionNamespace CollectionNamespace;
             public TResult[] Results;
         }
 
@@ -420,18 +451,28 @@ public override AggregateResult Deserialize(BsonDeserializationContext context,
                 while (reader.ReadBsonType() != 0)
                 {
                     var elementName = reader.ReadName();
-                    if (elementName == "id")
-                    {
-                        result.CursorId = new Int64Serializer().Deserialize(context);
-                    }
-                    else if (elementName == "firstBatch")
+                    switch (elementName)
                     {
-                        var arraySerializer = new ArraySerializer<TResult>(_resultSerializer);
-                        result.Results = arraySerializer.Deserialize(context);
-                    }
-                    else
-                    {
-                        reader.SkipValue();
+                        case "id":
+                            result.CursorId = new Int64Serializer().Deserialize(context);
+                            break;
+
+                        case "ns":
+                            var ns = reader.ReadString();
+                            var separatorIndex = ns.IndexOf('.');
+                            var databaseName = ns.Substring(0, separatorIndex);
+                            var collectionName = ns.Substring(separatorIndex + 1);
+                            result.CollectionNamespace = new CollectionNamespace(new DatabaseNamespace(databaseName), collectionName);
+                            break;
+
+                        case "firstBatch":
+                            var arraySerializer = new ArraySerializer<TResult>(_resultSerializer);
+                            result.Results = arraySerializer.Deserialize(context);
+                            break;
+
+                        default:
+                            reader.SkipValue();
+                            break;
                     }
                 }
                 reader.ReadEndDocument();

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

@@ -40,7 +40,6 @@ internal sealed class ChangeStreamCursor<TDocument> : IAsyncCursor<TDocument>
         private IAsyncCursor<RawBsonDocument> _cursor;
         private bool _disposed;
         private IBsonSerializer<TDocument> _documentSerializer;
-        private BsonDocument _resumeToken;
 
         // public properties
         /// <inheritdoc />
@@ -90,7 +89,7 @@ public void Dispose()
             {
                 if (RetryabilityHelper.IsResumableChangeStreamException(ex))
                 {
-                    _cursor = _changeStreamOperation.Resume(_binding, _resumeToken, cancellationToken);
+                    _cursor = _changeStreamOperation.Resume(_binding, cancellationToken);
                     hasMore = _cursor.MoveNext(cancellationToken);
                 }
                 else
@@ -115,7 +114,7 @@ public void Dispose()
             {
                 if (RetryabilityHelper.IsResumableChangeStreamException(ex))
                 {
-                    _cursor = await _changeStreamOperation.ResumeAsync(_binding, _resumeToken, cancellationToken).ConfigureAwait(false);
+                    _cursor = await _changeStreamOperation.ResumeAsync(_binding, cancellationToken).ConfigureAwait(false);
                     hasMore = await _cursor.MoveNextAsync(cancellationToken).ConfigureAwait(false);
                 }
                 else
@@ -160,7 +159,8 @@ private IEnumerable<TDocument> DeserializeDocuments(IEnumerable<RawBsonDocument>
 
             if (lastRawDocument != null)
             {
-                _resumeToken = lastRawDocument["_id"].DeepClone().AsBsonDocument;
+                _changeStreamOperation.ResumeAfter = lastRawDocument["_id"].DeepClone().AsBsonDocument;
+                _changeStreamOperation.StartAtOperationTime = null;
             }
 
             return documents;