Skip to content

Commit d9c44a9

Browse files
papafepapafe
committed
Added retryability to Create/RenameCollectionOperation
1 parent 243c720 commit d9c44a9

File tree

4 files changed

+116
-57
lines changed

4 files changed

+116
-57
lines changed

‎src/MongoDB.Driver/Core/Operations/CreateCollectionOperation.cs‎

Lines changed: 52 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818
using MongoDB.Bson;
1919
using MongoDB.Bson.Serialization.Serializers;
2020
using MongoDB.Driver.Core.Bindings;
21+
using MongoDB.Driver.Core.Connections;
2122
using MongoDB.Driver.Core.Events;
2223
using MongoDB.Driver.Core.Misc;
2324
using MongoDB.Driver.Core.WireProtocol.Messages.Encoders;
@@ -26,7 +27,7 @@
2627

2728
namespace MongoDB.Driver.Core.Operations
2829
{
29-
internal sealed class CreateCollectionOperation : IWriteOperation<BsonDocument>
30+
internal sealed class CreateCollectionOperation : IWriteOperation<BsonDocument>, IRetryableWriteOperation<BsonDocument>
3031
{
3132
#region static
3233

@@ -211,7 +212,7 @@ public BsonDocument ClusteredIndex
211212
set => _clusteredIndex = value;
212213
}
213214

214-
internal BsonDocument CreateCommand(OperationContext operationContext, ICoreSessionHandle session)
215+
internal BsonDocument CreateCommand(OperationContext operationContext, ICoreSessionHandle session, ConnectionDescription connectionDescription, long? transactionNumber)
215216
{
216217
var writeConcern = WriteConcernHelper.GetEffectiveWriteConcern(operationContext, session, _writeConcern);
217218
return new BsonDocument
@@ -232,49 +233,76 @@ internal BsonDocument CreateCommand(OperationContext operationContext, ICoreSess
232233
{ "expireAfterSeconds", () => _expireAfter.Value.TotalSeconds, _expireAfter.HasValue },
233234
{ "timeseries", () => _timeSeriesOptions.ToBsonDocument(), _timeSeriesOptions != null },
234235
{ "encryptedFields", _encryptedFields, _encryptedFields != null },
235-
{ "changeStreamPreAndPostImages", _changeStreamPreAndPostImages, _changeStreamPreAndPostImages != null }
236+
{ "changeStreamPreAndPostImages", _changeStreamPreAndPostImages, _changeStreamPreAndPostImages != null },
237+
{ "txnNumber", () => transactionNumber.Value, transactionNumber.HasValue }
236238
};
237239
}
238240

239241
public BsonDocument Execute(OperationContext operationContext, IWriteBinding binding)
240242
{
241-
Ensure.IsNotNull(binding, nameof(binding));
243+
using (BeginOperation())
244+
{
245+
return RetryableWriteOperationExecutor.Execute(operationContext, this, binding, retryRequested: false);
246+
}
247+
}
242248

249+
public BsonDocument Execute(OperationContext operationContext, RetryableWriteContext context)
250+
{
243251
using (BeginOperation())
244-
using (var channelSource = binding.GetWriteChannelSource(operationContext))
245-
using (var channel = channelSource.GetChannel(operationContext))
246252
{
247-
EnsureServerIsValid(channel.ConnectionDescription.MaxWireVersion);
248-
using (var channelBinding = new ChannelReadWriteBinding(channelSource.Server, channel, binding.Session.Fork()))
249-
{
250-
var operation = CreateOperation(operationContext, channelBinding.Session);
251-
return operation.Execute(operationContext, channelBinding);
252-
}
253+
return RetryableWriteOperationExecutor.Execute(operationContext, this, context);
253254
}
254255
}
255256

256-
public async Task<BsonDocument> ExecuteAsync(OperationContext operationContext, IWriteBinding binding)
257+
public Task<BsonDocument> ExecuteAsync(OperationContext operationContext, IWriteBinding binding)
257258
{
258-
Ensure.IsNotNull(binding, nameof(binding));
259+
using (BeginOperation())
260+
{
261+
return RetryableWriteOperationExecutor.ExecuteAsync(operationContext, this, binding, retryRequested: false);
262+
}
263+
}
259264

265+
public Task<BsonDocument> ExecuteAsync(OperationContext operationContext, RetryableWriteContext context)
266+
{
260267
using (BeginOperation())
261-
using (var channelSource = await binding.GetWriteChannelSourceAsync(operationContext).ConfigureAwait(false))
262-
using (var channel = await channelSource.GetChannelAsync(operationContext).ConfigureAwait(false))
263268
{
264-
EnsureServerIsValid(channel.ConnectionDescription.MaxWireVersion);
265-
using (var channelBinding = new ChannelReadWriteBinding(channelSource.Server, channel, binding.Session.Fork()))
266-
{
267-
var operation = CreateOperation(operationContext, channelBinding.Session);
268-
return await operation.ExecuteAsync(operationContext, channelBinding).ConfigureAwait(false);
269-
}
269+
return RetryableWriteOperationExecutor.ExecuteAsync(operationContext, this, context);
270+
}
271+
}
272+
273+
public BsonDocument ExecuteAttempt(OperationContext operationContext, RetryableWriteContext context, int attempt, long? transactionNumber)
274+
{
275+
var binding = context.Binding;
276+
var channelSource = context.ChannelSource;
277+
var channel = context.Channel;
278+
279+
EnsureServerIsValid(channel.ConnectionDescription.MaxWireVersion);
280+
using (var channelBinding = new ChannelReadWriteBinding(channelSource.Server, channel, binding.Session.Fork()))
281+
{
282+
var operation = CreateOperation(operationContext, channelBinding.Session, channel.ConnectionDescription, transactionNumber);
283+
return operation.Execute(operationContext, channelBinding);
284+
}
285+
}
286+
287+
public async Task<BsonDocument> ExecuteAttemptAsync(OperationContext operationContext, RetryableWriteContext context, int attempt, long? transactionNumber)
288+
{
289+
var binding = context.Binding;
290+
var channelSource = context.ChannelSource;
291+
var channel = context.Channel;
292+
293+
EnsureServerIsValid(channel.ConnectionDescription.MaxWireVersion);
294+
using (var channelBinding = new ChannelReadWriteBinding(channelSource.Server, channel, binding.Session.Fork()))
295+
{
296+
var operation = CreateOperation(operationContext, channelBinding.Session, channel.ConnectionDescription, transactionNumber);
297+
return await operation.ExecuteAsync(operationContext, channelBinding).ConfigureAwait(false);
270298
}
271299
}
272300

273301
private IDisposable BeginOperation() => EventContext.BeginOperation("create");
274302

275-
private WriteCommandOperation<BsonDocument> CreateOperation(OperationContext operationContext, ICoreSessionHandle session)
303+
private WriteCommandOperation<BsonDocument> CreateOperation(OperationContext operationContext, ICoreSessionHandle session, ConnectionDescription connectionDescription, long? transactionNumber)
276304
{
277-
var command = CreateCommand(operationContext, session);
305+
var command = CreateCommand(operationContext, session, connectionDescription, transactionNumber);
278306
return new WriteCommandOperation<BsonDocument>(_collectionNamespace.DatabaseNamespace, command, BsonDocumentSerializer.Instance, _messageEncoderSettings);
279307
}
280308

‎src/MongoDB.Driver/Core/Operations/RenameCollectionOperation.cs‎

Lines changed: 46 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,7 @@
2525

2626
namespace MongoDB.Driver.Core.Operations
2727
{
28-
internal sealed class RenameCollectionOperation : IWriteOperation<BsonDocument>
28+
internal sealed class RenameCollectionOperation : IWriteOperation<BsonDocument>, IRetryableWriteOperation<BsonDocument>
2929
{
3030
private readonly CollectionNamespace _collectionNamespace;
3131
private bool? _dropTarget;
@@ -70,51 +70,82 @@ public WriteConcern WriteConcern
7070
set { _writeConcern = value; }
7171
}
7272

73-
public BsonDocument CreateCommand(OperationContext operationContext, ICoreSessionHandle session, ConnectionDescription connectionDescription)
73+
public BsonDocument CreateCommand(OperationContext operationContext, ICoreSessionHandle session, ConnectionDescription connectionDescription, long? transactionNumber)
7474
{
7575
var writeConcern = WriteConcernHelper.GetEffectiveWriteConcern(operationContext, session, _writeConcern);
7676
return new BsonDocument
7777
{
7878
{ "renameCollection", _collectionNamespace.FullName },
7979
{ "to", _newCollectionNamespace.FullName },
8080
{ "dropTarget", () => _dropTarget.Value, _dropTarget.HasValue },
81-
{ "writeConcern", writeConcern, writeConcern != null }
81+
{ "writeConcern", writeConcern, writeConcern != null },
82+
{ "txnNumber", () => transactionNumber.Value, transactionNumber.HasValue }
8283
};
8384
}
8485

8586
public BsonDocument Execute(OperationContext operationContext, IWriteBinding binding)
8687
{
87-
Ensure.IsNotNull(binding, nameof(binding));
88+
using (BeginOperation())
89+
{
90+
return RetryableWriteOperationExecutor.Execute(operationContext, this, binding, retryRequested: false);
91+
}
92+
}
93+
94+
public BsonDocument Execute(OperationContext operationContext, RetryableWriteContext context)
95+
{
96+
using (BeginOperation())
97+
{
98+
return RetryableWriteOperationExecutor.Execute(operationContext, this, context);
99+
}
100+
}
101+
102+
public Task<BsonDocument> ExecuteAsync(OperationContext operationContext, IWriteBinding binding)
103+
{
104+
using (BeginOperation())
105+
{
106+
return RetryableWriteOperationExecutor.ExecuteAsync(operationContext, this, binding, retryRequested: false);
107+
}
108+
}
88109

110+
public Task<BsonDocument> ExecuteAsync(OperationContext operationContext, RetryableWriteContext context)
111+
{
89112
using (BeginOperation())
90-
using (var channelSource = binding.GetWriteChannelSource(operationContext))
91-
using (var channel = channelSource.GetChannel(operationContext))
113+
{
114+
return RetryableWriteOperationExecutor.ExecuteAsync(operationContext, this, context);
115+
}
116+
}
117+
118+
public BsonDocument ExecuteAttempt(OperationContext operationContext, RetryableWriteContext context, int attempt, long? transactionNumber)
119+
{
120+
var binding = context.Binding;
121+
var channelSource = context.ChannelSource;
122+
var channel = context.Channel;
123+
92124
using (var channelBinding = new ChannelReadWriteBinding(channelSource.Server, channel, binding.Session.Fork()))
93125
{
94-
var operation = CreateOperation(operationContext, channelBinding.Session, channel.ConnectionDescription);
126+
var operation = CreateOperation(operationContext, channelBinding.Session, channel.ConnectionDescription, transactionNumber);
95127
return operation.Execute(operationContext, channelBinding);
96128
}
97129
}
98130

99-
public async Task<BsonDocument> ExecuteAsync(OperationContext operationContext, IWriteBinding binding)
131+
public async Task<BsonDocument> ExecuteAttemptAsync(OperationContext operationContext, RetryableWriteContext context, int attempt, long? transactionNumber)
100132
{
101-
Ensure.IsNotNull(binding, nameof(binding));
133+
var binding = context.Binding;
134+
var channelSource = context.ChannelSource;
135+
var channel = context.Channel;
102136

103-
using (BeginOperation())
104-
using (var channelSource = await binding.GetWriteChannelSourceAsync(operationContext).ConfigureAwait(false))
105-
using (var channel = await channelSource.GetChannelAsync(operationContext).ConfigureAwait(false))
106137
using (var channelBinding = new ChannelReadWriteBinding(channelSource.Server, channel, binding.Session.Fork()))
107138
{
108-
var operation = CreateOperation(operationContext, channelBinding.Session, channel.ConnectionDescription);
139+
var operation = CreateOperation(operationContext, channelBinding.Session, channel.ConnectionDescription, transactionNumber);
109140
return await operation.ExecuteAsync(operationContext, channelBinding).ConfigureAwait(false);
110141
}
111142
}
112143

113144
private IDisposable BeginOperation() => EventContext.BeginOperation("renameCollection");
114145

115-
private WriteCommandOperation<BsonDocument> CreateOperation(OperationContext operationContext, ICoreSessionHandle session, ConnectionDescription connectionDescription)
146+
private WriteCommandOperation<BsonDocument> CreateOperation(OperationContext operationContext, ICoreSessionHandle session, ConnectionDescription connectionDescription, long? transactionNumber)
116147
{
117-
var command = CreateCommand(operationContext, session, connectionDescription);
148+
var command = CreateCommand(operationContext, session, connectionDescription, transactionNumber);
118149
return new WriteCommandOperation<BsonDocument>(DatabaseNamespace.Admin, command, BsonDocumentSerializer.Instance, _messageEncoderSettings);
119150
}
120151
}

0 commit comments

Comments
 (0)