Skip to content

Commit 232e180

Browse files
ealsurj82w
ealsur
authored and
j82w
committed
SynchronizationContext and blocking calls (#1078)
* Adding context * Adding netfx test project * conditional Task.Run * Using TaskHelper * Wrapping more calls * yml update * Adding query in test * Undoing * Removing duplicates * Removing more duplicates * Adding UTs for TaskHelper * Removing unneeded file * Undoing more changes * missing indent * Changing assemblyname * Testing another SNK * YML changes * AssemblyName * Testing properties * Testing with 471 * Testing netcore * net461 * Original setup * Testing with vstest * paths * Using full path * Updating vstest * Forcing version * Refactoring * Undoing other changes * Parameters * breaklines * Undoing some changes * Inline wrapper not conditional * Fixing tests * internal properties * Fixing tests * Fixing more tests * Fixing tests
1 parent 736369f commit 232e180

36 files changed

+1172
-78
lines changed

‎Microsoft.Azure.Cosmos/src/ChangeFeedProcessor/ChangeFeedProcessorBuilder.cs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -162,7 +162,7 @@ public ChangeFeedProcessorBuilder WithLeaseContainer(Container leaseContainer)
162162
if (this.leaseContainer != null) throw new InvalidOperationException("The builder already defined a lease container.");
163163
if (this.LeaseStoreManager != null) throw new InvalidOperationException("The builder already defined an in-memory lease container instance.");
164164

165-
this.leaseContainer = (ContainerCore)leaseContainer;
165+
this.leaseContainer = (ContainerInlineCore)leaseContainer;
166166
return this;
167167
}
168168

‎Microsoft.Azure.Cosmos/src/CosmosClient.cs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -324,7 +324,7 @@ public virtual Task<AccountProperties> ReadAccountAsync()
324324
/// <returns>Cosmos database proxy</returns>
325325
public virtual Database GetDatabase(string id)
326326
{
327-
return new DatabaseCore(this.ClientContext, id);
327+
return new DatabaseInlineCore(new DatabaseCore(this.ClientContext, id));
328328
}
329329

330330
/// <summary>

‎Microsoft.Azure.Cosmos/src/Resource/Conflict/ConflictsInlineCore.cs‎

Lines changed: 79 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,79 @@
1+
//------------------------------------------------------------
2+
// Copyright (c) Microsoft Corporation. All rights reserved.
3+
//------------------------------------------------------------
4+
5+
namespace Microsoft.Azure.Cosmos
6+
{
7+
using System;
8+
using System.Threading;
9+
using System.Threading.Tasks;
10+
11+
// This class acts as a wrapper for environments that use SynchronizationContext.
12+
internal sealed class ConflictsInlineCore : Conflicts
13+
{
14+
private readonly ConflictsCore conflicts;
15+
16+
internal ConflictsInlineCore(ConflictsCore conflicts)
17+
{
18+
if (conflicts == null)
19+
{
20+
throw new ArgumentNullException(nameof(conflicts));
21+
}
22+
23+
this.conflicts = conflicts;
24+
}
25+
26+
public override Task<ResponseMessage> DeleteAsync(
27+
ConflictProperties conflict,
28+
PartitionKey partitionKey,
29+
CancellationToken cancellationToken = default(CancellationToken))
30+
{
31+
return TaskHelper.RunInlineIfNeededAsync(() => this.conflicts.DeleteAsync(conflict, partitionKey, cancellationToken));
32+
}
33+
34+
public override FeedIterator GetConflictQueryStreamIterator(
35+
string queryText = null,
36+
string continuationToken = null,
37+
QueryRequestOptions requestOptions = null)
38+
{
39+
return this.conflicts.GetConflictQueryStreamIterator(queryText, continuationToken, requestOptions);
40+
}
41+
42+
public override FeedIterator<T> GetConflictQueryIterator<T>(
43+
string queryText = null,
44+
string continuationToken = null,
45+
QueryRequestOptions requestOptions = null)
46+
{
47+
return this.conflicts.GetConflictQueryIterator<T>(queryText, continuationToken, requestOptions);
48+
}
49+
50+
public override FeedIterator GetConflictQueryStreamIterator(
51+
QueryDefinition queryDefinition,
52+
string continuationToken = null,
53+
QueryRequestOptions requestOptions = null)
54+
{
55+
return this.conflicts.GetConflictQueryStreamIterator(queryDefinition, continuationToken, requestOptions);
56+
}
57+
58+
public override FeedIterator<T> GetConflictQueryIterator<T>(
59+
QueryDefinition queryDefinition,
60+
string continuationToken = null,
61+
QueryRequestOptions requestOptions = null)
62+
{
63+
return this.conflicts.GetConflictQueryIterator<T>(queryDefinition, continuationToken, requestOptions);
64+
}
65+
66+
public override Task<ItemResponse<T>> ReadCurrentAsync<T>(
67+
ConflictProperties cosmosConflict,
68+
PartitionKey partitionKey,
69+
CancellationToken cancellationToken = default(CancellationToken))
70+
{
71+
return TaskHelper.RunInlineIfNeededAsync(() => this.conflicts.ReadCurrentAsync<T>(cosmosConflict, partitionKey, cancellationToken));
72+
}
73+
74+
public override T ReadConflictContent<T>(ConflictProperties cosmosConflict)
75+
{
76+
return this.conflicts.ReadConflictContent<T>(cosmosConflict);
77+
}
78+
}
79+
}

