Skip to content

CSHARP-5998: RawBsonDocument refactoring/removal #2085

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
sanych-sun wants to merge 2 commits into
base: main
Choose a base branch
from
Open

CSHARP-5998: RawBsonDocument refactoring/removal #2085

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
fb562e9
CSHARP-5998: RawBsonDocument refactoring/removal
sanych-sun Jul 28, 2026
5f2d996
pr
sanych-sun Jul 29, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
73 changes: 6 additions & 67 deletions src/MongoDB.Driver/Core/Operations/ChangeStreamCursor.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -36,10 +34,8 @@ internal sealed class ChangeStreamCursor<TDocument> : IChangeStreamCursor<TDocum
private readonly IReadBinding _binding;
private readonly IChangeStreamOperation<TDocument> _changeStreamOperation;
private IEnumerable<TDocument> _current;
private IAsyncCursor<RawBsonDocument> _cursor;
private IAsyncCursor<TDocument> _cursor;
private bool _disposed;
private BsonDocument _documentResumeToken;
private readonly IBsonSerializer<TDocument> _documentSerializer;
private readonly BsonTimestamp _initialOperationTime;
private readonly ICoreSessionHandle _session;
private BsonDocument _postBatchResumeToken;
Expand All @@ -57,7 +53,6 @@ internal sealed class ChangeStreamCursor<TDocument> : IChangeStreamCursor<TDocum
/// Initializes a new instance of the <see cref="ChangeStreamCursor{TDocument}" /> class.
/// </summary>
/// <param name="cursor">The cursor.</param>
/// <param name="documentSerializer">The document serializer.</param>
/// <param name="binding">The binding.</param>
/// <param name="session">The session.</param>
/// <param name="changeStreamOperation">The change stream operation.</param>
Expand All @@ -68,8 +63,7 @@ internal sealed class ChangeStreamCursor<TDocument> : IChangeStreamCursor<TDocum
/// <param name="initialStartAtOperationTime">The start at operation time value.</param>
/// <param name="maxWireVersion">The maximum wire version.</param>
public ChangeStreamCursor(
IAsyncCursor<RawBsonDocument> cursor,
IBsonSerializer<TDocument> documentSerializer,
IAsyncCursor<TDocument> cursor,
IReadBinding binding,
ICoreSessionHandle session,
IChangeStreamOperation<TDocument> changeStreamOperation,
Expand All @@ -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));
Expand Down Expand Up @@ -124,7 +117,6 @@ public BsonDocument GetResumeToken()
{
return
_postBatchResumeToken ??
_documentResumeToken ??
_initialStartAfter ??
_initialResumeAfter;
}
Expand Down Expand Up @@ -176,57 +168,13 @@ 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<TDocument> DeserializeDocuments(IEnumerable<RawBsonDocument> rawDocuments)
{
var documents = new List<TDocument>();
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)
{
return new ResumeValues { ResumeAfter = _postBatchResumeToken };
}

if (_documentResumeToken != null)
{
return new ResumeValues { ResumeAfter = _documentResumeToken };
}

if (_initialStartAfter != null)
{
return new ResumeValues { ResumeAfter = _initialStartAfter };
Expand All @@ -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;
}
Comment on lines 198 to 202
else
{
Expand All @@ -275,15 +214,15 @@ private void ReconfigureOperationResumeValues()
_changeStreamOperation.StartAtOperationTime = resumeValues.StartAtOperationTime;
}

private IAsyncCursor<RawBsonDocument> Resume(CancellationToken cancellationToken)
private IAsyncCursor<TDocument> Resume(CancellationToken cancellationToken)
{
ReconfigureOperationResumeValues();
// TODO: CSOT implement proper way to obtain the operationContext
using var operationContext = new OperationContext(_session, null, cancellationToken);
return _changeStreamOperation.Resume(operationContext, _binding);
}

