diff --git a/src/MongoDB.Driver/Core/Operations/ChangeStreamCursor.cs b/src/MongoDB.Driver/Core/Operations/ChangeStreamCursor.cs index a0a23274041..06208f77381 100644 --- a/src/MongoDB.Driver/Core/Operations/ChangeStreamCursor.cs +++ b/src/MongoDB.Driver/Core/Operations/ChangeStreamCursor.cs @@ -14,8 +14,6 @@ */ using MongoDB.Bson; -using MongoDB.Bson.IO; -using MongoDB.Bson.Serialization; using MongoDB.Driver.Core.Bindings; using MongoDB.Driver.Core.Misc; using System; @@ -36,10 +34,8 @@ internal sealed class ChangeStreamCursor : IChangeStreamCursor _changeStreamOperation; private IEnumerable _current; - private IAsyncCursor _cursor; + private IAsyncCursor _cursor; private bool _disposed; - private BsonDocument _documentResumeToken; - private readonly IBsonSerializer _documentSerializer; private readonly BsonTimestamp _initialOperationTime; private readonly ICoreSessionHandle _session; private BsonDocument _postBatchResumeToken; @@ -57,7 +53,6 @@ internal sealed class ChangeStreamCursor : IChangeStreamCursor class. /// /// The cursor. - /// The document serializer. /// The binding. /// The session. /// The change stream operation. @@ -68,8 +63,7 @@ internal sealed class ChangeStreamCursor : IChangeStreamCursorThe start at operation time value. /// The maximum wire version. public ChangeStreamCursor( - IAsyncCursor cursor, - IBsonSerializer documentSerializer, + IAsyncCursor cursor, IReadBinding binding, ICoreSessionHandle session, IChangeStreamOperation changeStreamOperation, @@ -81,7 +75,6 @@ public ChangeStreamCursor( int maxWireVersion) { _cursor = Ensure.IsNotNull(cursor, nameof(cursor)); - _documentSerializer = Ensure.IsNotNull(documentSerializer, nameof(documentSerializer)); _binding = Ensure.IsNotNull(binding, nameof(binding)); _changeStreamOperation = Ensure.IsNotNull(changeStreamOperation, nameof(changeStreamOperation)); _session = Ensure.IsNotNull(session, nameof(session)); @@ -124,7 +117,6 @@ public BsonDocument GetResumeToken() { return _postBatchResumeToken ?? - _documentResumeToken ?? _initialStartAfter ?? _initialResumeAfter; } @@ -176,45 +168,6 @@ public BsonDocument GetResumeToken() } // private methods - [System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Usage", "CA2202:Do not dispose objects multiple times")] - private TDocument DeserializeDocument(RawBsonDocument rawDocument) - { - using (var stream = new ByteBufferStream(rawDocument.Slice, ownsBuffer: false)) - using (var reader = new BsonBinaryReader(stream)) - { - var context = BsonDeserializationContext.CreateRoot(reader); - return _documentSerializer.Deserialize(context); - } - } - - private IEnumerable DeserializeDocuments(IEnumerable rawDocuments) - { - var documents = new List(); - RawBsonDocument lastRawDocument = null; - - _postBatchResumeToken = ((ICursorBatchInfo)_cursor).PostBatchResumeToken; - - foreach (var rawDocument in rawDocuments) - { - if (!rawDocument.Contains("_id")) - { - throw new MongoClientException("Cannot provide resume functionality when the resume token is missing."); - } - - var document = DeserializeDocument(rawDocument); - documents.Add(document); - - lastRawDocument = rawDocument; - } - - if (lastRawDocument != null) - { - _documentResumeToken = lastRawDocument["_id"].DeepClone().AsBsonDocument; - } - - return documents; - } - private ResumeValues GetResumeValues() { if (_postBatchResumeToken != null) @@ -222,11 +175,6 @@ private ResumeValues GetResumeValues() return new ResumeValues { ResumeAfter = _postBatchResumeToken }; } - if (_documentResumeToken != null) - { - return new ResumeValues { ResumeAfter = _documentResumeToken }; - } - if (_initialStartAfter != null) { return new ResumeValues { ResumeAfter = _initialStartAfter }; @@ -249,17 +197,8 @@ private void ProcessBatch(bool hasMore) { if (hasMore) { - try - { - _current = DeserializeDocuments(_cursor.Current); - } - finally - { - foreach (var rawDocument in _cursor.Current) - { - rawDocument.Dispose(); - } - } + _current = _cursor.Current; + _postBatchResumeToken = ((ICursorBatchInfo)_cursor).PostBatchResumeToken; } else { @@ -275,7 +214,7 @@ private void ReconfigureOperationResumeValues() _changeStreamOperation.StartAtOperationTime = resumeValues.StartAtOperationTime; } - private IAsyncCursor Resume(CancellationToken cancellationToken) + private IAsyncCursor Resume(CancellationToken cancellationToken) { ReconfigureOperationResumeValues(); // TODO: CSOT implement proper way to obtain the operationContext @@ -283,7 +222,7 @@ private IAsyncCursor Resume(CancellationToken cancellationToken return _changeStreamOperation.Resume(operationContext, _binding); } - private async Task> ResumeAsync(CancellationToken cancellationToken) + private async Task> ResumeAsync(CancellationToken cancellationToken) { ReconfigureOperationResumeValues(); // TODO: CSOT implement proper way to obtain the operationContext diff --git a/src/MongoDB.Driver/Core/Operations/ChangeStreamOperation.cs b/src/MongoDB.Driver/Core/Operations/ChangeStreamOperation.cs index c676d139017..45c485b7c96 100644 --- a/src/MongoDB.Driver/Core/Operations/ChangeStreamOperation.cs +++ b/src/MongoDB.Driver/Core/Operations/ChangeStreamOperation.cs @@ -19,7 +19,6 @@ using System.Threading.Tasks; using MongoDB.Bson; using MongoDB.Bson.Serialization; -using MongoDB.Bson.Serialization.Serializers; using MongoDB.Driver.Core.Bindings; using MongoDB.Driver.Core.Misc; using MongoDB.Driver.Core.WireProtocol.Messages.Encoders; @@ -29,12 +28,11 @@ namespace MongoDB.Driver.Core.Operations internal interface IChangeStreamOperation : IReadOperation> { BsonDocument ResumeAfter { get; set; } - bool? ShowExpandedEvents { get; set; } BsonDocument StartAfter { get; set; } BsonTimestamp StartAtOperationTime { get; set; } - IAsyncCursor Resume(OperationContext operationContext, IReadBinding binding); - Task> ResumeAsync(OperationContext operationContext, IReadBinding binding); + IAsyncCursor Resume(OperationContext operationContext, IReadBinding binding); + Task> ResumeAsync(OperationContext operationContext, IReadBinding binding); } internal sealed class ChangeStreamOperation : IChangeStreamOperation @@ -280,7 +278,7 @@ public IChangeStreamCursor Execute(OperationContext operationContext, I throw new ArgumentException("The binding value passed to ChangeStreamOperation.Execute must implement IReadBindingHandle.", nameof(binding)); } - IAsyncCursor cursor; + IAsyncCursor cursor; ICursorBatchInfo cursorBatchInfo; BsonTimestamp initialOperationTime; using (var context = new RetryableReadContext(binding, _retryRequested, _maxAdaptiveRetries, _enableOverloadRetargeting)) @@ -293,7 +291,6 @@ public IChangeStreamCursor Execute(OperationContext operationContext, I return new ChangeStreamCursor( cursor, - _resultSerializer, bindingHandle.Fork(), operationContext.Session.Fork(), this, @@ -316,7 +313,7 @@ public async Task> ExecuteAsync(OperationContext op throw new ArgumentException("The binding value passed to ChangeStreamOperation.ExecuteAsync must implement IReadBindingHandle.", nameof(binding)); } - IAsyncCursor cursor; + IAsyncCursor cursor; ICursorBatchInfo cursorBatchInfo; BsonTimestamp initialOperationTime; using (var context = new RetryableReadContext(binding, _retryRequested, _maxAdaptiveRetries, _enableOverloadRetargeting)) @@ -329,7 +326,6 @@ public async Task> ExecuteAsync(OperationContext op return new ChangeStreamCursor( cursor, - _resultSerializer, bindingHandle.Fork(), operationContext.Session.Fork(), this, @@ -343,7 +339,7 @@ public async Task> ExecuteAsync(OperationContext op } /// - public IAsyncCursor Resume(OperationContext operationContext, IReadBinding binding) + public IAsyncCursor Resume(OperationContext operationContext, IReadBinding binding) { using (var context = new RetryableReadContext(binding, retryRequested: false, _maxAdaptiveRetries, _enableOverloadRetargeting)) { @@ -352,7 +348,7 @@ public IAsyncCursor Resume(OperationContext operationContext, I } /// - public async Task> ResumeAsync(OperationContext operationContext, IReadBinding binding) + public async Task> ResumeAsync(OperationContext operationContext, IReadBinding binding) { using (var context = new RetryableReadContext(binding, retryRequested: false, _maxAdaptiveRetries, _enableOverloadRetargeting)) { @@ -361,15 +357,15 @@ public async Task> ResumeAsync(OperationContext op } // private methods - private AggregateOperation CreateAggregateOperation() + private AggregateOperation CreateAggregateOperation() { var changeStreamStage = CreateChangeStreamStage(); var combinedPipeline = CreateCombinedPipeline(changeStreamStage); - AggregateOperation operation; + AggregateOperation operation; if (_collectionNamespace != null) { - operation = new AggregateOperation(_collectionNamespace, combinedPipeline, RawBsonDocumentSerializer.Instance, _messageEncoderSettings) + operation = new AggregateOperation(_collectionNamespace, combinedPipeline, _resultSerializer, _messageEncoderSettings) { EnableOverloadRetargeting = _enableOverloadRetargeting, MaxAdaptiveRetries = _maxAdaptiveRetries, @@ -379,7 +375,7 @@ private AggregateOperation CreateAggregateOperation() else { var databaseNamespace = _databaseNamespace ?? DatabaseNamespace.Admin; - operation = new AggregateOperation(databaseNamespace, combinedPipeline, RawBsonDocumentSerializer.Instance, _messageEncoderSettings) + operation = new AggregateOperation(databaseNamespace, combinedPipeline, _resultSerializer, _messageEncoderSettings) { EnableOverloadRetargeting = _enableOverloadRetargeting, MaxAdaptiveRetries = _maxAdaptiveRetries, @@ -419,13 +415,13 @@ private List CreateCombinedPipeline(BsonDocument changeStreamStage return combinedPipeline; } - private IAsyncCursor ExecuteAggregateOperation(OperationContext operationContext, RetryableReadContext context) + private IAsyncCursor ExecuteAggregateOperation(OperationContext operationContext, RetryableReadContext context) { var aggregateOperation = CreateAggregateOperation(); return aggregateOperation.Execute(operationContext, context); } - private Task> ExecuteAggregateOperationAsync(OperationContext operationContext, RetryableReadContext context) + private Task> ExecuteAggregateOperationAsync(OperationContext operationContext, RetryableReadContext context) { var aggregateOperation = CreateAggregateOperation(); return aggregateOperation.ExecuteAsync(operationContext, context); diff --git a/tests/MongoDB.Driver.Tests/Core/Operations/ChangeStreamCursorTests.cs b/tests/MongoDB.Driver.Tests/Core/Operations/ChangeStreamCursorTests.cs index 9d235773f48..c9122d90d72 100644 --- a/tests/MongoDB.Driver.Tests/Core/Operations/ChangeStreamCursorTests.cs +++ b/tests/MongoDB.Driver.Tests/Core/Operations/ChangeStreamCursorTests.cs @@ -41,18 +41,16 @@ public class ChangeStreamCursorTests : OperationTestBase #endregion [Theory] - [InlineData("{ 'd' : '4' }", "{ 'a' : '1' }", "{ 'b' : '2' }", 3L, "{ 'c' : '3' }", null, "{ 'd' : '4' }", null)] - [InlineData(null, "{ 'a' : '1' }", "{ 'b' : '2' }", 3L, "{ 'c' : '3' }", null, "{ 'c' : '3' }", null)] - [InlineData(null, "{ 'a' : '1' }", "{ 'b' : '2' }", 3L, null, null, "{ 'b' : '2' }", null)] - [InlineData(null, "{ 'a' : '1' }", null, 3L, null, null, "{ 'a' : '1' }", null)] - [InlineData(null, null, null, 3L, null, null, null, 3L)] - [InlineData(null, null, null, null, null, 4L, null, 4L)] + [InlineData("{ 'd' : '4' }", "{ 'a' : '1' }", "{ 'b' : '2' }", 3L, null, "{ 'd' : '4' }", null)] + [InlineData(null, "{ 'a' : '1' }", "{ 'b' : '2' }", 3L, null, "{ 'b' : '2' }", null)] + [InlineData(null, "{ 'a' : '1' }", null, 3L, null, "{ 'a' : '1' }", null)] + [InlineData(null, null, null, 3L, null, null, 3L)] + [InlineData(null, null, null, null, 4L, null, 4L)] public void ChangeStreamOperation_should_have_expected_change_stream_operation_options_for_resume_process_after_resumable_error( string postBatchResumeTokenJson, string resumeAfterJson, string startAfterJson, object startAtOperationTimeValue, - string documentResumeTokenJson, object initialOperationTimeObj, string expectedResumeAfter, object expectedStartAtOperationTimeValue) @@ -61,7 +59,6 @@ public void ChangeStreamOperation_should_have_expected_change_stream_operation_o var resumeAfter = resumeAfterJson != null ? BsonDocument.Parse(resumeAfterJson) : null; var startAfter = startAfterJson != null ? BsonDocument.Parse(startAfterJson) : null; var startAtOperationTime = startAtOperationTimeValue != null ? BsonTimestamp.Create(startAtOperationTimeValue) : null; - var documentResumeToken = documentResumeTokenJson != null ? BsonDocument.Parse(documentResumeTokenJson) : null; var initialOperationTime = initialOperationTimeObj != null ? BsonTimestamp.Create(initialOperationTimeObj) : null; var mockCursor = CreateMockCursor(); @@ -73,8 +70,6 @@ public void ChangeStreamOperation_should_have_expected_change_stream_operation_o postBatchResumeToken: postBatchResumeToken, initialOperationTime: initialOperationTime); - subject._documentResumeToken(documentResumeToken); - var result = subject.GetResumeValues(); result.ResumeAfter.Should().Be(expectedResumeAfter != null ? BsonDocument.Parse(expectedResumeAfter) : null); @@ -85,8 +80,7 @@ public void ChangeStreamOperation_should_have_expected_change_stream_operation_o [Fact] public void constructor_should_initialize_instance() { - var cursor = new Mock>().Object; - var documentSerializer = new Mock>().Object; + var cursor = new Mock>().Object; var binding = new Mock().Object; var session = new Mock().Object; var initialOperationTime = new BsonTimestamp(3L); @@ -98,7 +92,6 @@ public void constructor_should_initialize_instance() var subject = new ChangeStreamCursor( cursor, - documentSerializer, binding, session, changeStreamOperation, @@ -114,7 +107,6 @@ public void constructor_should_initialize_instance() subject._current().Should().BeNull(); subject._cursor().Should().BeSameAs(cursor); subject._disposed().Should().BeFalse(); - subject._documentSerializer().Should().BeSameAs(documentSerializer); subject._postBatchResumeToken().Should().BeSameAs(postBatchResumeToken); subject._initialOperationTime().Should().BeSameAs(initialOperationTime); subject._initialStartAfter().Should().Be(startAfter); @@ -127,82 +119,62 @@ public void constructor_should_initialize_instance() [Fact] public void constructor_should_throw_when_cursor_is_null() { - var documentSerializer = new Mock>().Object; var binding = new Mock().Object; var session = new Mock().Object; var initialOperationTime = new BsonTimestamp(3L); var postBatchResumeToken = Mock.Of(); var changeStreamOperation = CreateChangeStreamOperation(); - var exception = Record.Exception(() => new ChangeStreamCursor(null, documentSerializer, binding, session, changeStreamOperation, postBatchResumeToken, initialOperationTime, null, null, null, __dummyMaxWireVersion)); + var exception = Record.Exception(() => new ChangeStreamCursor(null, binding, session, changeStreamOperation, postBatchResumeToken, initialOperationTime, null, null, null, __dummyMaxWireVersion)); var argumnetNullException = exception.Should().BeOfType().Subject; argumnetNullException.ParamName.Should().Be("cursor"); } - [Fact] - public void constructor_should_throw_when_documentSerializer_is_null() - { - var cursor = new Mock>().Object; - var binding = new Mock().Object; - var session = new Mock().Object; - var initialOperationTime = new BsonTimestamp(3L); - var postBatchResumeToken = Mock.Of(); - var changeStreamOperation = CreateChangeStreamOperation(); - - var exception = Record.Exception(() => new ChangeStreamCursor(cursor, null, binding, session, changeStreamOperation, postBatchResumeToken, initialOperationTime, null, null, null, __dummyMaxWireVersion)); - - var argumnetNullException = exception.Should().BeOfType().Subject; - argumnetNullException.ParamName.Should().Be("documentSerializer"); - } - [Fact] public void constructor_should_throw_when_binding_is_null() { - var cursor = new Mock>().Object; - var documentSerializer = new Mock>().Object; + var cursor = new Mock>().Object; var session = new Mock().Object; var initialOperationTime = new BsonTimestamp(3L); var postBatchResumeToken = Mock.Of(); var changeStreamOperation = CreateChangeStreamOperation(); - var exception = Record.Exception(() => new ChangeStreamCursor(cursor, documentSerializer, null, session, changeStreamOperation, postBatchResumeToken, initialOperationTime, null, null, null, __dummyMaxWireVersion)); + var exception = Record.Exception(() => new ChangeStreamCursor(cursor, null, session, changeStreamOperation, postBatchResumeToken, initialOperationTime, null, null, null, __dummyMaxWireVersion)); - var argumnetNullException = exception.Should().BeOfType().Subject; - argumnetNullException.ParamName.Should().Be("binding"); + exception.Should().BeOfType().Subject + .ParamName.Should().Be("binding"); } [Fact] public void constructor_should_throw_when_changeStreamOperation_is_null() { - var cursor = new Mock>().Object; - var documentSerializer = new Mock>().Object; + var cursor = new Mock>().Object; var initialOperationTime = new BsonTimestamp(3L); var postBatchResumeToken = Mock.Of(); var binding = new Mock().Object; var session = new Mock().Object; - var exception = Record.Exception(() => new ChangeStreamCursor(cursor, documentSerializer, binding, session, null, postBatchResumeToken, initialOperationTime, null, null, null, __dummyMaxWireVersion)); + var exception = Record.Exception(() => new ChangeStreamCursor(cursor, binding, session, null, postBatchResumeToken, initialOperationTime, null, null, null, __dummyMaxWireVersion)); - var argumnetNullException = exception.Should().BeOfType().Subject; - argumnetNullException.ParamName.Should().Be("changeStreamOperation"); + exception.Should().BeOfType().Subject + .ParamName.Should().Be("changeStreamOperation"); } [Fact] public void constructor_should_throw_when_maxWireVersion_is_negative() { - var cursor = new Mock>().Object; - var documentSerializer = new Mock>().Object; + var cursor = new Mock>().Object; var initialOperationTime = new BsonTimestamp(3L); var postBatchResumeToken = Mock.Of(); var binding = new Mock().Object; var session = new Mock().Object; var changeStreamOperation = CreateChangeStreamOperation(); - var exception = Record.Exception(() => new ChangeStreamCursor(cursor, documentSerializer, binding, session, changeStreamOperation, postBatchResumeToken, initialOperationTime, null, null, null, -1)); + var exception = Record.Exception(() => new ChangeStreamCursor(cursor, binding, session, changeStreamOperation, postBatchResumeToken, initialOperationTime, null, null, null, -1)); - var argumnetNullException = exception.Should().BeOfType().Subject; - argumnetNullException.ParamName.Should().Be("maxWireVersion"); + exception.Should().BeOfType().Subject + .ParamName.Should().Be("maxWireVersion"); } [Fact] @@ -218,7 +190,7 @@ public void Dispose_should_set_disposed_to_true() [Fact] public void Dispose_should_call_Dispose_on_cursor() { - var mockCursor = new Mock>(); + var mockCursor = new Mock>(); var subject = CreateSubject(cursor: mockCursor.Object); subject.Dispose(); @@ -240,7 +212,7 @@ public void Dispose_should_call_Dispose_on_binding() [Fact] public void Dispose_can_be_called_more_than_once() { - var mockCursor = new Mock>(); + var mockCursor = new Mock>(); var mockBinding = new Mock(); var subject = CreateSubject(cursor: mockCursor.Object, binding: mockBinding.Object); @@ -264,7 +236,7 @@ public async Task DisposeAsync_should_set_disposed_to_true() [Fact] public async Task DisposeAsync_should_call_DisposeAsync_on_cursor() { - var mockCursor = new Mock>(); + var mockCursor = new Mock>(); var subject = CreateSubject(cursor: mockCursor.Object); await subject.DisposeAsync(); @@ -286,7 +258,7 @@ public async Task DisposeAsync_should_call_Dispose_on_binding() [Fact] public async Task DisposeAsync_can_be_called_more_than_once() { - var mockCursor = new Mock>(); + var mockCursor = new Mock>(); var mockBinding = new Mock(); var subject = CreateSubject(cursor: mockCursor.Object, binding: mockBinding.Object); @@ -298,21 +270,18 @@ public async Task DisposeAsync_can_be_called_more_than_once() } [Theory] - [InlineData("{ a : 1 }", "{ b : 2 }", "{ c : 3 }", "{ d : 4 }", "{ a : 1 }")] - [InlineData(null, "{ b : 2 }", "{ c : 3 }", "{ d : 4 }", "{ b : 2 }")] - [InlineData(null, null, "{ c : 3 }", "{ d : 4 }", "{ c : 3 }")] - [InlineData(null, null, null, "{ d : 4 }", "{ d : 4 }")] - [InlineData(null, null, null, null, null)] + [InlineData("{ a : 1 }", "{ b : 2 }", "{ c : 3 }", "{ a : 1 }")] + [InlineData(null, "{ b : 2 }", "{ c : 3 }", "{ b : 2 }")] + [InlineData(null, null, "{ c : 3 }", "{ c : 3 }")] + [InlineData(null, null, null, null)] public void GetResumeToken_should_return_expected_result( string postBatchResumeTokenJson, - string documentResumeTokenJson, string startAfterJson, string resumeAfterJson, string expectedResult) { var mockCursor = CreateMockCursor(); var postBatchResumeToken = postBatchResumeTokenJson != null ? BsonDocument.Parse(postBatchResumeTokenJson) : null; - var documentResumeToken = documentResumeTokenJson != null ? BsonDocument.Parse(documentResumeTokenJson) : null; var startAfter = startAfterJson != null ? BsonDocument.Parse(startAfterJson) : null; var resumeAfter = resumeAfterJson != null ? BsonDocument.Parse(resumeAfterJson) : null; @@ -321,7 +290,6 @@ public void GetResumeToken_should_return_expected_result( postBatchResumeToken: postBatchResumeToken, startAfter: startAfter, resumeAfter: resumeAfter); - subject._documentResumeToken(documentResumeToken); subject.GetResumeToken().Should().Be(expectedResult); } @@ -462,9 +430,8 @@ public void MoveNext_should_call_Resume_after_resumable_exception( var mockResumedCursor = CreateMockCursor(); // process the first batch so that we have a resume token - var resumeToken = BsonDocument.Parse("{ resumeToken : 1 }"); - var firstDocument = BsonDocument.Parse("{ _id : { resumeToken : 1 }, operationType : \"insert\", ns : { db : \"db\", coll : \"coll\" }, documentKey : { _id : 1 }, fullDocument : { _id : 1 } }"); - var firstBatch = new[] { ToRawDocument(firstDocument) }; + var firstDocument = BsonDocument.Parse("{ _id : { resumeToken : 1 }, operationType : 'insert', ns : { db : 'db', coll : 'coll' }, documentKey : { _id : 1 }, fullDocument : { _id : 1 } }"); + var firstBatch = new[] { firstDocument }; mockCursor.Setup(c => c.MoveNext(It.IsAny())).Returns(true); mockCursor.SetupGet(c => c.Current).Returns(firstBatch); subject.MoveNext(CancellationToken.None); @@ -498,143 +465,6 @@ public void MoveNext_should_call_Resume_after_resumable_exception( result.Should().Be(expectedResult); } - [Theory] - [ParameterAttributeData] - void ProcessBatch_should_deserialize_documents( - [Values(false, true)] bool async) - { - var mockCursor = CreateMockCursor(); - var mockSerializer = new Mock>(); - var subject = CreateSubject(cursor: mockCursor.Object, documentSerializer: mockSerializer.Object); - var document = BsonDocument.Parse("{ _id : { resumeAfter : 1 }, operationType : \"insert\", ns : { db : \"db\", coll : \"coll\" }, documentKey : { _id : 1 }, fullDocument : { _id : 1 } }"); - var rawDocuments = new[] { ToRawDocument(document) }; - using var cancellationTokenSource = new CancellationTokenSource(); - var cancellationToken = cancellationTokenSource.Token; - - mockCursor.SetupGet(c => c.Current).Returns(rawDocuments); - mockSerializer.Setup(s => s.Deserialize(It.IsAny(), It.IsAny())).Returns(document); - - bool result; - if (async) - { - mockCursor.Setup(c => c.MoveNextAsync(cancellationToken)).Returns(Task.FromResult(true)); - - result = subject.MoveNextAsync(cancellationToken).GetAwaiter().GetResult(); - } - else - { - mockCursor.Setup(c => c.MoveNext(cancellationToken)).Returns(true); - - result = subject.MoveNext(cancellationToken); - } - - result.Should().BeTrue(); - subject.Current.Should().Equal(new[] { document }); - mockSerializer.Verify(s => s.Deserialize(It.IsAny(), It.IsAny()), Times.Once); - } - - [Theory] - [ParameterAttributeData] - void ProcessBatch_should_Dispose_rawDocuments( - [Values(false, true)] bool async) - { - var mockCursor = CreateMockCursor(); - var mockSerializer = new Mock>(); - var subject = CreateSubject(cursor: mockCursor.Object, documentSerializer: mockSerializer.Object); - var document = BsonDocument.Parse("{ _id : { resumeAfter : 1 }, operationType : \"insert\", ns : { db : \"db\", coll : \"coll\" }, documentKey : { _id : 1 }, fullDocument : { _id : 1 } }"); - var rawDocument = ToRawDocument(document); - var rawDocuments = new[] { rawDocument }; - using var cancellationTokenSource = new CancellationTokenSource(); - var cancellationToken = cancellationTokenSource.Token; - - mockCursor.SetupGet(c => c.Current).Returns(rawDocuments); - mockSerializer.Setup(s => s.Deserialize(It.IsAny(), It.IsAny())).Returns(document); - - bool result; - if (async) - { - mockCursor.Setup(c => c.MoveNextAsync(cancellationToken)).Returns(Task.FromResult(true)); - - result = subject.MoveNextAsync(cancellationToken).GetAwaiter().GetResult(); - } - else - { - mockCursor.Setup(c => c.MoveNext(cancellationToken)).Returns(true); - - result = subject.MoveNext(cancellationToken); - } - - var exception = Record.Exception(() => rawDocument.Contains("x")); - exception.Should().BeOfType(); - } - - [Theory] - [ParameterAttributeData] - void ProcessBatch_should_save_documentResumeToken( - [Values(false, true)] bool async) - { - var postBatchResumeToken = new BsonDocument("a", 1); - var mockCursor = CreateMockCursor(postBatchResumeToken); - var mockSerializer = new Mock>(); - var subject = CreateSubject(cursor: mockCursor.Object, documentSerializer: mockSerializer.Object); - var document = BsonDocument.Parse("{ _id : { resumeAfter : 1 }, operationType : \"insert\", ns : { db : \"db\", coll : \"coll\" }, documentKey : { _id : 1 }, fullDocument : { _id : 1 } }"); - var rawDocuments = new[] { ToRawDocument(document) }; - using var cancellationTokenSource = new CancellationTokenSource(); - var cancellationToken = cancellationTokenSource.Token; - - mockCursor.SetupGet(c => c.Current).Returns(rawDocuments); - mockSerializer.Setup(s => s.Deserialize(It.IsAny(), It.IsAny())).Returns(document); - - bool result; - if (async) - { - mockCursor.Setup(c => c.MoveNextAsync(cancellationToken)).Returns(Task.FromResult(true)); - - result = subject.MoveNextAsync(cancellationToken).GetAwaiter().GetResult(); - } - else - { - mockCursor.Setup(c => c.MoveNext(cancellationToken)).Returns(true); - - result = subject.MoveNext(cancellationToken); - } - - subject._documentResumeToken().Should().Be("{ resumeAfter : 1 }"); - subject._postBatchResumeToken().Should().Be(postBatchResumeToken); - } - - [Theory] - [ParameterAttributeData] - void ProcessBatch_should_throw_when_resume_token_is_missing( - [Values(false, true)] bool async) - { - var mockCursor = CreateMockCursor(); - var subject = CreateSubject(cursor: mockCursor.Object); - var document = BsonDocument.Parse("{ operationType : \"insert\", ns : { db : \"db\", coll : \"coll\" }, documentKey : { _id : 1 }, fullDocument : { _id : 1 } }"); - var rawDocuments = new[] { ToRawDocument(document) }; - using var cancellationTokenSource = new CancellationTokenSource(); - var cancellationToken = cancellationTokenSource.Token; - - mockCursor.SetupGet(c => c.Current).Returns(rawDocuments); - - Exception exception; - if (async) - { - mockCursor.Setup(c => c.MoveNextAsync(cancellationToken)).Returns(Task.FromResult(true)); - - exception = Record.Exception(() => subject.MoveNextAsync(cancellationToken).GetAwaiter().GetResult()); - } - else - { - mockCursor.Setup(c => c.MoveNext(cancellationToken)).Returns(true); - - exception = Record.Exception(() => subject.MoveNext(cancellationToken)); - } - - exception.Should().BeOfType(); - exception.Message.Should().Be("Cannot provide resume functionality when the resume token is missing."); - } - // private methods private ChangeStreamOperation CreateChangeStreamOperation() { @@ -652,16 +482,15 @@ private Task CreateFaultedTask(Exception exception) return completionSource.Task; } - private Mock> CreateMockCursor(BsonDocument postBatchResumeToken = null) + private Mock> CreateMockCursor(BsonDocument postBatchResumeToken = null) { var mockBatchInfo = new Mock(); mockBatchInfo.Setup(c => c.PostBatchResumeToken).Returns(postBatchResumeToken); - return mockBatchInfo.As>(); + return mockBatchInfo.As>(); } private ChangeStreamCursor CreateSubject( - IAsyncCursor cursor = null, - IBsonSerializer documentSerializer = null, + IAsyncCursor cursor = null, IReadBinding binding = null, ICoreSessionHandle session = null, IChangeStreamOperation changeStreamOperation = null, @@ -671,12 +500,11 @@ private ChangeStreamCursor CreateSubject( BsonTimestamp startAtOperationTime = null, BsonTimestamp initialOperationTime = null) { - cursor ??= new Mock>().Object; - documentSerializer ??= new Mock>().Object; + cursor ??= new Mock>().Object; binding ??= new Mock().Object; session ??= new Mock().Object; changeStreamOperation ??= Mock.Of>(); - return new ChangeStreamCursor(cursor, documentSerializer, binding, session, changeStreamOperation, postBatchResumeToken, initialOperationTime, startAfter, resumeAfter, startAtOperationTime, __dummyMaxWireVersion); + return new ChangeStreamCursor(cursor, binding, session, changeStreamOperation, postBatchResumeToken, initialOperationTime, startAfter, resumeAfter, startAtOperationTime, __dummyMaxWireVersion); } private BsonDocument GenerateResumeAfterToken(bool async, bool shouldBeEmpty = false) @@ -705,15 +533,6 @@ private BsonDocument GenerateResumeAfterToken(bool async, bool shouldBeEmpty = f return enumerator.Current.ResumeToken; } } - - private RawBsonDocument ToRawDocument(BsonDocument document) - { - using (var reader = new BsonDocumentReader(document)) - { - var context = BsonDeserializationContext.CreateRoot(reader); - return RawBsonDocumentSerializer.Instance.Deserialize(context); - } - } } internal static class ChangeStreamCursorReflector @@ -727,21 +546,12 @@ public static IChangeStreamOperation _changeStreamOperation(this C public static IEnumerable _current(this ChangeStreamCursor cursor) => (IEnumerable)Reflector.GetFieldValue(cursor, nameof(_current)); - public static IAsyncCursor _cursor(this ChangeStreamCursor cursor) => - (IAsyncCursor)Reflector.GetFieldValue(cursor, nameof(_cursor)); + public static IAsyncCursor _cursor(this ChangeStreamCursor cursor) => + (IAsyncCursor)Reflector.GetFieldValue(cursor, nameof(_cursor)); public static bool _disposed(this ChangeStreamCursor cursor) => (bool)Reflector.GetFieldValue(cursor, nameof(_disposed)); - public static IBsonSerializer _documentSerializer(this ChangeStreamCursor cursor) => - (IBsonSerializer)Reflector.GetFieldValue(cursor, nameof(_documentSerializer)); - - public static BsonDocument _documentResumeToken(this IChangeStreamCursor cursor) => - (BsonDocument)Reflector.GetFieldValue(cursor, nameof(_documentResumeToken)); - - public static void _documentResumeToken(this IChangeStreamCursor cursor, BsonDocument value) => - Reflector.SetFieldValue(cursor, nameof(_documentResumeToken), value); - public static ChangeStreamCursor.ResumeValues GetResumeValues(this IChangeStreamCursor cursor) => (ChangeStreamCursor.ResumeValues)Reflector.Invoke(cursor, nameof(GetResumeValues)); diff --git a/tests/MongoDB.Driver.Tests/Core/Operations/ChangeStreamOperationTests.cs b/tests/MongoDB.Driver.Tests/Core/Operations/ChangeStreamOperationTests.cs index fd7860f56dc..8b38fc0e72f 100644 --- a/tests/MongoDB.Driver.Tests/Core/Operations/ChangeStreamOperationTests.cs +++ b/tests/MongoDB.Driver.Tests/Core/Operations/ChangeStreamOperationTests.cs @@ -702,7 +702,7 @@ public void CreateAggregateOperation_should_return_expected_result( result.MessageEncoderSettings.Should().BeSameAs(messageEncoderSettings); result.Pipeline.Should().Equal(expectedPipeline); result.ReadConcern.Should().Be(readConcern); - result.ResultSerializer.Should().Be(RawBsonDocumentSerializer.Instance); + result.ResultSerializer.Should().Be(BsonDocumentSerializer.Instance); result.RetryRequested.Should().BeFalse(); } @@ -723,9 +723,9 @@ private ChangeStreamOperation CreateSubject( internal static class ChangeStreamOperationReflector { - public static AggregateOperation CreateAggregateOperation(this ChangeStreamOperation subject) + public static AggregateOperation CreateAggregateOperation(this ChangeStreamOperation subject) { - return (AggregateOperation)Reflector.Invoke(subject, nameof(CreateAggregateOperation)); + return (AggregateOperation)Reflector.Invoke(subject, nameof(CreateAggregateOperation)); } public static BsonDocument CreateChangeStreamStage(this ChangeStreamOperation> subject)