‎Microsoft.Azure.Cosmos/src/Resource/Container/ContainerCore.cs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -43,8 +43,8 @@ internal ContainerCore(
4343
id: containerId);
4444

4545
this.Database = database;
46-
this.Conflicts = new ConflictsCore(this.ClientContext, this);
47-
this.Scripts = new ScriptsCore(this, this.ClientContext);
46+
this.Conflicts = new ConflictsInlineCore(new ConflictsCore(this.ClientContext, this));
47+
this.Scripts = new ScriptsInlineCore(new ScriptsCore(this, this.ClientContext));
4848
this.cachedUriSegmentWithoutId = this.GetResourceSegmentUriWithoutId();
4949
this.queryClient = cosmosQueryClient ?? new CosmosQueryClientCore(this.ClientContext, this);
5050
this.BatchExecutor = this.InitializeBatchExecutorForContainer();

‎Microsoft.Azure.Cosmos/src/Resource/Container/ContainerInlineCore.cs‎

Lines changed: 252 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,252 @@
1+
//------------------------------------------------------------
2+
// Copyright (c) Microsoft Corporation. All rights reserved.
3+
//------------------------------------------------------------
4+
5+
namespace Microsoft.Azure.Cosmos
6+
{
7+
using System;
8+
using System.IO;
9+
using System.Linq;
10+
using System.Threading;
11+
using System.Threading.Tasks;
12+
13+
// This class acts as a wrapper for environments that use SynchronizationContext.
14+
internal sealed partial class ContainerInlineCore : Container
15+
{
16+
private readonly ContainerCore container;
17+
18+
public override string Id => this.container.Id;
19+
20+
public override Conflicts Conflicts => this.container.Conflicts;
21+
22+
public override Scripts.Scripts Scripts => this.container.Scripts;
23+
24+
internal CosmosClientContext ClientContext => this.container.ClientContext;
25+
26+
internal Uri LinkUri => this.container.LinkUri;
27+
28+
internal ContainerInlineCore(ContainerCore container)
29+
{
30+
if (container == null)
31+
{
32+
throw new ArgumentNullException(nameof(container));
33+
}
34+
35+
this.container = container;
36+
}
37+
38+
public override Task<ContainerResponse> ReadContainerAsync(
39+
ContainerRequestOptions requestOptions = null,
40+
CancellationToken cancellationToken = default)
41+
{
42+
return TaskHelper.RunInlineIfNeededAsync(() => this.container.ReadContainerAsync(requestOptions, cancellationToken));
43+
}
44+
45+
public override Task<ResponseMessage> ReadContainerStreamAsync(
46+
ContainerRequestOptions requestOptions = null,
47+
CancellationToken cancellationToken = default)
48+
{
49+
return TaskHelper.RunInlineIfNeededAsync(() => this.container.ReadContainerStreamAsync(requestOptions, cancellationToken));
50+
}
51+
52+
public override Task<ContainerResponse> ReplaceContainerAsync(
53+
ContainerProperties containerProperties,
54+
ContainerRequestOptions requestOptions = null,
55+
CancellationToken cancellationToken = default)
56+
{
57+
return TaskHelper.RunInlineIfNeededAsync(() => this.container.ReplaceContainerAsync(containerProperties, requestOptions, cancellationToken));
58+
}
59+
60+
public override Task<ResponseMessage> ReplaceContainerStreamAsync(
61+
ContainerProperties containerProperties,
62+
ContainerRequestOptions requestOptions = null,
63+
CancellationToken cancellationToken = default)
64+
{
65+
return TaskHelper.RunInlineIfNeededAsync(() => this.container.ReplaceContainerStreamAsync(containerProperties, requestOptions, cancellationToken));
66+
}
67+
68+
public override Task<ContainerResponse> DeleteContainerAsync(
69+
ContainerRequestOptions requestOptions = null,
70+
CancellationToken cancellationToken = default)
71+
{
72+
return TaskHelper.RunInlineIfNeededAsync(() => this.container.DeleteContainerAsync(requestOptions, cancellationToken));
73+
}
74+
75+
public override Task<ResponseMessage> DeleteContainerStreamAsync(
76+
ContainerRequestOptions requestOptions = null,
77+
CancellationToken cancellationToken = default)
78+
{
79+
return TaskHelper.RunInlineIfNeededAsync(() => this.container.DeleteContainerStreamAsync(requestOptions, cancellationToken));
80+
}
81+
82+
public override Task<int?> ReadThroughputAsync(CancellationToken cancellationToken = default)
83+
{
84+
return TaskHelper.RunInlineIfNeededAsync(() => this.container.ReadThroughputAsync(cancellationToken));
85+
}
86+
87+
public override Task<ThroughputResponse> ReadThroughputAsync(
88+
RequestOptions requestOptions,
89+
CancellationToken cancellationToken = default)
90+
{
91+
return TaskHelper.RunInlineIfNeededAsync(() => this.container.ReadThroughputAsync(requestOptions, cancellationToken));
92+
}
93+
94+
public override Task<ThroughputResponse> ReplaceThroughputAsync(
95+
int throughput,
96+
RequestOptions requestOptions = null,
97+
CancellationToken cancellationToken = default)
98+
{
99+
return TaskHelper.RunInlineIfNeededAsync(() => this.container.ReplaceThroughputAsync(throughput, requestOptions, cancellationToken));
100+
}
101+
102+
public override Task<ResponseMessage> CreateItemStreamAsync(
103+
Stream streamPayload,
104+
PartitionKey partitionKey,
105+
ItemRequestOptions requestOptions = null,
106+
CancellationToken cancellationToken = default)
107+
{
108+
return TaskHelper.RunInlineIfNeededAsync(() => this.container.CreateItemStreamAsync(streamPayload, partitionKey, requestOptions, cancellationToken));
109+
}
110+
111+
public override Task<ItemResponse<T>> CreateItemAsync<T>(T item,
112+
PartitionKey? partitionKey = null,
113+
ItemRequestOptions requestOptions = null,
114+
CancellationToken cancellationToken = default)
115+
{
116+
return TaskHelper.RunInlineIfNeededAsync(() => this.container.CreateItemAsync<T>(item, partitionKey, requestOptions, cancellationToken));
117+
}
118+
119+
public override Task<ResponseMessage> ReadItemStreamAsync(
120+
string id,
121+
PartitionKey partitionKey,
122+
ItemRequestOptions requestOptions = null,
123+
CancellationToken cancellationToken = default)
124+
{
125+
return TaskHelper.RunInlineIfNeededAsync(() => this.container.ReadItemStreamAsync(id, partitionKey, requestOptions, cancellationToken));
126+
}
127+
128+
public override Task<ItemResponse<T>> ReadItemAsync<T>(
129+
string id,
130+
PartitionKey partitionKey,
131+
ItemRequestOptions requestOptions = null,
132+
CancellationToken cancellationToken = default)
133+
{
134+
return TaskHelper.RunInlineIfNeededAsync(() => this.container.ReadItemAsync<T>(id, partitionKey, requestOptions, cancellationToken));
135+
}
136+
137+
public override Task<ResponseMessage> UpsertItemStreamAsync(
138+
Stream streamPayload,
139+
PartitionKey partitionKey,
140+
ItemRequestOptions requestOptions = null,
141+
CancellationToken cancellationToken = default)
142+
{
143+
return TaskHelper.RunInlineIfNeededAsync(() => this.container.UpsertItemStreamAsync(streamPayload, partitionKey, requestOptions, cancellationToken));
144+
}
145+
146+
public override Task<ItemResponse<T>> UpsertItemAsync<T>(
147+
T item,
148+
PartitionKey? partitionKey = null,
149+
ItemRequestOptions requestOptions = null,
150+
CancellationToken cancellationToken = default)
151+
{
152+
return TaskHelper.RunInlineIfNeededAsync(() => this.container.UpsertItemAsync<T>(item, partitionKey, requestOptions, cancellationToken));
153+
}
154+
155+
public override Task<ResponseMessage> ReplaceItemStreamAsync(
156+
Stream streamPayload,
157+
string id,
158+
PartitionKey partitionKey,
159+
ItemRequestOptions requestOptions = null,
160+
CancellationToken cancellationToken = default)
161+
{
162+
return TaskHelper.RunInlineIfNeededAsync(() => this.container.ReplaceItemStreamAsync(streamPayload, id, partitionKey, requestOptions, cancellationToken));
163+
}
164+
165+
public override Task<ItemResponse<T>> ReplaceItemAsync<T>(
166+
T item,
167+
string id,
168+
PartitionKey? partitionKey = null,
169+
ItemRequestOptions requestOptions = null,
170+
CancellationToken cancellationToken = default)
171+
{
172+
return TaskHelper.RunInlineIfNeededAsync(() => this.container.ReplaceItemAsync<T>(item, id, partitionKey, requestOptions, cancellationToken));
173+
}
174+
175+
public override Task<ResponseMessage> DeleteItemStreamAsync(
176+
string id,
177+
PartitionKey partitionKey,
178+
ItemRequestOptions requestOptions = null,
179+
CancellationToken cancellationToken = default)
180+
{
181+
return TaskHelper.RunInlineIfNeededAsync(() => this.container.DeleteItemStreamAsync(id, partitionKey, requestOptions, cancellationToken));
182+
}
183+
184+
public override Task<ItemResponse<T>> DeleteItemAsync<T>(
185+
string id,
186+
PartitionKey partitionKey,
187+
ItemRequestOptions requestOptions = null,
188+
CancellationToken cancellationToken = default)
189+
{
190+
return TaskHelper.RunInlineIfNeededAsync(() => this.container.DeleteItemAsync<T>(id, partitionKey, requestOptions, cancellationToken));
191+
}
192+
193+
public override FeedIterator GetItemQueryStreamIterator(
194+
QueryDefinition queryDefinition,
195+
string continuationToken = null,
196+
QueryRequestOptions requestOptions = null)
197+
{
198+
return this.container.GetItemQueryStreamIterator(queryDefinition, continuationToken, requestOptions);
199+
}
200+
201+
public override FeedIterator<T> GetItemQueryIterator<T>(
202+
QueryDefinition queryDefinition,
203+
string continuationToken = null,
204+
QueryRequestOptions requestOptions = null)
205+
{
206+
return this.container.GetItemQueryIterator<T>(queryDefinition, continuationToken, requestOptions);
207+
}
208+
209+
public override FeedIterator GetItemQueryStreamIterator(string queryText = null,
210+
string continuationToken = null,
211+
QueryRequestOptions requestOptions = null)
212+
{
213+
return this.container.GetItemQueryStreamIterator(queryText, continuationToken, requestOptions);
214+
}
215+
216+
public override FeedIterator<T> GetItemQueryIterator<T>(
217+
string queryText = null,
218+
string continuationToken = null,
219+
QueryRequestOptions requestOptions = null)
220+
{
221+
return this.container.GetItemQueryIterator<T>(queryText, continuationToken, requestOptions);
222+
}
223+
224+
public override IOrderedQueryable<T> GetItemLinqQueryable<T>(bool allowSynchronousQueryExecution = false,
225+
string continuationToken = null,
226+
QueryRequestOptions requestOptions = null)
227+
{
228+
return this.container.GetItemLinqQueryable<T>(allowSynchronousQueryExecution, continuationToken, requestOptions);
229+
}
230+
231+
public override ChangeFeedProcessorBuilder GetChangeFeedProcessorBuilder<T>(
232+
string processorName,
233+
ChangesHandler<T> onChangesDelegate)
234+
{
235+
return this.container.GetChangeFeedProcessorBuilder<T>(processorName, onChangesDelegate);
236+
}
237+
238+
public override ChangeFeedProcessorBuilder GetChangeFeedEstimatorBuilder(string processorName,
239+
ChangesEstimationHandler estimationDelegate,
240+
TimeSpan? estimationPeriod = null)
241+
{
242+
return this.container.GetChangeFeedEstimatorBuilder(processorName, estimationDelegate, estimationPeriod);
243+
}
244+
245+
public override TransactionalBatch CreateTransactionalBatch(PartitionKey partitionKey)
246+
{
247+
return this.container.CreateTransactionalBatch(partitionKey);
248+
}
249+
250+
public static implicit operator ContainerCore(ContainerInlineCore containerInlineCore) => containerInlineCore.container;
251+
}
252+
}

‎Microsoft.Azure.Cosmos/src/Resource/Database/DatabaseCore.cs‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -262,10 +262,10 @@ public override Container GetContainer(string id)
262262
throw new ArgumentNullException(nameof(id));
263263
}
264264

265-
return new ContainerCore(
265+
return new ContainerInlineCore(new ContainerCore(
266266
this.ClientContext,
267267
this,
268-
id);
268+
id));
269269
}
270270

271271
public override Task<ResponseMessage> CreateContainerStreamAsync(
@@ -315,10 +315,10 @@ public override User GetUser(string id)
315315
throw new ArgumentNullException(nameof(id));
316316
}
317317

318-
return new UserCore(
318+
return new UserInlineCore(new UserCore(
319319
this.ClientContext,
320320
this,
321-
id);
321+
id));
322322
}
323323

324324
public Task<ResponseMessage> CreateUserStreamAsync(

0 commit comments

Comments
 (0)