private async Task<IAsyncCursor<RawBsonDocument>> ResumeAsync(CancellationToken cancellationToken)
private async Task<IAsyncCursor<TDocument>> ResumeAsync(CancellationToken cancellationToken)
{
ReconfigureOperationResumeValues();
// TODO: CSOT implement proper way to obtain the operationContext
Expand Down
28 changes: 12 additions & 16 deletions src/MongoDB.Driver/Core/Operations/ChangeStreamOperation.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -29,12 +28,11 @@ namespace MongoDB.Driver.Core.Operations
internal interface IChangeStreamOperation<TResult> : IReadOperation<IChangeStreamCursor<TResult>>
{
BsonDocument ResumeAfter { get; set; }
bool? ShowExpandedEvents { get; set; }
BsonDocument StartAfter { get; set; }
BsonTimestamp StartAtOperationTime { get; set; }

IAsyncCursor<RawBsonDocument> Resume(OperationContext operationContext, IReadBinding binding);
Task<IAsyncCursor<RawBsonDocument>> ResumeAsync(OperationContext operationContext, IReadBinding binding);
IAsyncCursor<TResult> Resume(OperationContext operationContext, IReadBinding binding);
Task<IAsyncCursor<TResult>> ResumeAsync(OperationContext operationContext, IReadBinding binding);
}

internal sealed class ChangeStreamOperation<TResult> : IChangeStreamOperation<TResult>
Expand Down Expand Up @@ -280,7 +278,7 @@ public IChangeStreamCursor<TResult> Execute(OperationContext operationContext, I
throw new ArgumentException("The binding value passed to ChangeStreamOperation.Execute must implement IReadBindingHandle.", nameof(binding));
}

IAsyncCursor<RawBsonDocument> cursor;
IAsyncCursor<TResult> cursor;
ICursorBatchInfo cursorBatchInfo;
BsonTimestamp initialOperationTime;
using (var context = new RetryableReadContext(binding, _retryRequested, _maxAdaptiveRetries, _enableOverloadRetargeting))
Expand All @@ -293,7 +291,6 @@ public IChangeStreamCursor<TResult> Execute(OperationContext operationContext, I

return new ChangeStreamCursor<TResult>(
cursor,
_resultSerializer,
bindingHandle.Fork(),
operationContext.Session.Fork(),
this,
Expand All @@ -316,7 +313,7 @@ public async Task<IChangeStreamCursor<TResult>> ExecuteAsync(OperationContext op
throw new ArgumentException("The binding value passed to ChangeStreamOperation.ExecuteAsync must implement IReadBindingHandle.", nameof(binding));
}

IAsyncCursor<RawBsonDocument> cursor;
IAsyncCursor<TResult> cursor;
ICursorBatchInfo cursorBatchInfo;
BsonTimestamp initialOperationTime;
using (var context = new RetryableReadContext(binding, _retryRequested, _maxAdaptiveRetries, _enableOverloadRetargeting))
Expand All @@ -329,7 +326,6 @@ public async Task<IChangeStreamCursor<TResult>> ExecuteAsync(OperationContext op

return new ChangeStreamCursor<TResult>(
cursor,
_resultSerializer,
bindingHandle.Fork(),
operationContext.Session.Fork(),
this,
Expand All @@ -343,7 +339,7 @@ public async Task<IChangeStreamCursor<TResult>> ExecuteAsync(OperationContext op
}

/// <inheritdoc />
public IAsyncCursor<RawBsonDocument> Resume(OperationContext operationContext, IReadBinding binding)
public IAsyncCursor<TResult> Resume(OperationContext operationContext, IReadBinding binding)
{
using (var context = new RetryableReadContext(binding, retryRequested: false, _maxAdaptiveRetries, _enableOverloadRetargeting))
{
Expand All @@ -352,7 +348,7 @@ public IAsyncCursor<RawBsonDocument> Resume(OperationContext operationContext, I
}

/// <inheritdoc />
public async Task<IAsyncCursor<RawBsonDocument>> ResumeAsync(OperationContext operationContext, IReadBinding binding)
public async Task<IAsyncCursor<TResult>> ResumeAsync(OperationContext operationContext, IReadBinding binding)
{
using (var context = new RetryableReadContext(binding, retryRequested: false, _maxAdaptiveRetries, _enableOverloadRetargeting))
{
Expand All @@ -361,15 +357,15 @@ public async Task<IAsyncCursor<RawBsonDocument>> ResumeAsync(OperationContext op
}

// private methods
private AggregateOperation<RawBsonDocument> CreateAggregateOperation()
private AggregateOperation<TResult> CreateAggregateOperation()
{
var changeStreamStage = CreateChangeStreamStage();
var combinedPipeline = CreateCombinedPipeline(changeStreamStage);

AggregateOperation<RawBsonDocument> operation;
AggregateOperation<TResult> operation;
if (_collectionNamespace != null)
{
operation = new AggregateOperation<RawBsonDocument>(_collectionNamespace, combinedPipeline, RawBsonDocumentSerializer.Instance, _messageEncoderSettings)
operation = new AggregateOperation<TResult>(_collectionNamespace, combinedPipeline, _resultSerializer, _messageEncoderSettings)
{
EnableOverloadRetargeting = _enableOverloadRetargeting,
MaxAdaptiveRetries = _maxAdaptiveRetries,
Expand All @@ -379,7 +375,7 @@ private AggregateOperation<RawBsonDocument> CreateAggregateOperation()
else
{
var databaseNamespace = _databaseNamespace ?? DatabaseNamespace.Admin;
operation = new AggregateOperation<RawBsonDocument>(databaseNamespace, combinedPipeline, RawBsonDocumentSerializer.Instance, _messageEncoderSettings)
operation = new AggregateOperation<TResult>(databaseNamespace, combinedPipeline, _resultSerializer, _messageEncoderSettings)
{
EnableOverloadRetargeting = _enableOverloadRetargeting,
MaxAdaptiveRetries = _maxAdaptiveRetries,
Expand Down Expand Up @@ -419,13 +415,13 @@ private List<BsonDocument> CreateCombinedPipeline(BsonDocument changeStreamStage
return combinedPipeline;
}

private IAsyncCursor<RawBsonDocument> ExecuteAggregateOperation(OperationContext operationContext, RetryableReadContext context)
private IAsyncCursor<TResult> ExecuteAggregateOperation(OperationContext operationContext, RetryableReadContext context)
{
var aggregateOperation = CreateAggregateOperation();
return aggregateOperation.Execute(operationContext, context);
}

private Task<IAsyncCursor<RawBsonDocument>> ExecuteAggregateOperationAsync(OperationContext operationContext, RetryableReadContext context)
private Task<IAsyncCursor<TResult>> ExecuteAggregateOperationAsync(OperationContext operationContext, RetryableReadContext context)
{
var aggregateOperation = CreateAggregateOperation();
return aggregateOperation.ExecuteAsync(operationContext, context);
Expand Down
Loading
Loading